我正在尝试从Kafka结构化流。我计划在hdfs中存储检查点。我读了一篇cloudera博客，建议不要在hdfs中存储spark流的检查点。对于结构流检查点也是同样的问题。https://blog.cloudera.com/blog/2017/06/offset-management-for-apache-kafka-with-apache-spark-streaming/.
在结构化流媒体中，如果我的spark程序关闭了一段时间，如何从checkpoint目录获取最新的偏移量，并在该偏移量之后加载数据。我将检查点存储在一个目录中，如下所示。

df.writeStream\
        .format("text")\
        .option("path", '\files') \
        .option("checkpointLocation", 'checkpoints\chkpt') \
        .start()

更新：
这是我的结构化流媒体程序读取Kafka消息，解压并写入hdfs。

df = spark \
        .readStream \
        .format("kafka") \
        .option("kafka.bootstrap.servers", KafkaServer) \
        .option("subscribe", KafkaTopics) \
        .option("failOnDataLoss", "false")\
         .load()
Transaction_DF = df.selectExpr("CAST(value AS STRING)")
Transaction_DF.printSchema()

decomp = Transaction_DF.select(zip_extract("value").alias("decompress"))

# zip_extract is a UDF to decompress the stream

query = decomp.writeStream\
    .format("text")\
    .option("path", \Data_directory_inHDFS) \
    .option("checkpointLocation", \pathinDHFS\) \
    .start()

query.awaitTermination()