spark structured steaming与kafka集成offset管理方式

Question

spark structured steaming与kafka集成，管理offset的方式有哪几种呢？目的是从故障中恢复流应用程序，比如重启consumer。
我知道的是：1. 利用spark checkpoint 文件，2. 提交offset给kafka。
还有别的方法吗？ 哪种是常用的方案呢？

期待回复，非常感谢！！

Answer 1

Spark Structured Streaming 集成 Kafka 的 offset 管理方式有如下几种：

利用 Spark checkpoint 文件
提交 offset 给 Kafka
常用的方案是使用 Spark checkpoint，因为它可以提供更好的故障恢复性和高可用性，同时也不依赖于 Kafka 的 offset 管理。但是，提交 offset 给 Kafka 也是一个不错的选择，特别是在简单的场景中。在Spark Structured Streaming与Kafka集成时，这两种方法是比较常见的offset管理方法。有时也会通过将offset存储在外部数据存储中（如HDFS，RDBMS）来实现offset管理，以将offset存储在MySQL数据库为例：

# define offsetRanges variable
val offsetRanges = Array(
  // topic, partition, inclusive starting offset, exclusive ending offset
  TopicAndPartition("topicA", 0, 0, 100),
  TopicAndPartition("topicA", 1, 0, 100),
  TopicAndPartition("topicB", 0, 0, 100),
  TopicAndPartition("topicB", 1, 0, 100)
)

# Update offsets in database
offsetRanges.foreach { case (topicAndPartition, offset) =>
  val sql = s"INSERT INTO offsets (topic, partition, offset) VALUES ('${topicAndPartition.topic}', ${topicAndPartition.partition}, $offset) ON DUPLICATE KEY UPDATE offset = $offset"
  conn.createStatement().execute(sql)
}

# Fetch current offsets from database
val currentOffsets = offsetRanges.map { topicAndPartition =>
  val sql = s"SELECT offset FROM offsets WHERE topic = '${topicAndPartition.topic}' AND partition = ${topicAndPartition.partition}"
  val resultSet = conn.createStatement().executeQuery(sql)
  resultSet.next()
  val offset = resultSet.getLong("offset")
  topicAndPartition -> offset
}

但这不太常见。
下面是一个使用 Apache Spark Structured Streaming 集成 Apache Kafka 的简单例子：

val df = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "topic1")
  .load()

val query = df
  .writeStream
  .format("console")
  .outputMode("append")
  .start()

在这个例子中，Spark 从 Kafka 的 topic1 读取数据并写入到控制台。对于 offset 管理，默认情况下 Spark 使用 checkpoint 文件进行管理。可以使用 .option("checkpointLocation", "path/to/checkpoint/dir") 来设置 checkpoint 位置。