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

Spark流媒体应用程序订阅相同的卡夫卡主题

spark-streaming-kafka spark-streaming apache-kafka apache-spark

4

Gurubg · 技术社区 · 9 年前

我正在使用

spark-core_2.10 - 2.1.1
spark-streaming_2.10 - 2.1.1
spark-streaming-kafka-0-10_2.10 - 2.0.0
kafka_2.10 - 0.10.1.1

连续的事件数据流传输到卡夫卡主题,我需要从多个spark流应用程序处理该主题。但当我运行spark流媒体应用程序时,其中只有一个收到数据。

     Map<String, Object> kafkaParams = new HashMap<String, Object>();

     kafkaParams.put("bootstrap.servers", "localhost:9092");
     kafkaParams.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     kafkaParams.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); 
     kafkaParams.put("auto.offset.reset", "latest");
     kafkaParams.put("group.id", "test-consumer-group");
     kafkaParams.put("enable.auto.commit", "true");
     kafkaParams.put("auto.commit.interval.ms", "1000");
     kafkaParams.put("session.timeout.ms", "30000");

     Collection<String> topics =  Arrays.asList("4908100105999_000005");;
     JavaInputDStream<ConsumerRecord<String, String>> stream =  org.apache.spark.streaming.kafka010.KafkaUtils.createDirectStream(
                    ssc,
                    LocationStrategies.PreferConsistent(),
                    ConsumerStrategies.<String, String> Subscribe(topics, kafkaParams) );

      ... //spark processing

我有两个spark流媒体应用程序,通常我提交的第一个应用程序会使用kafka消息。第二个应用程序只是等待消息,从不继续。

2 回复 | 直到 7 年前

1

2

Sachin Thapa 9 年前

您可以使用相同的GroupID创建不同的流。以下是有关0.8集成的在线文档的更多详细信息,有两种方法:

可以使用不同的组和创建多个Kafka输入数据流使用多个接收器并行接收数据的主题。

方法2:直接方法(无接收器)

无需创建多个输入Kafka流并将其合并。具有有Kafka分区要使用,这些分区都将从中读取数据与卡夫卡平行。所以卡夫卡和 RDD分区,更易于理解和调整。

你可以在 Spark Streaming + Kafka Integration Guide 0.8

Spark Streaming + Kafka Integration Guide (Kafka broker version 0.10.0

即使它使用的是spark streaming api,但一切都由kafka属性控制,因此取决于您在属性文件中指定的组id,您可以使用不同的组id启动多个流。

干杯

2

1

Vinoth Chinnasamy 8 年前

[在消费者组下]的消费者数量不能超过主题中的分区数量。如果您想并行使用消息,那么需要引入适当数量的分区,并创建接收器来处理每个分区。