spark消费kafka消息

待我称王封你为后i 2022-05-14 05:45 288阅读 0赞

教程：

[http://spark.apache.org/docs/2.2.0/streaming-kafka-0-10-integration.html][http_spark.apache.org_docs_2.2.0_streaming-kafka-0-10-integration.html]

pom：

<dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.11</artifactId>
        <version>2.3.1</version>
    </dependency>
    
    
    
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming_2.11</artifactId>
        <version>2.3.1</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
        <version>2.3.1</version>
    </dependency>

String brokers = "127.0.0.1:9092";
    String topics = "helloSpark";
    
    // Create context with a 2 seconds batch interval
    SparkConf sparkConf = new SparkConf().setMaster("local").setAppName("JavaDirectKafkaWordCount");
    JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, Durations.seconds(2));
    
    Set<String> topicsSet = new HashSet<>(Arrays.asList(topics.split(",")));
    Map<String, Object> kafkaParams = new HashMap<>();
    kafkaParams.put("bootstrap.servers", brokers);
    kafkaParams.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    kafkaParams.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    kafkaParams.put("serializer.class", "kafka.serializer.StringEncoder");
    kafkaParams.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    kafkaParams.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    kafkaParams.put("group.id","helloSpark");
    
    // Create direct kafka stream with brokers and topics
    JavaInputDStream<ConsumerRecord<String, String>> messages =
            KafkaUtils.createDirectStream(jssc, LocationStrategies.PreferConsistent(), ConsumerStrategies.Subscribe(topicsSet, kafkaParams));
    
    // Get the lines, split them into words, count the words and print
    JavaDStream<String> lines = messages.map(ConsumerRecord::value);
    JavaDStream<String> words = lines.flatMap(x -> Arrays.asList(SPACE.split(x)).iterator());
    JavaPairDStream<String, Integer> wordCounts = words.mapToPair(s->new Tuple2<>(s,1)).reduceByKey((i1, i2) -> i1 + i2);
    wordCounts.print();
    
    // Start the computation
    jssc.start();
    jssc.awaitTermination();
    jssc.stop(false);

#### 1、LocationStrategies ####

新的Kafka消费者API将预先获取消息到缓冲区，出于性能原因，Spark会将缓存过的消费者保留在执行程序上（而不是为每个批处理重新创建它们）。

在大多数情况下，使用LocationStrategies.PreferConsistent如上所示，这将在可用执行程序之间均匀分配分区。如果执行程序与Kafka brokers位于同一主机上，使用PreferBrokers，这将优先为该分区的Kafka领导者安排分区。如果分区之间的负载有明显偏差，请使用PreferFixed，可以指定分区到主机的显式映射。

消费者的缓存的默认最大大小为64.如果您希望处理超过（64个执行程序数）Kafka分区，可以更改设置：spark.streaming.kafka.consumer.cache.maxCapacity。

如果要禁用Kafka消费者的缓存，可以设置spark.streaming.kafka.consumer.cache.enabled为false。可能需要禁用缓存来解决SPARK-19185中描述的问题。一旦SPARK-19185解决，可以在Spark的更高版本中删除此属性。

#### 2、ConsumerStrategies ####

三种订阅topic的策略：

*  ConsumerStrategies.Subscribe，如上所示，允许订阅固定的主题集合
 *  SubscribePattern允许使用正则表达式指定感兴趣的主题
 *  Assign允许您指定固定的分区集合

这三种策略都有重载的构造函数，允许指定特定分区的起始偏移量

### 指定消息offset批量处理 ###

如果用例更适合批处理，则可以为定义的偏移范围创建RDD：

OffsetRange[] offsetRanges = {
      // topic, partition, inclusive starting offset, exclusive ending offset
      OffsetRange.create("test", 0, 0, 100),
      OffsetRange.create("test", 1, 0, 100)
    };
    
    JavaRDD<ConsumerRecord<String, String>> rdd = KafkaUtils.createRDD(
      sparkContext,
      kafkaParams,
      offsetRanges,
      LocationStrategies.PreferConsistent()
    );

[http_spark.apache.org_docs_2.2.0_streaming-kafka-0-10-integration.html]: http://spark.apache.org/docs/2.2.0/streaming-kafka-0-10-integration.html