【SparkStreaming学习之三】 SparkStreaming和kafka整合

环境
　　虚拟机：VMware 10
　　Linux版本：CentOS-6.5-x86_64
　　客户端：Xshell4
　　FTP：Xftp4
　　jdk1.8
　　scala-2.10.4(依赖jdk1.8)
　　spark-1.6

一、receiver模式

1、receiver模式理解
在SparkStreaming程序运行起来后，Executor中会有receiver tasks接收kafka推送过来的数据。数据会被持久化，默认级别为MEMORY_AND_DISK_SER_2,这个级别也可以修改。receiver task对接收过来的数据进行存储和备份，这个过程会有节点之间的数据传输。备份完成后去zookeeper中更新消费偏移量，然后向Driver中的receiver tracker汇报数据的位置。最后Driver根据数据本地化将task分发到不同节点上执行。

2、receiver模式中存在的问题
当Driver进程挂掉后，Driver下的Executor都会被杀掉，当更新完zookeeper消费偏移量的时候，Driver如果挂掉了，就会存在找不到数据的问题，相当于丢失数据。
如何解决这个问题？
开启WAL(write ahead log)预写日志机制,在接受过来数据备份到其他节点的时候，同时备份到HDFS上一份（我们需要将接收来的数据的持久化级别降级到MEMORY_AND_DISK），这样就能保证数据的安全性。不过，因为写HDFS比较消耗性能，要在备份完数据之后才能进行更新zookeeper以及汇报位置等，这样会增加job的执行时间，这样对于任务的执行提高了延迟度。

3、receiver模式代码

package com.wjy.ss;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

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.storage.StorageLevel;
import org.apache.spark.streaming.Durations;
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;

public class SparkStreamingOnKafkaReceiver {

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("SparkStreamingOnKafkaReceiver").setMaster("local[2]");
        //开启预写日志 WAL机制
        conf.set("spark.streaming.receiver.writeAheadLog.enable", "true");
        JavaStreamingContext jsc = new JavaStreamingContext(conf,Durations.seconds(5));
        //可以保存接收的数据
        jsc.checkpoint("./receivedata");
        
        /**
         * 设置读取的topic和接受数据的线程数
         */
        Map<String, Integer> topicConsumerConcurrency = new HashMap<String, Integer>();
        topicConsumerConcurrency.put("MYTOPIC", 1);
        
        /**
         * 第一个参数是StreamingContext
         * 第二个参数是ZooKeeper集群信息（接受Kafka数据的时候会从Zookeeper中获得Offset等元数据信息）
         * 第三个参数是Consumer Group 消费者组
         * 第四个参数是消费的Topic以及并发读取Topic中Partition的线程数
         * 第五个参数设置receiver的存储级别  开启WAL机制 接收的数据存储级别要降级
         * 注意：
         * KafkaUtils.createStream 使用五个参数的方法，设置receiver的存储级别
         */
        JavaPairReceiverInputDStream<String, String> receiverDStream = KafkaUtils.createStream(jsc,
                "node3:2181,node4:2181,node5:2181",
                "MyFirstConsumerGroup", 
                topicConsumerConcurrency,
                StorageLevel.MEMORY_AND_DISK());
        
        receiverDStream.flatMap(new FlatMapFunction<Tuple2<String,String>, String>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Iterable<String> call(Tuple2<String, String> tuple)
                    throws Exception {
                return Arrays.asList(tuple._2.split("	"));
            }
        }).mapToPair(new PairFunction<String, String, Integer>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<String, Integer> call(String word) throws Exception {
                return new Tuple2<String, Integer>(word,1);
            }
        }).reduceByKey(new Function2<Integer, Integer, Integer>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1+v2;
            }
        }).print(100);
        
        jsc.start();
        jsc.awaitTermination();
        jsc.close();
    }

}

