增加对京东数据的支持

src/main/scala/entry/Entry.scala

@@ -3,10 +3,10 @@ package entry
 import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
 import tools.kafka.KafkaConsumer
 import org.apache.flink.api.scala._
-import tools.dataformat.TB
-import tools.es.SaveData
+import tools.dataformat.Data
 import tools.json.Jackson
-
+import tools.properties.PropertiesTools
+import java.util.Properties
 import scala.collection.mutable
 
 /**
@@ -19,25 +19,53 @@ import scala.collection.mutable
  */
 object Entry {
   def main(args: Array[String]): Unit = {
+
+    val properties: Properties = PropertiesTools.getProperties
+
+    //     消费京东和淘宝的数据
+    val tb_topic: String = properties.getProperty("kafka.tb.topic")
+    val tb_group: String = properties.getProperty("kafka.tb.consumer.group")
+    val jd_topic: String = properties.getProperty("kafka.jd.topic")
+    val jd_group: String = properties.getProperty("kafka.jd.consumer.group")
+
     //   配置环境
     val environment = StreamExecutionEnvironment.getExecutionEnvironment
-    environment.enableCheckpointing(1000L)
+    //    .enableCheckpointing(1000L)
+
     //   kafka源
-    val consumer = KafkaConsumer.getConsumer
-    consumer.setStartFromEarliest()
+    val tb_consumer = KafkaConsumer.getConsumer(tb_topic, tb_group)
+    val jd_consumer = KafkaConsumer.getConsumer(jd_topic, jd_group)
+
+    tb_consumer.setStartFromEarliest()
+    jd_consumer.setStartFromEarliest()
 
-    //    添加源
-    environment.addSource(consumer)
+    // 清洗淘宝数据
+    val tbSource = environment
+      .addSource(tb_consumer)
+      .map(tb => {
+        val res = mutable.Map[String, String]()
+        Jackson.parseAllKey(tb, res)
+      })
+      .map(x => Data.formatTBData(x))
+
+    // 清洗京东数据
+    val jdSource = environment
+      .addSource(jd_consumer)
       .map(x => {
         val res = mutable.Map[String, String]()
-        val map = Jackson.parseAllKey(x, res)
-        map
+        Jackson.parseAllKey(x, res)
       })
-      .map(x => TB.formatTBData(x))
-      .map(x => SaveData.saveToES(x))
-    //      .executeAndCollect()
-    //      .foreach(println)
+      .map(x => Data.formatJDData(x))
+    // 合并流
+    val ds = tbSource.union(jdSource)
+
+
+    //存入ES
+    ds
+      //      .map(x => SaveData.saveToES(x))
+      .executeAndCollect()
+      .foreach(println)
 
-    environment.execute("tb")
+    //        environment.execute("jingdong")
   }
 }

src/main/scala/tools/kafka/KafkaConsumer.scala

@@ -15,15 +15,13 @@ import tools.properties.PropertiesTools
  */
 object KafkaConsumer {
   private[this] val properties: Properties = PropertiesTools.getProperties
-  private[this] val topic: String = properties.getProperty("kafka.topic")
   private[this] val bootstrapServer: String = properties.getProperty("kafka.cluster")
-  private[this] val group: String = properties.getProperty("kafka.consumer.group")
   private[this] val keyDeserializer: String = properties.getProperty("kafka.consumer.key.deserializer")
   private[this] val valueDeserializer: String = properties.getProperty("kafka.consumer.value.deserializer")
   private[this] val commitInterval: String = properties.getProperty("kafka.consumer.offset.commit.auto.interval")
   private[this] val AutoCommit: String = properties.getProperty("kafka.consumer.offset.commit.auto")
 
-  def getConsumer: FlinkKafkaConsumer[String] = {
+  def getConsumer(topic: String, group: String): FlinkKafkaConsumer[String] = {
     val property = new Properties()
     // 对接kafka
     property.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer)