Spark Streaming使用Kafka数据流示例

迷南。 2023-10-02 23:39 11阅读 0赞

Spark作为目前最流行的分布式计算框架，可用于处理实时数据流，本文主要介绍Spark Streaming如何使用Kafka数据流做分析，具体使用的版本信息见pom配置。

## 一、pom配置 ##

在IDE中创建好maven项目，在pom.xml添加如下配置：

<dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming_2.12</artifactId>
        <version>2.4.4</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
        <version>2.4.4</version>
    </dependency>

## 二、代码示例 ##

具体步骤详见代码示例：

import java.util.*;
    import org.apache.spark.SparkConf;
    import org.apache.spark.TaskContext;
    import org.apache.spark.api.java.*;
    import org.apache.spark.api.java.function.*;
    import org.apache.spark.streaming.api.java.*;
    import org.apache.spark.streaming.kafka010.*;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.common.TopicPartition;
    import org.apache.kafka.common.serialization.StringDeserializer;
    import scala.Tuple2;
    
    public class SparkStreamTest{
        
        public static void main(String[] args){
            // kafka配置信息
            Map<String, Object> kafkaConfig = new HashMap<>();
            kafkaConfig.put("bootstrap.servers", "localhost:9092,anotherhost:9092");
            kafkaConfig.put("key.deserializer", StringDeserializer.class);
            kafkaConfig.put("value.deserializer", StringDeserializer.class);
            kafkaConfig.put("group.id", "spark-test");
            kafkaConfig.put("auto.offset.reset", "latest");
            kafkaConfig.put("enable.auto.commit", true);
            // 需要消费的topic
            List<String> topics = Arrays.asList("test-topic");
            // 方便调试与查看输出，本例使用的是local模式，生产环境请不要使用。
            SparkConf sparkConf = new SparkConf().setMaster("local").setAppName("stream-test")
            // 这个地方说明下Durations.seconds(5)表示隔5秒钟处理一次
            JavaStreamingContext sparkContext = new JavaStreamingContext(conf,Durations.seconds(5));
            JavaInputDStream<ConsumerRecord<String, String>> dStream =
              KafkaUtils.createDirectStream(
                streamingContext,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.<String, String>Subscribe(topics, kafkaConfig)
              );
            // 转换成java键值对
            JavaPairDStream<String,String> pairStream = dStream.mapToPair(record -> new Tuple2<>(record.key(), record.value()));
            // 统计并打印（本地调试用）
            pairStream.count().print();
            //启用
            sparkContext.start();
            //本地调试，为避免主线程死亡加一个while循环
            while(true){
                try{
                    TimeUnit.SECONS.sleep(1);
                }catch(Exception e){
                    e.printStackTrace();
                }
            }
        }
    }

本地运行后，使用浏览器进入地址：localhost:4040，可查看运行状态，见下图。

![c1b7c040f19d40269c469f865f209154.jpeg][]

《完》

你学会了吗？赶紧动手试试吧。

[c1b7c040f19d40269c469f865f209154.jpeg]: https://img-blog.csdnimg.cn/c1b7c040f19d40269c469f865f209154.jpeg