4、receiver的并行度设置
receiver的并行度是由spark.streaming.blockInterval来决定的，默认为200ms,假设batchInterval为5s,那么每隔blockInterval就会产生一个block,这里就对应每批次产生RDD的partition,这样5秒产生的这个Dstream中的这个RDD的partition为25个，并行度就是25。如果想提高并行度可以减少blockInterval的数值，但是最好不要低于50ms。

由于receiver模式存在的问题，目前这种模式在实际生产中用的较少。

二、Driect模式

  

1、Direct模式理解
SparkStreaming+kafka 的Driect模式就是将kafka看成存数据的一方，不是被动接收数据，而是主动去取数据。消费者偏移量也不是用zookeeper来管理，而是SparkStreaming内部对消费者偏移量自动来维护，默认消费偏移量是在内存中，当然如果设置了checkpoint目录，那么消费偏移量也会保存在checkpoint中。当然也可以实现用zookeeper来管理。

2、Direct模式并行度设置
Direct模式的并行度是由读取的kafka中topic的partition数决定的。

3、Direct模式代码

package com.wjy.ss;

import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;

import kafka.serializer.StringDecoder;

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.JavaPairInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka.KafkaUtils;

import scala.Tuple2;

public class SparkStreamingOnKafkaDirected {

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setMaster("local").setAppName("SparkStreamingOnKafkaDirected");
        conf.set("spark.streaming.backpressure.enabled", "false");
        conf.set("spark.streaming.kafka.maxRatePerPartition    ", "100");
        JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));
        /**
         * 可以不设置checkpoint 不设置不保存offset,offset默认在内存中有一份，如果设置checkpoint在checkpoint也有一份offset， 一般要设置。
         */
        jsc.checkpoint("./checkpoint");
        Map<String, String> kafkaParameters = new HashMap<String, String>();
        kafkaParameters.put("metadata.broker.list", "node1:9092,node2:9092,node3:9092");
        kafkaParameters.put("auto.offset.reset", "smallest");
        HashSet<String> topics = new HashSet<String>();
        topics.add("Mytopic");

        JavaPairInputDStream<String, String> lines = 
                KafkaUtils.createDirectStream(jsc,
                String.class,  
                String.class,
                StringDecoder.class,
                StringDecoder.class,
                kafkaParameters,
                topics);
        
        lines.flatMap(new FlatMapFunction<Tuple2<String,String>, String>() { 
            //如果是Scala，由于SAM转换，所以可以写成val words = lines.flatMap { line => line.split(" ")}
            private static final long serialVersionUID = 1L;

            public Iterable<String> call(Tuple2<String,String> tuple) throws Exception {
                return Arrays.asList(tuple._2.split("	"));
            }
        }).mapToPair(new PairFunction<String, String, Integer>() {
            private static final long serialVersionUID = 1L;

            public Tuple2<String, Integer> call(String word) throws Exception {
                return new Tuple2<String, Integer>(word, 1);
            }
        }).reduceByKey(new Function2<Integer, Integer, Integer>() { //对相同的Key，进行Value的累计（包括Local和Reducer级别同时Reduce）
            private static final long serialVersionUID = 1L;
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        }).print();
        
        jsc.start();
        jsc.awaitTermination();
        jsc.close();
        
    }

}

三、相关配置
1、反压机制:
spark.streaming.backpressure.enabled 默认false

2、blockInterval:
spark.streaming.blockInterval 默认200ms

3、接收数据速率:
spark.streaming.receiver.maxRate 默认没有设置

4、预写日志:
spark.streaming.receiver.writeAheadLog.enable 默认false没有开启

5、该参数决定是否需要以Gracefully方式来关闭Streaming程序（详情请参见SPARK-7776）。Spark会在启动 StreamingContext 的时候注册这个钩子
spark.streaming.stopGracefullyOnShutdown

6、每个分区每秒钟接收的消息数量
spark.streaming.kafka.maxRatePerPartition

参考：
Spark