apache-spark - 在同一个 Spark 会话中运行多个 Spark Kafka 结构化流查询会增加偏移量但显示 numInputRows 0
问题描述
我有一个 Spark Structured Streaming 消费来自 Kafka 主题的记录,有 2 个分区。
Spark Job: 2 个查询,每个查询来自 2 个单独的分区,从同一个 spark 会话运行。
val df1 = session.readStream.format("kafka")
.option("kafka.bootstrap.servers", kafkaBootstrapServer)
.option("assign", "{\"multi-stream1\" : [0]}")
.option("startingOffsets", latest)
.option("key.deserializer", classOf[StringDeserializer].getName)
.option("value.deserializer", classOf[StringDeserializer].getName)
.option("max.poll.records", 500)
.option("failOnDataLoss", true)
.load()
val query1 = df1
.select(col("key").cast("string"),from_json(col("value").cast("string"), schema, Map.empty[String, String]).as("data"))
.select("key","data.*")
.writeStream.format("parquet").option("path", path).outputMode("append")
.option("checkpointLocation", checkpoint_dir1)
.partitionBy("key")/*.trigger(Trigger.ProcessingTime("5 seconds"))*/
.queryName("query1").start()
val df2 = session.readStream.format("kafka")
.option("kafka.bootstrap.servers", kafkaBootstrapServer)
.option("assign", "{\"multi-stream1\" : [1]}")
.option("startingOffsets", latest)
.option("key.deserializer", classOf[StringDeserializer].getName)
.option("value.deserializer", classOf[StringDeserializer].getName)
.option("max.poll.records", 500)
.option("failOnDataLoss", true)
.load()
val query2 = df2.select(col("key").cast("string"),from_json(col("value").cast("string"), schema, Map.empty[String, String]).as("data"))
.select("key","data.*")
.writeStream.format("parquet").option("path", path).outputMode("append")
.option("checkpointLocation", checkpoint_dir2)
.partitionBy("key")/*.trigger(Trigger.ProcessingTime("5 seconds"))*/
.queryName("query2").start()
session.streams.awaitAnyTermination()
问题:每次在两个分区中推送记录时,两个查询都显示进度,但只有一个正在发出输出。我可以看到那些记录已处理的查询的输出。例如,Kafka Partition 0 - 记录被推送,spark 将处理 query1。Kafka Partition 1 - 当 query1 忙于处理时推送记录,spark 将显示开始偏移量和结束偏移量增加,但查询 2 的 numInputRows = 0。
运行 env : Local PC - 同样的问题。Dataproc 集群 - spark-submit --packages
org.apache.spark:spark-sql-kafka-0-10_2.12:2.4.5 --class org.DifferentPartitionSparkStreaming --master yarn --deploy-mode cluster --num-executors 2 --driver-memory 4g - -executor-cores 4 --executor-memory 4g gs://dpl-ingestion-event/jars/stream_consumer-jar-with-dependencies.jar "{"multiple-streaming" : [0]}" 最新 "10.wxy :9092,10.rst:9092,10.abc:9092" "{"multiple-streaming" : [1]}" - 同样的问题。
检查点和输出路径是 Google Bucket。
日志
20/07/24 19:37:27 INFO MicroBatchExecution: Streaming query made progress: {
"id" : "e7d026f7-bf62-4a86-8697-a95a2fc893bb",
"runId" : "21169889-6e4b-419d-b338-2d4d61999f5b",
"name" : "reconcile",
"timestamp" : "2020-07-24T14:06:55.002Z",
"batchId" : 2,
"numInputRows" : 0,
"inputRowsPerSecond" : 0.0,
"processedRowsPerSecond" : 0.0,
"durationMs" : {
"addBatch" : 3549,
"getBatch" : 0,
"getEndOffset" : 1,
"queryPlanning" : 32,
"setOffsetRange" : 1,
"triggerExecution" : 32618,
"walCommit" : 15821
},
"stateOperators" : [ ],
"sources" : [ {
"description" : "KafkaV2[Assign[multi-stream1-1]]",
"startOffset" : {
"multi-stream1" : {
"1" : 240
}
},
"endOffset" : {
"multi-stream1" : {
"1" : 250
}
},
"numInputRows" : 0,
"inputRowsPerSecond" : 0.0,
"processedRowsPerSecond" : 0.0
} ],
"sink" : {
"description" : "FileSink[gs://dpl-ingestion-event/demo/test/single-partition/data]"
}
解决方案
我能够解决这个问题。根本原因是两个查询都试图写入相同的基本路径。因此,_spark_meta 信息存在重叠。Spark Structured Streaming 维护检查点以及 _spark_metadata 文件以跟踪正在处理的批处理。
源 Spark 文档:
为了正确处理部分故障,同时保持一次语义,每个批次的文件都被写入一个唯一的目录,然后自动附加到元数据日志中。当基于 parquet 的 DataSource 被初始化以供读取时,我们首先检查此日志目录并在存在时使用它而不是文件列表。
因此,现在每个查询都应该有一个单独的路径。与检查点不同,没有配置 _spark_matadata 位置的选项。
推荐阅读
- html - 垂直排列相邻的 DIV,无需强制
- laravel - 自定义 webpack 配置的问题
- pandas - 在 Dataframes 上将 2 列绘制为 2 行,将 1 列绘制为 x 轴
- java - 我收到提到的代码超时错误。需要帮助来优化嵌套循环
- c# - 迁移到 .NET Core 3 后出现“项目 'Web' 必须为配置提供值”错误
- r - 如何不从原始 data.table 中删除列?
- python - 在 pandas 数据框上为产品创建每日价格变化
- rust - 无锁堆栈,将 is_empty() 中的 Acquire 替换为 Relaxed
- c++ - dll中存在并在客户端中调用的具体模板主体
- database - 获取日期范围内每天仅存在一次的唯一记录