Spark Streaming与Kafka的整合

    Spark Streaming实时流处理,与Kafka消息队列的整合。实时处理中,kafka作为消息中间件,Spark Streaming作为数据处理工具,根据Spark Streaming数据接收的方式,将Spark Streaming与kafka的整合分为两种:1. Receiver模式,由kafka将数据发送数据,Spark Streaming被动接收数据; 2. Direct模式,由Spark Streaming主动去kafka中拉取数据。

1. Receiver模式

    Receiver模式,在Spark Streaming程序启动后,由receiver task接收kafka推送过来的数据,并将数据进行持久化【默认持久化等级:Memory_And_Disk_Ser_2】及备份到其他节点上,当数据完成持久化过程后,会将此次消费的offset偏移量信息保存到Zookeeper中。然后,由Driver根据数据位置信息,分发任务到最佳节点上运行。

    代码实现:

     Kafka生产数据代码

/**

* 模拟kafka的数据生产者

*/

public class KafkaProducer {

    private static String topic_name = "test";

    public static void main(String[] args) {

        Properties prop = new Properties();

        prop.put("metadata.broker.list", "node01:9092,node02:9092,node03:9092");

        prop.put("serializer.class", StringEncoder.class.getName());

        Producer<Integer, String> producer = new Producer<Integer, String>(new ProducerConfig(prop));

        String[] words = new String[]{"JianShu", "ZhiHu", "CSDN", "BoKeYuan"};

        while (true) {

            Random random = new Random();

            int index = random.nextInt(4);

            System.out.println(index);

            producer.send(new KeyedMessage<>(topic_name, words[index]));

        }

    }

}

Receiver模式 Spark Streaming实时处理数据

public class SparkStream_Receiver {

    private static String topic_name = "test";

    public static void main(String[] args) {

        SparkConf conf = new SparkConf();

        // 至少需要2个executor进程,一个启动driver的receive task,一个处理数据

        conf.setMaster("local[2]").setAppName("Receiver");

        // 每隔3秒将数据封装成一个DStream

        JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(3));

        Map<String, Integer> map = new HashMap<>();

        map.put(topic_name, 1);

        /**

        * 简单单词计数

        */

        JavaPairReceiverInputDStream<String, String> receiverInputDStream = KafkaUtils.createStream(jsc,

                "node01:2181,node02:2181,node03:2181", "test", map);

        JavaPairDStream<Tuple2<String, String>, Integer> pairWord = receiverInputDStream.mapToPair(word -> new Tuple2<>(word, 1));

        JavaPairDStream<Tuple2<String, String>, Integer> result = pairWord.reduceByKey((x, y) -> x + y);

        result.print();

        // 启动Spark Streaming

        jsc.start();

        try {

            jsc.awaitTermination();

        } catch (InterruptedException e) {

            e.printStackTrace();

        }

        // 关闭Spark Streaming

        jsc.stop();

    }

}

    到此,简单Receiver模式已实现。

    Receiver模式存在的问题:

        在数据已持久化,且偏移量已提交到Zookeeper后,driver挂掉,此时executor也会挂掉,此时则会导致当前正在处理的数据还未计算完,而下次处理,则会从zookeeper中获取偏移量,往后继续获取数据,那么此次处理的数据就会丢失

    解决方案:开启WAL机制,将数据备份到HDFS,当遇到上述情况,则可以从HDFS上读取数据,防止数据丢失的问题。此种机制的弊端是,需要消耗性能将数据写入到HDFS。

2. Direct模式

    Direct模式,在Spark Streaming程序启动后,Spark Streaming程序主动去Kafka中拉取数据,且数据消费的偏移量也是由自身维护,若由设置checkpoint目录,也会将其持久化到checkpoint中。

    代码实现:【kafka数据生产实现代码与Receiver模式相同】

    Direct模式:

public class SparkStream_Direct {

private static Stringtopic_name ="test";

public static void main(String[] args) {

SparkConf conf =new SparkConf();

// 至少需要2个executor进程,一个启动driver的receive task,一个处理数据

        conf.setMaster("local[2]").setAppName("Direct");

// 每隔3秒将数据封装成一个DStream

        JavaStreamingContext jsc =new JavaStreamingContext(conf, Durations.seconds(3));

Map kafkaconf =new HashMap();

kafkaconf.put("metadata.broker.list","node01:9092,node02:9092,node03:9092");

Set topics =new HashSet<>();

topics.add(topic_name);

/**

        * 简单单词计数

        */

        // 与Receiver不同之处

        JavaPairInputDStream directStream = KafkaUtils.createDirectStream(jsc, String.class, String.class, StringDecoder.class, StringDecoder.class, kafkaconf, topics);

JavaPairDStream, Integer> pairWord = directStream.mapToPair(word ->new Tuple2<>(word,1));

JavaPairDStream, Integer> result = pairWord.reduceByKey((x, y) -> x + y);

result.print();

// 启动Spark Streaming

        jsc.start();

try {

jsc.awaitTermination();

}catch (InterruptedException e) {

e.printStackTrace();

}

// 关闭Spark Streaming

        jsc.stop();

}

}



    

        

©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容