apache-spark - 将单个 DStream 拆分为多个 Hive 表

问题描述

我正在研究 Kafka Spark 流媒体项目。Spark 流从 Kafka 获取数据。数据为 json 格式。样本输入

{ “table”：“tableA”，“Product_ID”：“AGSVGF.upf”，“file_timestamp”：“2018-07-26T18:58:08.4485558Z000000000000000”，“hdfs_file_name”：“null_1532631600050”，“Date_Time”：“2018” -07-26T13：45：01.0000000Z”，“用户名”：“UBAHTSD”}

{“表”：“tableB”，“Test_ID”：“FAGS.upf”，“时间戳”：“2018-07-26T18：58：08.4485558Z000000000000000”，“名称”：“flink”，“时间”：“2018 -07-26T13：45：01.0000000Z”，“Id”：“UBAHTGADSGSCVDGHASD”}

一个 JSON 字符串就是一条消息。JSON 字符串有 15 种类型，使用表列进行区分。现在我想在 Apache Hive 中保存这 15 种不同的 JSON。所以我创建了一个 dstream 并在表列的基础上过滤了 rdd 并保存到 Hive 中。代码工作正常。但是有些时候它会花很多时间然后激发批处理。我已经使用spark.streaming.kafka.maxRatePerPartition=10. 我已将 rdd 重新分区为 9 个分区，但在 Spark UI 上，它显示未知阶段。

这是我的代码。

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)
              }

    }

}

我如何优化代码？

谢谢你。

标签： apache-sparkoptimizationapache-kafkaspark-streaming

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

这样，您可以更轻松地监控哪个 Spark 作业的性能不如其他作业，并且一个表的数据倾斜不会导致其他表出现问题。

目前尚不清楚 Kafka 消息键是什么，但如果以表为键生成，那么 Spark 可以与 kafka 分区一起扩展，并且您可以保证每个表的所有消息都按顺序排列。

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

apache-spark - 将单个 DStream 拆分为多个 Hive 表

问题描述

解决方案

推荐阅读