add try except

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

@@ -28,77 +28,94 @@ class UserInfoDeserializationSchema extends KafkaDeserializationSchema[UserInfo]
 }
 
 object Main {
-  def main(args: Array[String]): Unit = {
+  def main(args: Array[String]): Unit =
     // println("###############")
     // val query = ES.generateServiceDiaryRequest(List("光子嫩肤", "水光针"), List("丰胸", "胸部塑身"), -1)
     // println("###############")
+    try {
+      val env = StreamExecutionEnvironment.getExecutionEnvironment
 
-    val env = StreamExecutionEnvironment.getExecutionEnvironment
+      // TODO read from config
+      val kafkaConsumerProperties = new Properties
+      kafkaConsumerProperties.setProperty("group.id", "user_portrait_flink_streaming")
+      kafkaConsumerProperties.setProperty(
+        "bootstrap.servers",
+        "172.16.44.25:9092,172.16.44.31:9092,172.16.44.45:9092"
+      )
 
-    // TODO read from config
-    val kafkaConsumerProperties = new Properties
-    kafkaConsumerProperties.setProperty("group.id", "user_portrait_flink_streaming")
-    kafkaConsumerProperties.setProperty(
-      "bootstrap.servers",
-      "172.16.44.25:9092,172.16.44.31:9092,172.16.44.45:9092"
-    )
+      val kafkaConsumer = new FlinkKafkaConsumer[UserInfo](
+        "gm-portrait-update-device",
+        new UserInfoDeserializationSchema,
+        kafkaConsumerProperties
+      )
 
-    val kafkaConsumer = new FlinkKafkaConsumer[UserInfo](
-      "gm-portrait-update-device",
-      new UserInfoDeserializationSchema,
-      kafkaConsumerProperties
-    )
+      val stream: DataStream[UserInfo] = env.addSource(kafkaConsumer)
 
-    val stream: DataStream[UserInfo] = env.addSource(kafkaConsumer)
+      stream.map { user =>
+        try {
+          val deviceId = user.deviceId
+          val projects = user.projects.toList
+          val secondDemands = user.secondDemands.toList
+          val secondPositions = user.secondPositions.toList
+          val secondSolutions = user.secondSolutions.toList
+          val cityId = user.cityId
+          println(deviceId)
+          // println(projects.mkString(" "))
 
-    stream.map { user =>
-      val deviceId = user.deviceId
-      val projects = user.projects.toList
-      val secondDemands = user.secondDemands.toList
-      val secondPositions = user.secondPositions.toList
-      val secondSolutions = user.secondSolutions.toList
-      val cityId = user.cityId
-      println(deviceId)
-      // println(projects.mkString(" "))
+          val diaryRead = Redis.getRead(deviceId, "diary")
+          val tractateRead = Redis.getRead(deviceId, "tractate")
+          val answerRead = Redis.getRead(deviceId, "answer")
 
-      val diaryRead = Redis.getRead(deviceId, "diary")
-      val tractateRead = Redis.getRead(deviceId, "tractate")
-      val answerRead = Redis.getRead(deviceId, "answer")
+          val diaryReq =
+            ES.generateDiaryRequest(
+              projects,
+              secondDemands,
+              secondPositions,
+              secondSolutions,
+              diaryRead,
+              cityId,
+              300
+            )
+          val tractateReq =
+            ES.generateTractateRequest(
+              projects,
+              secondDemands,
+              secondPositions,
+              secondSolutions,
+              tractateRead,
+              300
+            )
+          val answerReq =
+            ES.generateAnswerRequest(projects, secondDemands, secondPositions, secondSolutions, answerRead, 100)
+          val serviceDiaryReq =
+            ES.generateServiceDiaryRequest(
+              projects,
+              secondDemands,
+              secondPositions,
+              secondSolutions,
+              diaryRead,
+              cityId,
+              100
+            )
 
-      val diaryReq =
-        ES.generateDiaryRequest(
-          projects,
-          secondDemands,
-          secondPositions,
-          secondSolutions,
-          diaryRead,
-          cityId,
-          300
-        )
-      val tractateReq =
-        ES.generateTractateRequest(projects, secondDemands, secondPositions, secondSolutions, tractateRead, 300)
-      val answerReq =
-        ES.generateAnswerRequest(projects, secondDemands, secondPositions, secondSolutions, answerRead, 100)
-      val serviceDiaryReq =
-        ES.generateServiceDiaryRequest(
-          projects,
-          secondDemands,
-          secondPositions,
-          secondSolutions,
-          diaryRead,
-          cityId,
-          100
-        )
+          Redis.save(ES.request(diaryReq), deviceId, "diary")
+          Redis.save(ES.request(tractateReq), deviceId, "tractate")
+          Redis.save(ES.request(answerReq), deviceId, "answer")
+          Redis.save(ES.request(serviceDiaryReq), deviceId, "service_diary")
 
-      Redis.save(ES.request(diaryReq), deviceId, "diary")
-      Redis.save(ES.request(tractateReq), deviceId, "tractate")
-      Redis.save(ES.request(answerReq), deviceId, "answer")
-      Redis.save(ES.request(serviceDiaryReq), deviceId, "service_diary")
+          deviceId
+        } catch {
+          case e: Throwable =>
+            e.printStackTrace()
+            ""
+        }
+      }
 
-      deviceId
-    }
+      // stream.print
+      env.execute("flink streaming user portrait")
 
-    // stream.print
-    env.execute("flink streaming user portrait")
-  }
+    } catch {
+      case e: Throwable =>
+        e.printStackTrace()
+    }
 }

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

@@ -65,7 +65,7 @@ object ES {
 
   def request(req: SearchRequest): Future[IndexedSeq[Content]] = {
     val client: ElasticClient = ESClient.create("172.16.52.33", 9200, "elastic", "gengmei!@#")
-    println(req.show)
+    // println(req.show)
     client.execute(req).map(resp => resp.result.to[Content])
     // client.close()
   }