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

Question

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

运行环境：本地PC-同样的问题。Dataproc集群-spark-submit--包

org.apache.spark:spark-sql-kafka-0-102.12:2.4.5--类org.differentPartitionSparkStreaming--主纱--部署模式集群--num-executors 2--驱动程序--内存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”{“Multiple-Streaming”:[1]}“--同样的问题。

检查点和输出路径是谷歌桶。

原木

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]"
  }

Answer 1

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

源火花单据：

为了正确处理部分故障，同时保持仅一次语义，每个批处理的文件都被写到一个唯一的目录中，然后原子地追加到元数据日志中。当初始化一个基于拼花的数据源以便读取时，我们首先检查这个日志目录并使用它而不是文件列表。