文档章节

Spark Streaming StreamingContext详解+和Receiver详解+updateStateByKey+基于Direct的Kafka数据源

o
 osc_wws45aot
发布于 2019/08/20 15:27
字数 2986
阅读 15
收藏 0

精选30+云产品,助力企业轻松上云!>>>

一.StreamingContext详解

有两种创建StreamingContext的方式:

val conf = new SparkConf().setAppName(appName).setMaster(master);
val ssc = new StreamingContext(conf, Seconds(1));

StreamingContext,还可以使用已有的SparkContext来创建
val sc = new SparkContext(conf)
val ssc = new StreamingContext(sc, Seconds(1));

appName,是用来在Spark UI上显示的应用名称。master,是一个Spark、Mesos或者Yarn集群的URL,或者是local[*]。

batch interval可以根据你的应用程序的延迟要求以及可用的集群资源情况来设置。

 

一个StreamingContext定义之后,必须做以下几件事情:
1、通过创建输入DStream来创建输入数据源。
2、通过对DStream定义transformation和output算子操作,来定义实时计算逻辑。
3、调用StreamingContext的start()方法,来开始实时处理数据。
4、调用StreamingContext的awaitTermination()方法,来等待应用程序的终止。可以使用CTRL+C手动停止,或者就是让它持续不断的运行进行计算。
5、也可以通过调用StreamingContext的stop()方法,来停止应用程序。

需要注意的要点:
1、只要一个StreamingContext启动之后,就不能再往其中添加任何计算逻辑了。比如执行start()方法之后,还给某个DStream执行一个算子。
2、一个StreamingContext停止之后,是肯定不能够重启的。调用stop()之后,不能再调用start()
3、一个JVM同时只能有一个StreamingContext启动。在你的应用程序中,不能创建两个StreamingContext。
4、调用stop()方法时,会同时停止内部的SparkContext,如果不希望如此,还希望后面继续使用SparkContext创建其他类型的Context,比如SQLContext,那么就用stop(false)。
5、一个SparkContext可以创建多个StreamingContext,只要上一个先用stop(false)停止,再创建下一个即可。

 

二.输入DStream和Receiver详解

 

输入DStream代表了来自数据源的输入数据流。在之前的wordcount例子中,lines就是一个输入DStream(JavaReceiverInputDStream),代表了从netcat(nc)服务接收到的数据流。
除了文件数据流之外,所有的输入DStream都会绑定一个Receiver对象,该对象是一个关键的组件,用来从数据源接收数据,并将其存储在Spark的内存中,以供后续处理。
Spark Streaming提供了两种内置的数据源支持:
1、基础数据源:StreamingContext API中直接提供了对这些数据源的支持,比如文件、socket、Akka Actor等。
2、高级数据源:诸如Kafka、Flume、Kinesis、Twitter等数据源,通过第三方工具类提供支持。这些数据源的使用,需要引用其依赖。
3、自定义数据源:我们可以自己定义数据源,来决定如何接受和存储数据。
要注意的是,如果你想要在实时计算应用中并行接收多条数据流,可以创建多个输入DStream。这样就会创建多个Receiver,从而并行地接收多个数据流。但是要注意的是,一个Spark Streaming Application的Executor,是一个长时间运行的任务,因此,它会独占分配给Spark Streaming Application的cpu core。从而只要Spark Streaming运行起来以后,这个节点上的cpu core,就没法给其他应用使用了。

使用本地模式,运行程序时,绝对不能用local或者local[1],因为那样的话,只会给执行输入DStream的executor分配一个线程。而Spark Streaming底层的原理是,至少要有两条线程,一条线程用来分配给Receiver接收数据,一条线程用来处理接收到的数据。因此必须使用local[n],n>=2的模式。

如果不设置Master,也就是直接将Spark Streaming应用提交到集群上运行,那么首先,必须要求集群节点上,有>1个cpu core,其次,给Spark Streaming的每个executor分配的core,必须>1,这样,才能保证分配到executor上运行的输入DStream,两条线程并行,一条运行Receiver,接收数据;一条处理数据。否则的话,只会接收数据,不会处理数据。

