代码之家 › 专栏 › 技术社区 › AmZ62210

使用Spark从Kafka主题中的分区中流数据

kafka-consumer-api spark-streaming apache-spark-sql apache-kafka apache-spark

0

AmZ62210 · 技术社区 · 8 年前

我已经看到一个类似的问题 clickhere

但我还是想知道从特定分区流式传输数据是否不可能?我已经用过 Spark流媒体订阅方法中的Kafka消费者策略 。

ConsumerStrategies.Subscribe[字符串,字符串](主题,Kafkaparams, 偏移量)

这是我尝试订阅主题和分区的代码片段,

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)

备注:cdc classic是一个主题名,有17个分区

2 回复 | 直到 8 年前

1

3

Bartosz Konieczny 8 年前

卡夫卡的分区是斯帕克的并行化单元。因此,即使从技术上讲这是可能的,这也没有意义,因为所有的数据都将由一个执行器处理。不必使用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开头?).

2

1

Alper t. Turker 8 年前

指定分区的分区号和起始偏移量,以便在此行中传输数据,

Map(new TopicPartition(topic, partition) -> 2L)

哪里,

partition是分区号
2l是分区的起始偏移量。

然后我们可以从选定的分区流式传输数据。