Spark Streaming Direct Kafka API, OffsetRanges : 如何先处理 运行
Spark Streaming Direct Kafka API, OffsetRanges : How to handle first run
我的 spark-streaming 应用程序在没有 ZooKeeper 帮助的情况下使用直接流方法从 Kafka 读取数据。我想处理故障，以便在我的应用程序中遵循 Exactly-once 语义。我正在关注 this 以供参考。一切看起来都很完美，除了 :
val stream: InputDStream[(String,Long)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, Long)](
ssc, kafkaParams, fromOffsets,
// we're just going to count messages per topic, don't care about the contents, so convert each message to (topic, 1)
(mmd: MessageAndMetadata[String, String]) => (mmd.topic, 1L))
在应用程序的第一个 运行 中，由于不会读取偏移量，因此要为 fromOffsets Map 参数传递什么值？我肯定漏掉了什么。
感谢并感谢任何帮助！
第一个偏移量不一定是 0L，具体取决于主题存在的时间。
我个人只是将适当的偏移量分别预先插入到数据库中。然后 spark 作业在启动时从数据库中读取偏移量。
spark Kafka 集成中的文件 kafkacluster.scala 具有可以更轻松地查询 Kafka 以获取最早可用偏移量的方法。该文件是私有的，但已在最近的 spark 代码中创建 public。
我的 spark-streaming 应用程序在没有 ZooKeeper 帮助的情况下使用直接流方法从 Kafka 读取数据。我想处理故障，以便在我的应用程序中遵循 Exactly-once 语义。我正在关注 this 以供参考。一切看起来都很完美，除了 :
val stream: InputDStream[(String,Long)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, Long)](
ssc, kafkaParams, fromOffsets,
// we're just going to count messages per topic, don't care about the contents, so convert each message to (topic, 1)
(mmd: MessageAndMetadata[String, String]) => (mmd.topic, 1L))
在应用程序的第一个 运行 中，由于不会读取偏移量，因此要为 fromOffsets Map 参数传递什么值？我肯定漏掉了什么。
感谢并感谢任何帮助！
第一个偏移量不一定是 0L，具体取决于主题存在的时间。
我个人只是将适当的偏移量分别预先插入到数据库中。然后 spark 作业在启动时从数据库中读取偏移量。
spark Kafka 集成中的文件 kafkacluster.scala 具有可以更轻松地查询 Kafka 以获取最早可用偏移量的方法。该文件是私有的，但已在最近的 spark 代码中创建 public。