因此,基于此,特此声明,我们本系列课程所有的练习,都是基于local[2]的本地模式,因为我们的虚拟机上都只有一个1个cpu core。但是大家在实际企业工作中,机器肯定是不只一个cpu core的,现在都至少4核了。到时记得给每个executor的cpu core,设置为超过1个即可。

 

 

三. updateStateByKey

 updateStateByKey操作,可以让我们为每个key维护一份state,并持续不断的更新该state。

 

1、首先,要定义一个state,可以是任意的数据类型;

2、其次,要定义state更新函数——指定一个函数如何使用之前的state和新值来更新state。

对于每个batch,Spark都会为每个之前已经存在的key去应用一次state更新函数,无论这个key在batch中是否有新的数据。如果state更新函数返回none,那么key对应的state就会被删除。

当然,对于每个新出现的key,也会执行state更新函数。

注意,updateStateByKey操作,要求必须开启Checkpoint机制。

案例:基于缓存的实时wordcount程序(在实际业务场景中,这个是非常有用的)

 代码如下

package com.hzk.sparkStreaming;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.Optional;
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.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

/**
 * 基于updateStateByKey算子实现缓存机制的实时wordcount程序
 * @author Administrator
 *
 */
public class UpdateStateByKeyWordCount  {
    public static void main(String[] args) throws InterruptedException {
        SparkConf conf = new SparkConf()
                .setMaster("local[2]")
                .setAppName("HDFSWordCount");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));
        // 第一点,如果要使用updateStateByKey算子,就必须设置一个checkpoint目录,开启checkpoint机制
        // 这样的话才能把每个key对应的state除了在内存中有,那么是不是也要checkpoint一份
        // 因为你要长期保存一份key的state的话,那么spark streaming是要求必须用checkpoint的,以便于在
        // 内存数据丢失的时候,可以从checkpoint中恢复数据
        // 开启checkpoint机制,很简单,只要调用jssc的checkpoint()方法,设置一个hdfs目录即可
        jssc.checkpoint("hdfs://hadoop-001:9000/datas/checkpoint");
        JavaReceiverInputDStream<String> lines=jssc.socketTextStream("localhost",9999);
        JavaDStream<String> words=lines.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String s) throws Exception {
                return Arrays.asList(s.split(" ")).iterator();
            }
        });
        JavaPairDStream<String,Integer> pairs=words.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<>(s,1);
            }
        });
        // 到了这里,就不一样了,之前的话,是不是直接就是pairs.reduceByKey
        // 然后,就可以得到每个时间段的batch对应的RDD,计算出来的单词计数
        // 然后,可以打印出那个时间段的单词计数
        // 但是,有个问题,你如果要统计每个单词的全局的计数呢?
        // 就是说,统计出来,从程序启动开始,到现在为止,一个单词出现的次数,那么就之前的方式就不好实现
        // 就必须基于redis这种缓存,或者是mysql这种db,来实现累加
        // 但是,我们的updateStateByKey,就可以实现直接通过Spark维护一份每个单词的全局的统计次数
        JavaPairDStream<String,Integer> wordCounts=pairs.updateStateByKey(new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() {
            // 这里的Optional,相当于Scala中的样例类,就是Option,可以这么理解
            // 它代表了一个值的存在状态,可能存在,也可能不存在
            // 这里两个参数
            // 实际上,对于每个单词,每次batch计算的时候,都会调用这个函数
            // 第一个参数,values,相当于是这个batch中,这个key的新的值,可能有多个吧
            // 比如说一个hello,可能有2个1,(hello, 1) (hello, 1),那么传入的是(1,1)
            // 第二个参数,就是指的是这个key之前的状态,state,其中泛型的类型是你自己指定的
            @Override
            public Optional<Integer> call(List<Integer> integers, Optional<Integer> state) throws Exception {
                // 首先定义一个全局的单词计数
                Integer newValue = 0;
                // 其次,判断,state是否存在,如果不存在,说明是一个key第一次出现
                // 如果存在,说明这个key之前已经统计过全局的次数了
                if (state.isPresent()){
                    newValue=state.get();
                }
                // 接着,将本次新出现的值,都累加到newValue上去,就是一个key目前的全局的统计
                // 次数
                for (Integer value:integers ) {
                    newValue+=value;
                }
                return Optional.of(newValue);
            }
        });
        // 到这里为止,相当于是,每个batch过来是,计算到pairs DStream,就会执行全局的updateStateByKey
        // 算子,updateStateByKey返回的JavaPairDStream,其实就代表了每个key的全局的计数
        // 打印出来
        wordCounts.print();

        jssc.start();
        jssc.awaitTermination();
        jssc.close();
    }
}

 

 

