spark streaming 接收 kafka 數據java代碼WordCount示例

1. 首先啓動zookeeper

2. 啓動kafka

3. 核心代碼

• 生產者生產消息的java代碼，生成要統計的單詞

package streaming;

import java.util.Properties; 
   
import kafka.javaapi.producer.Producer; 
import kafka.producer.KeyedMessage; 
import kafka.producer.ProducerConfig; 
   
public class MyProducer {   
     
        public static void main(String[] args) {   
            Properties props = new Properties();   
            props.setProperty("metadata.broker.list","localhost:9092");   
            props.setProperty("serializer.class","kafka.serializer.StringEncoder");   
            props.put("request.required.acks","1");   
            ProducerConfig config = new ProducerConfig(props);   
            //建立生產這對象
            Producer<String, String> producer = new Producer<String, String>(config);
            //生成消息
            KeyedMessage<String, String> data1 = new KeyedMessage<String, String>("top1","test kafka");
            KeyedMessage<String, String> data2 = new KeyedMessage<String, String>("top2","hello world");
            try {   
                int i =1; 
                while(i < 100){    
                    //發送消息
                    producer.send(data1);   
                    producer.send(data2);
                    i++;
                    Thread.sleep(1000);
                } 
            } catch (Exception e) {   
                e.printStackTrace();   
            }   
            producer.close();   
        }   
}

• 在SparkStreaming中接收指定話題的數據，對單詞進行統計

package streaming;
import java.util.HashMap;
import java.util.Map;
import java.util.regex.Pattern;

import org.apache.spark.*;
import org.apache.spark.api.java.function.*;
import org.apache.spark.streaming.*;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaPairReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka.KafkaUtils;

import scala.Tuple2;

import com.google.common.collect.Lists;
public class KafkaStreamingWordCount {

    public static void main(String[] args) {
        //設置匹配模式，以空格分隔
        final Pattern SPACE = Pattern.compile(" ");
        //接收數據的地址和端口
        String zkQuorum = "localhost:2181";
        //話題所在的組
        String group = "1";
        //話題名稱以「，」分隔
        String topics = "top1,top2";
        //每一個話題的分片數
        int numThreads = 2;        
        SparkConf sparkConf = new SparkConf().setAppName("KafkaWordCount").setMaster("local[2]");
        JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new Duration(10000));
//        jssc.checkpoint("checkpoint"); //設置檢查點
        //存放話題跟分片的映射關係
        Map<String, Integer> topicmap = new HashMap<>();
        String[] topicsArr = topics.split(",");
        int n = topicsArr.length;
        for(int i=0;i<n;i++){
            topicmap.put(topicsArr[i], numThreads);
        }
        //從Kafka中獲取數據轉換成RDD
        JavaPairReceiverInputDStream<String, String> lines = KafkaUtils.createStream(jssc, zkQuorum, group, topicmap);
        //從話題中過濾所需數據
        JavaDStream<String> words = lines.flatMap(new FlatMapFunction<Tuple2<String, String>, String>() {

            @Override
            public Iterable<String> call(Tuple2<String, String> arg0)
                    throws Exception {
                return Lists.newArrayList(SPACE.split(arg0._2));
            }
        });
        //對其中的單詞進行統計
        JavaPairDStream<String, Integer> wordCounts = words.mapToPair(
              new PairFunction<String, String, Integer>() {
                @Override
                public Tuple2<String, Integer> call(String s) {
                  return new Tuple2<String, Integer>(s, 1);
                }
              }).reduceByKey(new Function2<Integer, Integer, Integer>() {
                @Override
                public Integer call(Integer i1, Integer i2) {
                  return i1 + i2;
                }
              });
        //打印結果
        wordCounts.print();
        jssc.start();
        jssc.awaitTermination();

    }

}