Spark Streaming入门

￡神魔★判官ぃ 2022-05-21 00:50 202阅读 0赞

为了初始化Spark Streaming程序，一个StreamingContext对象必需被创建，它是Spark Streaming所有流操作的主要入口。一个StreamingContext 对象可以用SparkConf对象创建。 可以使用SparkConf对象创建JavaStreamingContext对象：

SparkConf conf = new SparkConf().setMaster("local").setAppName("testLocal");
    JavaStreamingContext ssc = new JavaStreamingContext(conf, Durations.seconds(2));

JavaStreamingContext对象也可以从现有的JavaSparkContext创建：

SparkConf conf = new SparkConf().setMaster("local").setAppName("testLocal");
    JavaSparkContext sc = new JavaSparkContext(conf);
    JavaStreamingContext ssc = new JavaStreamingContext(sc, Durations.seconds(2));

--------------------

> 实验：Flume +kafka+SparkStreaming

1.  启动kafka:./bin/kafka-server-start.sh ../config/server.properties
2.  启动flume:./bin/flume-ng agent –conf conf –conf-file ./conf/kafka.conf -name a1 -Dflume.root.logger=DEBUG,console
3.  启动kafka consumer：./kafka-console-consumer.sh –zookeeper vm04:2181 –topic test\_m\_brokers –from-beginning
4.  在eclipse里运行程序，会在控制台显示运行结果

kafka.conf

# Name the components on this agent
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    # Describe/configure the source
    a1.sources.r1.type = exec
    a1.sources.r1.command = tail -f /root/data/flume/data-produce.log
    
    # Describe the sink
    a1.sinks.k1.type= org.apache.flume.sink.kafka.KafkaSink
    a1.sinks.k1.brokerList=vm04:9092
    a1.sinks.k1.topic=test_m_brokers
    #a1.sinks.k1.serializer.class=kafka.serializer.StringEncoder
    
    # Use a channel which buffers events in memory
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    # Bind the source and sink to the channel
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

package test;
    
    import java.util.Arrays;
    import java.util.HashMap;
    import java.util.HashSet;
    import java.util.Iterator;
    import java.util.Map;
    import java.util.Set;
    import org.apache.spark.SparkConf;
    import org.apache.spark.api.java.function.FlatMapFunction;
    import org.apache.spark.api.java.function.Function2;
    import org.apache.spark.api.java.function.PairFunction;
    import org.apache.spark.streaming.Durations;
    import org.apache.spark.streaming.api.java.JavaDStream;
    import org.apache.spark.streaming.api.java.JavaPairDStream;
    import org.apache.spark.streaming.api.java.JavaPairInputDStream;
    import org.apache.spark.streaming.api.java.JavaStreamingContext;
    import org.apache.spark.streaming.kafka.KafkaUtils;
    import kafka.serializer.StringDecoder;
    import scala.Tuple2;
    
    public class KafkaTest {
        public static void main(String[] args) throws InterruptedException {
            String brokers = "192.168.122.250:6667"; 
            SparkConf conf = new SparkConf().setAppName("kafkatest").setMaster("local[4]");
            JavaStreamingContext ssc = new JavaStreamingContext(conf,Durations.seconds(2));
    
            Map<String, String> kafkaParams = new HashMap<String, String>();
            //zookeeper集群的配置信息
            kafkaParams.put("bootstrap.servers",
                    "vm04:9092,vm05:9092,vm06:9092");
            Set<String> topics = new HashSet<String>();
            topics.add("test_m_brokers");
            JavaPairInputDStream<String, String> lines = KafkaUtils.
                    createDirectStream(ssc, String.class, String.class,
                            StringDecoder.class, StringDecoder.class, kafkaParams, topics);
            JavaDStream<String> words = lines.flatMap(new FlatMapFunction<Tuple2<String,String>, String>() {
                public Iterator<String> call(Tuple2<String, String> tuple) throws Exception {
                    return Arrays.asList(tuple._2.split(",")).iterator();
                }
            });
    
            JavaPairDStream<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {
    
                public Tuple2<String, Integer> call(String word) throws Exception {
                    return new Tuple2<String, Integer>(word, 1);
                }
            });
    
            JavaPairDStream<String, Integer> word_count= pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
    
                public Integer call(Integer v1, Integer v2) throws Exception {
                    return v1+v2;
                }
            });
            //      word_count.foreachRDD(rdd -> {
    //          rdd.foreach(x -> {
    //              System.out.println(x);
    //          });
    //      });
            word_count.print();
            ssc.start();
            ssc.awaitTermination();
    
        }
    
    }

一开始报了错，我使用的是Spark 2.1.0，把`meteadata.broker.list`换成`bootstrap.servers`就可以了，具体原因还不是很清楚。  
![这里写图片描述][70]

结果如下：  
![这里写图片描述][70 1]

[70]: /images/20220521/3c0d8bd126824011903042e9c838fe2b.png
[70 1]: /images/20220521/8a28d475ed83405e9078a2d60c3c4078.png