apache-spark - 使用 Spark 从 Kafka 主题中的特定分区流式传输数据
问题描述
我已经看到了与clickhere类似的问题
但我仍然想知道来自特定分区的流数据是否不可能?我在 Spark Streaming 订阅方法中使用了 Kafka Consumer Strategies。
ConsumerStrategies.Subscribe[String, String](topics, kafkaParams, offsets)
这是我尝试订阅主题和分区的代码片段,
val topics = Array("cdc-classic")
val topic="cdc-classic"
val partition=2;
val offsets=
Map(new TopicPartition(topic, partition) -> 2L)//I am not clear with this line, (I tried to set topic and partition number as 2)
val stream = KafkaUtils.createDirectStream[String, String](
ssc,
PreferConsistent,
Subscribe[String, String](topics, kafkaParams,offsets))
但是当我运行这段代码时,我得到以下异常,
Exception in thread "main" org.apache.spark.SparkException: Job aborted due to stage failure: Task 5 in stage 0.0 failed 1 times, most recent failure: Lost task 5.0 in stage 0.0 (TID 5, localhost, executor driver): org.apache.kafka.clients.consumer.OffsetOutOfRangeException: Offsets out of range with no configured reset policy for partitions: {cdc-classic-2=2}
at org.apache.kafka.clients.consumer.internals.Fetcher.parseCompletedFetch(Fetcher.java:878)
at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:525)
at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1110)
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1043)
at org.apache.spark.streaming.kafka010.CachedKafkaConsumer.poll(CachedKafkaConsumer.scala:99)
at org.apache.spark.streaming.kafka010.CachedKafkaConsumer.get(CachedKafkaConsumer.scala:70)
Caused by: org.apache.kafka.clients.consumer.OffsetOutOfRangeException: Offsets out of range with no configured reset policy for partitions: {cdc-classic-2=2}
at org.apache.kafka.clients.consumer.internals.Fetcher.parseCompletedFetch(Fetcher.java:878)
at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:525)
at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1110)
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1043)
at org.apache.spark.streaming.kafka010.CachedKafkaConsumer.poll(CachedKafkaConsumer.scala:99)
PS:cdc-classic是17个分区的主题名
解决方案
Kafka 的分区是 Spark 的并行化单元。因此,即使从技术上讲它是可能的,它也没有任何意义,因为所有数据都将由单个执行程序处理。您可以简单地启动您的流程,而不是使用 Spark KafkaConsumer
:
String topic = "foo";
TopicPartition partition0 = new TopicPartition(topic, 0);
TopicPartition partition1 = new TopicPartition(topic, 1);
consumer.assign(Arrays.asList(partition0, partition1));
(https://kafka.apache.org/0110/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html)
如果你想从 Spark 自动重试中获益,你可以简单地用它创建一个 Docker 镜像,然后使用 Kubernetes 启动它,并使用适当的重试配置。
关于 Spark,如果你真的想使用它,你应该检查你读取的分区的偏移量是多少。可能您提供了一个不正确的信息,它会返回“超出范围”偏移消息(可能以 0 开头?)。
推荐阅读
- mongodb - app.delete 不从集合中删除文档
- arrays - 如何在打字稿中的地图编号后添加“%”以进行前端反应?
- flutter - 如何检测颤振中的阻力速度?
- lua - 与 Lua 编程相关的帮助,其中 Basic Calculator 需要使用 3 个运算符加法、减法和乘法构建
- google-analytics - 有没有比这更好的教程在谷歌分析中使用谷歌标签管理器为客户 ID 创建自定义维度?
- javascript - 在 Firebase 托管上部署后,使用 Firebase 身份验证功能“signInWithEmailAndPassword”在 Web 上登录失败
- java - 如何在 openjdk:11 docker 镜像中安装 python?
- python - 使用 Python SDK 在 Azure 中创建虚拟机的用户名
- git - 从 git stash 返回到预览状态和提交中的顺序
- coldfusion - ColdFusion 8 中的过滤系统