java spark-streaming接收TCP/Kafka数据

本文将展示

1、如何使用spark-streaming接入TCP数据并进行过滤；

2、如何使用spark-streaming接入TCP数据并进行wordcount；

内容如下：

1、使用maven，先解决pom依赖

<dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka_2.10</artifactId>
            <version>1.6.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.10</artifactId>
            <version>1.6.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.10</artifactId>
            <version>1.6.0</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.10</artifactId>
            <version>1.6.0</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.10</artifactId>
            <version>1.6.0</version>
            <scope>provided</scope>
        </dependency>

1、接收TCP数据并过滤，打印含有error的行

package com.xiaoju.dqa.realtime_streaming;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.Durations;

//nc -lk 9999
public class SparkStreamingTCP {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setMaster("local").setAppName("streaming word count");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(1));
        JavaDStream<String> lines = jssc.socketTextStream("10.93.21.21", 9999);
        JavaDStream<String> errorLines = lines.filter(new Function<String, Boolean>() {
            @Override
            public Boolean call(String s) throws Exception {
                return s.contains("error");
            }
        });
        errorLines.print();
        jssc.start();
        jssc.awaitTermination();
    }
}

执行方法

$ spark-submit realtime-streaming-1.0-SNAPSHOT-jar-with-dependencies.jar# 另起一个窗口$ nc -lk 9999# 输入数据

2、接收Kafka数据并进行计数(WordCount)

package com.xiaoju.dqa.realtime_streaming;

import java.util.*;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
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.api.java.*;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka.KafkaUtils;
import org.apache.spark.streaming.Durations;

import scala.Tuple2;

// bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
public class SparkStreamingKafka {
    public static void main(String[] args) throws InterruptedException {
        SparkConf conf = new SparkConf().setMaster("yarn-client").setAppName("streaming word count");
        //String topic = "offline_log_metrics";
        String topic = "test";
        int part = 1;
        JavaSparkContext sc = new JavaSparkContext(conf);
        sc.setLogLevel("WARN");
        JavaStreamingContext jssc = new JavaStreamingContext(sc, Durations.seconds(10));
        Map<String ,Integer> topicMap = new HashMap<String, Integer>();
        String[] topics = topic.split(";");
        for (int i=0; i<topics.length; i++) {
            topicMap.put(topics[i], 1);
        }
        List<JavaPairReceiverInputDStream<String, String>> list = new ArrayList<JavaPairReceiverInputDStream<String, String>>();
        for (int i = 0; i < part; i++) {
            list.add(KafkaUtils.createStream(jssc,
                    "10.93.21.21:2181",
                    "bigdata_qa",
                    topicMap));
        }
        JavaPairDStream<String, String> wordCountLines = list.get(0);
        for (int i = 1; i < list.size(); i++) {
            wordCountLines = wordCountLines.union(list.get(i));
        }
        JavaPairDStream<String, Integer> counts = wordCountLines.flatMap(new FlatMapFunction<Tuple2<String, String>, String>(){
            @Override
            public Iterable<String> call(Tuple2<String, String> stringStringTuple2){
                List<String> list2 = null;
                try {
                    if ("".equals(stringStringTuple2._2) || stringStringTuple2._2 == null) {
                        System.out.println("_2 is null");
                        throw new Exception("_2 is null");
                    }
                    list2 = Arrays.asList(stringStringTuple2._2.split(" "));
                } catch (Exception ex) {
                    ex.printStackTrace();
                    System.out.println(ex.getMessage());
                }
                return list2;
            }
        }).mapToPair(new PairFunction<String, String, Integer>() {
            public Tuple2<String, Integer> call(String s) throws Exception {
                Tuple2<String, Integer> tuple2 = null;
                try {
                    if (s==null || "".equals(s)) {
                        tuple2 = new Tuple2<String, Integer>(s, 0);
                        throw new Exception("s is null");
                    }
                    tuple2 = new Tuple2<String, Integer>(s, 1);
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
                return tuple2;
            }
        }).reduceByKey(new Function2<Integer, Integer, Integer>() {
            public Integer call(Integer x, Integer y) throws Exception {
                return x + y;
            }
        });
        counts.print();

        jssc.start();
        try {
            jssc.awaitTermination();
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            jssc.close();
        }
    }
}

执行方法

 $ spark-submit --queue=root.XXX realtime-streaming-1.0-SNAPSHOT-jar-with-dependencies.jar# 另开一个窗口，启动kafka生产者$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test# 输入数据