get exception

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

@@ -53,69 +53,69 @@ object Main extends LazyLogging {
 
       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(" "))
-
-          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
-            )
-
-          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")
-
-          logger.info(s"${user.eventCn} ${deviceId}")
-
-          deviceId
-        } catch {
-          case e: Throwable =>
-            e.printStackTrace()
-            ""
-        }
-      }
-
-      // stream.print
+      // 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(" "))
+
+      //     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
+      //       )
+
+      //     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")
+
+      //     logger.info(s"${user.eventCn} ${deviceId}")
+
+      //     deviceId
+      //   } catch {
+      //     case e: Throwable =>
+      //       e.printStackTrace()
+      //       ""
+      //   }
+      // }
+
+      stream.print
       env.execute("flink streaming user portrait")
 
     } catch {