kafka在Spark

Question

Kafka

enable.auto.commit

设置为

false

，Spark版本为2.4

如果使用最新偏移量，我们是否需要手动查找最后一个偏移详细信息并在Spark应用程序中的.CREATEDECTSTREAM（）中提及它？还是它会自动采用最新偏移？无论如何，我们都需要手动找到最后的偏移细节。
SparkSession.readstrem.format(kafka)....
和
```
KafkaUtils.createDirectStream()
```
？
当使用最早的偏移选项时，会自动考虑偏移吗？

Answer 1

QUE1：

enable.auto.commit

是一个相关的参数，如果设置为false，则需要您手动提交（读取更新）您对检查点目录的偏移。 If your application restarts it will look into the checkpoint directory and start reading from last committed offset + 1. same is mentioned here

。无需将偏移量指定为火花应用程序的一部分。您需要的只是检查点目录。此外，请记住，每个分区都在每个消费者的主题中维护偏移，因此在Spark上期望开发人员/用户提供此功能是不好的。
```
Ques2：Spark.ReadStream是一种通用方法，可以从流源，例如TCP套接字，Kafka主题等读取数据，而Kafkautils是一个专门的类，用于与Kafka集成Spark，因此我认为如果您使用KAFKA主题是更优化的作为来源。我通常自己使用Kafkautils，通过我没有进行任何性能基准。如果我没有错，那么Kafkautils也可以用来订阅一个以上的主题，而ReadStream则不能。
```
QUE3：最早的偏移意味着您的消费者将开始从可用的最古老的记录开始阅读，例如，如果您的主题是新的（没有清理已经发生），或者未为主题配置清理，则将从偏移0开始。已配置并删除了所有记录，所有记录已被删除，将从offset 2001中读取记录，而该主题可能具有记录直到offset 10000（这是假设只有一个分区，在主题中，多个分区将偏移值不同）。请参阅此处有关批次查询的部分。https：//spark.apache.org/docs/2.2.0/-structrud-streaming-kafka-integration.html有关更多详细信息。
，您可以找到大多数答案。关于kafka连接器的关于startingOffsets

Answer 2

如果您有偏移，如果它们可用，它将始终接收偏移，否则会要求kafka提供earliest或

latest

偏移。对于两种类型的流，直接和结构化流都应考虑偏移应该是正确的。

我看到您提到了

enable.auto.commit

选项，我只想确保您知道您上面提供的同一文档网站的以下报价。

注意，无法设置以下Kafka参数，而Kafka源或接收器将抛出异常：enable.auto.commit：kafka源不提交任何offset。

我正在建立一个CDC系统，其中包括Kafka和Debezium，以捕获源表（MySQL）上的更改。然后，我将使用Pyspark清洁数据，最后将数据插入目标表。假设我需要停止Pyspark脚本2天。当我重新打开它时，它将不再捕获这两天中出现的数据。有解决这个问题的解决方案吗？ `

df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", kafka_bootstrap_servers) \
    .option("subscribe", topic) \
    .option("startingOffsets", "latest") \
    .option("group.id", consumer_group_id) \
    .load()

kafka在Spark

问题描述投票：0回答：2

2个回答

最新问题

kafka在Spark

问题描述 投票：0回答：2

2个回答

最新问题

问题描述投票：0回答：2