在同一个 Spark 会话中运行多个 Spark Kafka 结构化流查询会增加偏移量但显示 numInputRows 0

Posted

技术标签:

【中文标题】在同一个 Spark 会话中运行多个 Spark Kafka 结构化流查询会增加偏移量但显示 numInputRows 0【英文标题】:Running multiple Spark Kafka Structured Streaming queries in same spark session increasing the offset but showing numInputRows 0 【发布时间】:2020-07-24 14:18:59 【问题描述】:

我有一个 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。

运行环境:本地 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 集群 --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.w.x.y:9092,10.r.s.t:9092,10.a.b.c:9092”““多流”: [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]"
  

【问题讨论】:

【参考方案1】:

我能够解决问题。 根本原因是两个查询都试图写入相同的基本路径。因此,_spark_meta 信息存在重叠。 Spark Structured Streaming 维护检查点以及 _spark_metadata 文件以跟踪正在处理的批处理。

源 Spark 文档:

为了正确处理部分故障同时保持一次 语义上,每个批次的文件都写到一个唯一的 目录,然后自动附加到元数据日志。当一个 parquet based DataSource 被初始化为读取,我们首先检查 对于此日志目录,并在以下情况下使用它而不是文件列表 现在。

因此,现在应该为每个查询指定一个单独的路径。与检查点不同,没有配置 _spark_matadata 位置的选项。

【讨论】:

“尝试写入相同的基本路径”是什么意思?你能用更新的代码更新你的答案吗?

以上是关于在同一个 Spark 会话中运行多个 Spark Kafka 结构化流查询会增加偏移量但显示 numInputRows 0的主要内容,如果未能解决你的问题,请参考以下文章

Spark 在具有多个应用程序的工作人员上运行了多少 JVM

Spark 管道中的 UDF

spark.driver.extraClassPath 多个罐子

可以在 Databricks 集群运行后设置/修改的 Spark 会话属性

Spark:从具有不同内存/核心配置的单个JVM作业同时启动

如何一次运行多个 Spark 2.0 实例(在多个 Jupyter Notebook 中)?