四.输入DStream之Kafka数据源实战(基于Direct的方式)

这种新的不基于Receiver的直接方式,是在Spark 1.3中引入的,从而能够确保更加健壮的机制。替代掉使用Receiver来接收数据后,这种方式会周期性地查询Kafka,来获得每个topic+partition的最新的offset,从而定义每个batch的offset的范围。当处理数据的job启动时,就会使用Kafka的简单consumer api来获取Kafka指定offset范围的数据。

这种方式有如下优点:
1、简化并行读取:如果要读取多个partition,不需要创建多个输入DStream然后对它们进行union操作。Spark会创建跟Kafka partition一样多的RDD partition,并且会并行从Kafka中读取数据。所以在Kafka partition和RDD partition之间,有一个一对一的映射关系。

2、高性能:如果要保证零数据丢失,在基于receiver的方式中,需要开启WAL机制。这种方式其实效率低下,因为数据实际上被复制了两份,Kafka自己本身就有高可靠的机制,会对数据复制一份,而这里又会复制一份到WAL中。而基于direct的方式,不依赖Receiver,不需要开启WAL机制,只要Kafka中作了数据的复制,那么就可以通过Kafka的副本进行恢复。

3、一次且仅一次的事务机制:
基于receiver的方式,是使用Kafka的高阶API来在ZooKeeper中保存消费过的offset的。这是消费Kafka数据的传统方式。这种方式配合着WAL机制可以保证数据零丢失的高可靠性,但是却无法保证数据被处理一次且仅一次,可能会处理两次。因为Spark和ZooKeeper之间可能是不同步的。
基于direct的方式,使用kafka的简单api,Spark Streaming自己就负责追踪消费的offset,并保存在checkpoint中。Spark自己一定是同步的,因此可以保证数据是消费一次且仅消费一次。

JavaPairReceiverInputDStream<String, String> directKafkaStream =
KafkaUtils.createDirectStream(streamingContext,
[key class], [value class], [key decoder class], [value decoder class],
[map of Kafka parameters], [set of topics to consume]);

4.WordCount程序代码如下

 

package com.hzk.sparkStreaming;

import kafka.serializer.StringDecoder;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.Optional;
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 scala.Tuple2;

import java.util.*;

public class KafkaDirectWordCount {
    public static void main(String[] args) throws InterruptedException {
        SparkConf conf = new SparkConf()
                .setMaster("local[2]")
                .setAppName("KafkaDirectWordCount");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));
        jssc.checkpoint("hdfs://hadoop-001:9000/datas/checkpoint");
        // 首先,要创建一份kafka参数map
        Map<String, String> kafkaParams = new HashMap<String, String>();
        kafkaParams.put("metadata.broker.list",
                "hadoop-001:9092");

        // 然后,要创建一个set,里面放入,你要读取的topic
        // 这个,就是我们所说的,它自己给你做的很好,可以并行读取多个topic
        Set<String> topics = new HashSet<String>();
        topics.add("test");

        // 创建输入DStream
        JavaPairInputDStream<String, String> lines = KafkaUtils.createDirectStream(
                jssc,
                String.class,
                String.class,
                StringDecoder.class,
                StringDecoder.class,
                kafkaParams,
                topics);

        // 执行wordcount操作
        JavaDStream<String> words = lines.flatMap(

                new FlatMapFunction<Tuple2<String,String>, String>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    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>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public Tuple2<String, Integer> call(String word) throws Exception {
                        return new Tuple2<String, Integer>(word, 1);
                    }

                });

        JavaPairDStream<String, Integer> wordCounts = pairs.updateStateByKey(new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() {
            @Override
            public Optional<Integer> call(List<Integer> integers, Optional<Integer> integerOptional) throws Exception {
               int value=0;
               if (integerOptional.isPresent()){
                   value=integerOptional.get();
               }
               for (Integer integer:integers){
                   value+=integer;
               }

                return Optional.of(value);
            }
        });

        wordCounts.print();

        jssc.start();
        jssc.awaitTermination();
        jssc.close();
    }
}

 

