持久化 Spark Streaming 输出答案

【问题标题】：Persisting Spark Streaming output持久化 Spark Streaming 输出
【发布时间】：2022-05-14 12:07:05
【问题描述】：

我正在从一个消息应用程序收集数据，我目前正在使用 Flume，它每天发送大约 5000 万条记录

我想用卡夫卡，使用 Spark Streaming 从 Kafka 消费并将其持久化到 hadoop 并使用 impala 进行查询

我尝试过的每种方法都有问题..

方法 1 - 将 RDD 保存为 parquet，将外部 hive parquet 表指向 parquet 目录

// scala
val ssc =  new StreamingContext(sparkConf, Seconds(bucketsize.toInt))
val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2)
lines.foreachRDD(rdd => {

    // 1 - Create a SchemaRDD object from the rdd and specify the schema
    val SchemaRDD1 = sqlContext.jsonRDD(rdd, schema)

    // 2 - register it as a spark sql table
    SchemaRDD1.registerTempTable("sparktable")

    // 3 - qry sparktable to produce another SchemaRDD object of the data needed 'finalParquet'. and persist this as parquet files
    val finalParquet = sqlContext.sql(sql)
    finalParquet.saveAsParquetFile(dir)

问题是 finalParquet.saveAsParquetFile 输出大量文件，从 Kafka 接收的 Dstream 输出 1 分钟批量大小的 200 多个文件。它输出许多文件的原因是因为计算是分布的，如另一篇文章中所述-how to make saveAsTextFile NOT split output into multiple file?

但是，所提出的解决方案对我来说似乎不是最佳的，例如正如一位用户所说 - 如果您的数据很少，只有一个输出文件才是一个好主意。

方法 2 - 使用 HiveContext。将 RDD 数据直接插入到 hive 表中

# python
sqlContext = HiveContext(sc)
ssc = StreamingContext(sc, int(batch_interval))
kvs = KafkaUtils.createStream(ssc, zkQuorum, group, {topics: 1})
lines = kvs.map(lambda x: x[1]).persist(StorageLevel.MEMORY_AND_DISK_SER)
lines.foreachRDD(sendRecord)

def sendRecord(rdd):

  sql = "INSERT INTO TABLE table select * from beacon_sparktable"

  # 1 - Apply the schema to the RDD creating a data frame 'beaconDF'
  beaconDF = sqlContext.jsonRDD(rdd,schema)

  # 2- Register the DataFrame as a spark sql table.
  beaconDF.registerTempTable("beacon_sparktable")

  # 3 - insert to hive directly from a qry on the spark sql table
  sqlContext.sql(sql);

这很好用，它直接插入到镶木地板表中，但由于处理时间超过了批处理间隔时间，因此批处理存在调度延迟。消费者跟不上正在生产的东西，要处理的批次开始排队。

似乎写入 hive 很慢。我尝试调整批处理间隔大小，运行更多消费者实例。

总结

鉴于存在多个文件的问题以及写入 hive 的潜在延迟，从 Spark Streaming 中持久保存大数据的最佳方法是什么？其他人在做什么？

此处已提出类似问题，但他对与太多文件相对的目录存在问题 How to make Spark Streaming write its output so that Impala can read it?

非常感谢您的帮助

【问题讨论】：

您可以为输出流设置不同的窗口。 val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2).window(Minutes(15)).foreachRDD(rdd =>
这对我来说似乎是一个非常常见的用例，我很惊讶没有人回答它。我想我会建议使用数据库，因为 Spark 本身并不能真正取代它。尝试 Cassandra 或 HBase（HBase 的学习曲线非常陡峭）。

标签： hadoop apache-kafka spark-streaming

【解决方案1】：

在方案#2中，创建文件的数量可以通过每个RDD的分区数量来控制。

看这个例子：

// create a Hive table (assume it's already existing)
sqlContext.sql("CREATE TABLE test (id int, txt string) STORED AS PARQUET")

// create a RDD with 2 records and only 1 partition
val rdd = sc.parallelize(List( List(1, "hello"), List(2, "world") ), 1)

// create a DataFrame from the RDD
val schema = StructType(Seq(
 StructField("id", IntegerType, nullable = false),
 StructField("txt", StringType, nullable = false)
))
val df = sqlContext.createDataFrame(rdd.map( Row(_:_*) ), schema)

// this creates a single file, because the RDD has 1 partition
df.write.mode("append").saveAsTable("test")

现在，我想您可以调整从 Kafka 提取数据的频率，以及每个 RDD 的分区数（默认为 Kafka 主题的分区，您可以通过重新分区来减少）。

我使用的是 CDH 5.5.1 中的 Spark 1.5，我使用 df.write.mode("append").saveAsTable("test") 或您的 SQL 字符串得到了相同的结果。

【讨论】：

很难控制，如前所述，如果数据很大，我认为不应该简单地重新分区到一个文件中。

【解决方案2】：

我认为小文件问题可以得到一些解决。您可能会获得大量基于 kafka 分区的文件。对我来说，我有 12 个分区的 Kafka 主题，我使用 spark.write.mode("append").parquet("/location/on/hdfs") 编写。

现在根据您的要求，您可以添加coalesce(1) 或更多以减少文件数量。另一种选择是增加微批处理持续时间。例如，如果您可以接受写作日延迟 5 分钟，则可以有 300 秒的微批。

对于第二个问题，批次排队只是因为您没有启用背压。首先，您应该验证您可以在单个批次中处理的最大值是多少。一旦您可以绕过该数字，您可以设置 spark.streaming.kafka.maxRatePerPartition 值和 spark.streaming.backpressure.enabled=true 以启用每个微批次的有限记录数。如果仍然无法满足需求，那么唯一的选择是增加主题分区或增加 Spark 应用程序的资源。

【讨论】：