我對 Spark 還很陌生，我還在學習。我遇到的更困難的概念之一是檢查點以及 Spark 如何使用它從故障中恢復。我正在使用結構化流從 Kafka 進行批量讀取，并將它們作為 Parquet 檔案寫入 S3，如下所示：
dataset
.write()
.mode(SaveMode.Append)
.option("checkpointLocation", checkpointLocation)
.partitionBy("date_hour")
.parquet(getS3PathForTopic(topicName));
檢查點位置是 S3 檔案系統路徑。但是，在作業運行時，我看不到檢查點檔案。在隨后的運行中，我看到以下日志：
21/10/14 12:20:51 INFO ConsumerCoordinator: [Consumer clientId=consumer-spark-kafka-relation-54f0cc87-e437-4582-b998-a33189e90bd7-driver-0-5, groupId=spark-kafka-relation-54f0cc87-e437-4582-b998-a33189e90bd7-driver-0] Found no committed offset for partition topic-1
這表明上次運行沒有檢查點任何偏移量來獲取它們。所以它從最早的偏移量開始消耗。
我怎樣才能讓我的作業獲得新的補償？請注意，這是此處描述的批處理查詢。
我是這樣讀的：
sparkSession
.read()
.format("kafka")
.option("kafka.bootstrap.servers", kafkaProperties.bootstrapServers())
.option("subscribe", topic)
.option("kafka.security.protocol", "SSL")
.option("kafka.ssl.truststore.location", sslConfig.truststoreLocation())
.option("kakfa.ssl.truststore.password", sslConfig.truststorePassword())
.option("kafka.ssl.keystore.location", sslConfig.keystoreLocation())
.option("kafka.ssl.keystore.password", sslConfig.keystorePassword())
.option("kafka.ssl.endpoint.identification.algorithm", "")
.option("failOnDataLoss", "true");
uj5u.com熱心網友回復：
我不知道為什么batchSpark Structured Streaming with Kafka 現在仍然存在。如果您想使用它，那么您必須撰寫自己的 Offset management. 請參閱指南，但它的解釋很糟糕。
我會說Trigger.Once對你來說是一個更好的用例；Offset management由 Spark 提供，因為它不是批處理模式。
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/322629.html