上一篇: Python Lambda
o
粉丝 0
博文 500
码字总数 0
作品 0
私信 提问
加载中
请先登录后再评论。
【Spark】Spark Streaming的程序运行原理及与Kafka的集成

1、Spark Streaming介绍 (1)Streaming Streaming:是一种数据传送技术,它把客户机收到的数据变成一个稳定连续的流,源源不断地送出,使用户听到的声音或看到的图象十分平稳,而且用户在整...

魏晓蕾
02/02
0
0
SparkStreaming基本架构及使用

1、简介 Spark Streaming处理的数据流图: Spark Streaming在内部的处理机制是,接收实时流的数据,并根据一定的时间间隔拆分成一批批的数据,然后通过Spark Engine处理这些批数据,最终得到...

^_TONY_^
2017/12/08
0
0
大数据技术之_19_Spark学习_04_Spark Streaming 应用解析小结

========== Spark Streaming 是什么 ========== 1、SPark Streaming 是 Spark 中一个组件,基于 Spark Core 进行构建,用于对流式进行处理,类似于 Storm。 2、Spark Streaming 能够和 Spar...

osc_fjbsh38p
2019/04/29
2
0
Spark Streaming 缓存与持久化机制+Checkpoint机制+部署、升级和监控应用程序

一.缓存与持久化机制 与RDD类似,Spark Streaming也可以让开发人员手动控制,将数据流中的数据持久化到内存中。对DStream调用persist()方法,就可以让Spark Streaming自动将该数据流中的所有...

osc_d2xp8kc1
2019/08/21
13
0
【Spark】SparkStreaming入门解析

一、Spark Streaming概述 官网 简介 是一个之上的、用于流式数据的处理框架,具有和等特点 可以从很多消费数据并对数据进行实时的处理,数据输入后可以用Spark的高度抽象原语如:map、reduc...

默默走开
04/10
0
0

没有更多内容

加载失败,请刷新页面

加载更多

为什么从HBase的0.96版本开始,舍弃了-ROOT-文件?

HBase结构的读写流程 (1). HBase0.96版本之前: (2). HBase0.96开始: a. 当客户端获取到.meta文件的位置之后,会缓存.meta.文件的位置 b. 客户端还会缓存HRegion的位置 -ROOT-存在的意义: ...

其乐m
36分钟前
18
0
volatile关键字对 - What is the volatile keyword useful for

问题: At work today, I came across the volatile keyword in Java. 今天的工作中,我遇到了Java中的volatile关键字。 Not being very familiar with it, I found this explanation: 不太熟......

技术盛宴
42分钟前
25
0
golang 封装 mysql 和 redis 连接

Mysql封装 package dbimport ("fmt"_ "github.com/go-sql-driver/mysql""github.com/jmoiron/sqlx")var DB *sqlx.DBfunc init(){database, err := sqlx.Op......

开源中国最牛的人
42分钟前
21
0
pdfbox 读取文件报错 java.io.IOException: Page tree root must be a dictionary

pdfbox java.io.IOException: Page tree root must be a dictionary 示例代码 public static void main(String[] args) { try (InputStream sampleInputs = new ClassPathResource("s......

lemos
50分钟前
28
0
整理 Linux下列出目录内容的命令

在 Linux 中,有非常多的命令可以让我们用来执行各种各样的任务。当我们想要像使用文件浏览器一样列出一个目录下的内容时,大家第一时间想到的是 ls 命令。但只有 ls 命令能实现这个目的吗?...

良许Linux
51分钟前
17
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部