add printer

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

@@ -11,27 +11,19 @@ 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
+import com.gmei.up.utils.UserInfo
 
-case class KafkaUserInfo(
-    device_id: String,
-    action: String,
-    log_time: Double,
-    event_cn: String,
-    second_demands: Array[String],
-    projects: Array[String]
-)
+class UserInfoDeserializationSchema extends KafkaDeserializationSchema[UserInfo] {
+  override def isEndOfStream(t: UserInfo): Boolean = false
 
-class UserInfoDeserializationSchema extends KafkaDeserializationSchema[KafkaUserInfo] {
-  override def isEndOfStream(t: KafkaUserInfo): Boolean = false
-
-  override def deserialize(consumerRecord: ConsumerRecord[Array[Byte], Array[Byte]]): KafkaUserInfo = {
+  override def deserialize(consumerRecord: ConsumerRecord[Array[Byte], Array[Byte]]): UserInfo = {
     val info = new String(consumerRecord.value(), "UTF-8")
-    val userInfo: KafkaUserInfo = JSON.parseObject(info, classOf[KafkaUserInfo])
+    val userInfo: UserInfo = JSON.parseObject(info, classOf[UserInfo])
     userInfo
   }
 
-  override def getProducedType: TypeInformation[KafkaUserInfo] =
-    Types.GENERIC(classOf[KafkaUserInfo])
+  override def getProducedType: TypeInformation[UserInfo] =
+    Types.GENERIC(classOf[UserInfo])
 }
 
 object Main {
@@ -46,21 +38,21 @@ object Main {
       "172.16.44.25:9092,172.16.44.31:9092,172.16.44.45:9092"
     )
 
-    val kafkaConsumer = new FlinkKafkaConsumer[KafkaUserInfo](
+    val kafkaConsumer = new FlinkKafkaConsumer[UserInfo](
       "gm-portrait-update-device",
       new UserInfoDeserializationSchema,
       kafkaConsumerProperties
     )
 
-    val stream: DataStream[KafkaUserInfo] = env.addSource(kafkaConsumer)
+    val stream: DataStream[UserInfo] = env.addSource(kafkaConsumer)
 
     stream.map { user =>
-      println(user.device_id)
+      println(user.deviceId)
       user.second_demands.foreach(println)
       println("########################")
     }
 
-    stream.print
+    // stream.print
     env.execute("flink streaming user portrait")
   }
 }

src/main/scala/com.gmei.up/utils/User.scala

 package com.gmei.up.utils
+
+case class UserInfo(
+    deviceId: String,
+    action: String,
+    log_time: Double,
+    event_cn: String,
+    second_demands: Array[String],
+    projects: Array[String]
+)
+
+object UserInfo