get service id

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

@@ -45,8 +45,9 @@ object Main extends LazyLogging {
         "172.16.44.25:9092,172.16.44.31:9092,172.16.44.45:9092"
       )
 
+      // TODO remove test
       val kafkaConsumer = new FlinkKafkaConsumer[UserInfo](
-        "gm-portrait-update-device",
+        "gm-portrait-update-device-test",
         new UserInfoDeserializationSchema,
         kafkaConsumerProperties
       )

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

 package com.gmei.up.utils
 
-case class Content(id: Long, index: String, projects: String)
+case class Content(id: Long, index: String, serviceId: Some[Long], projects: String)

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

@@ -57,7 +57,10 @@ object ES {
   implicit object ContentHitReader extends HitReader[Content] {
     override def read(hit: Hit): Try[Content] = {
       val source = hit.sourceAsMap
-      Try(Content(source("id").toString.toLong, hit.index, source("tags_v3").toString))
+      val serviceId = source.get("service.id").asInstanceOf[Some[Long]]
+      Try(
+        Content(source("id").toString.toLong, hit.index, serviceId, source("tags_v3").toString)
+      )
     }
   }
 
@@ -66,6 +69,11 @@ object ES {
   def request(req: SearchRequest): Either[Throwable, Future[IndexedSeq[Content]]] = {
     val client: ElasticClient = ESClient.create("172.16.52.33", 9200, "elastic", "gengmei!@#")
     // println(req.show)
+    client.execute(req).map { resp =>
+      val a = resp.result.to[Content]
+      a.foreach(x => println(x.id, x.index, x.serviceId))
+    }
+
     try {
       Right(client.execute(req).map(resp => resp.result.to[Content]))
     } catch {