如何管理Spark结构化流中的偏移？(_spark_metadata的问题）

Question

背景：我写了一个简单的spark结构化蒸app，把数据从Kafka搬到S3。我发现，为了支持一次准确的保证，spark创建了_spark_metadata文件夹，但该文件夹最终变得太大，当流应用程序运行很长时间时，元数据文件夹变得太大，以至于我们开始出现OOM错误。我想摆脱Spark结构化流的元数据和检查点文件夹，自己管理偏移量。

我们如何管理Spark Streaming中的偏移量：我使用了val offsetRanges=rdd.asInstanceOf[HasOffsetRanges].offsetRanges来获取Spark结构化流中的偏移量。但是想知道如何获取偏移量和其他元数据来使用Spark结构化流管理自己的检查点。您有没有实现检查点的示例程序？

我们是如何在Spark结构化流中管理偏移的？？看看这个JIRA https://issues-test.apache.org/JIRA/browse/spark-18258。似乎没有提供偏移。我们该怎么走呢？

代码：

1. Reading records from Kafka topic
  Dataset<Row> inputDf = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2") \
  .option("subscribe", "topic1") \
  .option("startingOffsets", "earliest") \
  .load()
2. Use from_json API from Spark to extract your data for further transformation in a dataset.
   Dataset<Row> dataDf = inputDf.select(from_json(col("value").cast("string"), EVENT_SCHEMA).alias("event"))
       ....withColumn("oem_id", col("metadata.oem_id"));
3. Construct a temp table of above dataset using SQLContext
   SQLContext sqlContext = new SQLContext(sparkSession);
   dataDf.createOrReplaceTempView("event");
4. Flatten events since Parquet does not support hierarchical data.
5. Store output in parquet format on S3
   StreamingQuery query = flatDf.writeStream().format("parquet")

Dataset dataDf=inputdf.select(from_json(col（“value”）.cast（“string”）,EVENT_SCHEMA).alias（“event”））.select（“event.metadata”,“event.data”,“event.connection”,“event.registration_event”,“event.version_event”）；SQLContext SQLContext=新建SQLContext（sparkSession）；DataDf.CreateOrReplaceTempView（“事件”）；Dataset flatDf=sqlContext.sql（“Select”+“date,time,id”+flattenSchema(EVENT_SCHEMA,“Event”）+“from Event”）；StreamingQuery query=flatDf.WriteStream（）.OutputMode（“append”）.Option（“Compression”,“Snappy”）.Format（“Parquet”）.Option（“checkpointLocation”,checkpointLocation“.Option（”Path“,outputPath）.PartitionBy（”Date“,”Time“,”ID“）.Trigger（Trigger.ProcessingTime（triggerProcessingTime））.Start（）；Query.AwaitTermination（）；

Answer 1

对于非批处理Spark结构化流KAFKA集成：

报价：

结构化流忽略Apache Kafka中提交的偏移量。

优秀参考：https://www.waitingforcode.com/apache-spark-structure-streaming/apache-spark-structure-streaming-apache-kafka-offsets-management/read

对于批处理，情况不同，您需要自己管理并存储偏移。

更新基于评论，我建议问题略有不同，建议您看看Spark结构化流检查点清理。除了您更新的评论和没有错误的事实之外，我建议您在Spark Structured Streaming的元数据https://www.waitingforcode.com/apache-spark-structured-streaming/checkpoint-storage-structured-streaming/read上确认这一点。看代码，与我的风格不同，但看不到任何明显的错误。