get query

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

@@ -11,14 +11,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
-import com.gmei.up.utils.{ UserInfo }
+import com.gmei.up.utils.{ UserInfo, ES }
 
 class UserInfoDeserializationSchema extends KafkaDeserializationSchema[UserInfo] {
   override def isEndOfStream(t: UserInfo): Boolean = false
 
   override def deserialize(consumerRecord: ConsumerRecord[Array[Byte], Array[Byte]]): UserInfo = {
     val info = new String(consumerRecord.value(), "UTF-8")
-    println(info)
+    // println(info)
     val userInfo: UserInfo = JSON.parseObject(info, classOf[UserInfo])
     userInfo
   }
@@ -53,9 +53,9 @@ object Main {
 
     stream.map { user =>
       println(user.deviceId)
-      // val query = ES.generateDiaryQuery(-1, user.projects, user.secondDemands)
-      // println(query)
-      // ES.test(query)
+      val query = ES.generateDiaryQuery(-1, user.projects.toList, user.secondDemands.toList)
+      println(query)
+      ES.test(query)
       println("########################")
     }