Spark2.x中如何实现SparkStreaming消费Kafka实例

package com.unicom.ljs.spark220.study.streaming;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.common.TopicPartition;import org.apache.spark.SparkConf;import org.apache.spark.streaming.Durations;import org.apache.spark.streaming.api.java.JavaInputDStream;import org.apache.spark.streaming.api.java.JavaPairDStream;import org.apache.spark.streaming.api.java.JavaStreamingContext;import org.apache.spark.streaming.kafka010.ConsumerStrategies;import org.apache.spark.streaming.kafka010.KafkaUtils;import org.apache.spark.streaming.kafka010.LocationStrategies;import scala.Tuple2;import java.util.*;/** * @author: Created By lujisen * @company ChinaUnicom Software JiNan * @date: 2020-01-31 20:30 * @version: v1.0 * @description: com.unicom.ljs.spark220.study.streaming */public class KafkaStreamingWordCount {    public static void main(String[] args) throws InterruptedException {        SparkConf sparkConf = new SparkConf().setMaster("local[*]").setAppName("KafkaStreamingWordCount");        JavaStreamingContext ssc=new JavaStreamingContext(sparkConf, Durations.seconds(5));        String  topic="kafkawordcount";        Collection<String> topics = new HashSet<>();        topics.add(topic);        //kafka相关参数，其他参数可自行百度        String brokerList = "10.124.165.31:6667，10.124.165.32:6667";        Map<String, Object> props = new HashMap<>();        props.put("bootstrap.servers", brokerList);        props.put("group.id", "groupLjs1");        props.put("auto.offset.reset", "earliest");        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");        /*指定kafka中topic的消费分区*/        Map<TopicPartition, Long> offsets = new HashMap<>();        offsets.put(new TopicPartition(topic, 0), 0L);        offsets.put(new TopicPartition(topic, 1), 0L);        //通过KafkaUtils.createDirectStream指定kafka数据源        // 三个参数   1 sparkcontext  2.LocationStrategies.PreferConsistent,如上所示。这将在可用执行程序之间均匀分配分区  3，订阅kafka 的配置        JavaInputDStream<ConsumerRecord<Object, Object>> lines = KafkaUtils.createDirectStream(                ssc,                LocationStrategies.PreferConsistent(),                ConsumerStrategies.Subscribe(topics, props, offsets)        );        JavaPairDStream<String, Integer> counts =lines.flatMap(                x -> Arrays.asList(x.value().toString().split(" ")).iterator())                .mapToPair(x -> new Tuple2<String, Integer>(x, 1)).reduceByKey((x, y) -> x + y);        /*打印结果*/        counts.print();        /*启动*/        ssc.start();        ssc.awaitTermination();        /*停止*/        ssc.close();    }}