将单个DStream拆分为多个Hive表

Question

我的工作是卡夫卡火花流媒体项目。火花流从卡夫卡获得数据。数据采用json格式。样本输入

{“tableA”："tableA"，"Product_ID"："AGSVGF.upf"，"file_timestamp"：file_timestamp "hdfs_file_name"："null_1532631600050"，"Date_Time"："2018-07-26T13:45:01.0000000Z"，"User_Name"："UBAHTSD“} {“表”："tableB"，"Test_ID"："FAGS.upf"，“时间戳”：“名称”："flink"，“时间”："2018-07-26T13:45:01.0000000Z"，"Id"："UBAHTGADSGSCVDGHASD“}

一个JSON字符串是一条消息。有15种类型的JSON字符串可以区分使用表列。现在，我想在Apache中保存这15个不同的JSON。因此，我创建了一个dstream，并在表列的基础上过滤了rdd并保存到Hive中。代码工作正常。但有些时间-很多时间我使用spark.streaming.kafka.maxRatePerPartition=10控制输入。我已经将rdd重新划分为9个分区，但是在Spark上，它显示了未知的阶段。

​

这是我的密码。

val dStream = dataStream.transform(rdd => rdd.repartition(9)).map(_._2)
dStream.foreachRDD { rdd =>
    if (!rdd.isEmpty()) {
      val sparkContext = rdd.sparkContext
      rdd.persist(StorageLevel.MEMORY_AND_DISK)
      val hiveContext = getInstance(sparkContext)
          val tableA = rdd.filter(_.contains("tableA"))
          if (!tableA.isEmpty()) {
            HiveUtil.tableA(hiveContext.read.json(tableA))
            tableA.unpersist(true)
          }

          val tableB = rdd.filter(_.contains("tableB"))
          if (!tableB.isEmpty()) {
            HiveUtil.tableB(hiveContext.read.json(tableB))
            tableB.unpersist(true)
          }
          .....
          .... upto 15 tables
          ....

            val tableK = rdd.filter(_.contains("tableK"))
              if (!tableB.isEmpty()) {
                HiveUtil.tableB(hiveContext.read.json(tableK))
                tableB.unpersist(true)
              }

    }

}

我如何优化代码？

谢谢。

Answer 1

纯粹从管理的角度出发，我建议您将作业参数化以接受表名，然后运行15个单独的Spark应用程序。还要确保每一个应用程序的kafka消费者组是不同的。

通过这种方式，您可以更容易地监视哪些星火作业执行得不如其他作业，并且将数据倾斜到一个表不会导致与其他表的问题。

还不清楚Kafka消息键是什么，但是如果生成的是表作为键，那么Spark可以与kafka分区一起伸缩，并且保证每个表的所有消息都是有序的。

总的来说，我实际上会使用Kafka Connect或Streamset来编写HDFS/Hive，而不必编写代码或调整Spark设置。