add printer

build.sbt

@@ -24,7 +24,7 @@ lazy val root = (project in file("."))
   .settings(
     name := "streamingUserPortrait",
 
-    // libraryDependencies += "com.alibaba" % "fastjson" % "1.2.75",
+    libraryDependencies += "com.alibaba" % "fastjson" % "1.2.75",
     libraryDependencies += es,
     libraryDependencies += esCore,
 

src/main/scala/com.gmei.up/Main.scala

@@ -10,13 +10,14 @@ import org.apache.flink.streaming.connectors.kafka.{
 }
 import org.apache.flink.api.common.typeinfo.{ TypeHint, TypeInformation, Types }
 import org.apache.kafka.clients.consumer.ConsumerRecord
+import com.alibaba.fastjson.JSON
 
 case class KafkaUserInfo(
-    deviceId: String,
+    device_id: String,
     action: String,
-    logTime: Double,
-    eventCn: String,
-    secondDemands: Array[String],
+    log_time: Double,
+    event_cn: String,
+    second_demands: Array[String],
     projects: Array[String]
 )
 
@@ -24,10 +25,11 @@ class UserInfoDeserializationSchema extends KafkaDeserializationSchema[KafkaUser
   override def isEndOfStream(t: KafkaUserInfo): Boolean = false
 
   override def deserialize(consumerRecord: ConsumerRecord[Array[Byte], Array[Byte]]): KafkaUserInfo = {
-    // KafkaUserInfo(consumerRecord.topic(), new String(consumerRecord.value(), "UTF-8"))
-    println(consumerRecord)
-    val value = new String(consumerRecord.value(), "UTF-8")
-    println(value)
+    val info = new String(consumerRecord.value(), "UTF-8")
+    val userInfo: KafkaUserInfo = JSON.parseObject(info, classOf[KafkaUserInfo])
+    println(userInfo.device_id)
+    userInfo.projects.foreach(println)
+
     println("##############")
     KafkaUserInfo(
       "abc",