Spark Streaming编程技巧是什么

发布时间：2021-12-16 14:52:20 来源：亿速云阅读：169 作者：iii 栏目：云计算

本篇内容主要讲解“Spark Streaming编程技巧是什么”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“Spark Streaming编程技巧是什么”吧!

#Spark Streaming 编程指南#

##概述## Spark Streaming 是核心Spark API的一个扩展，他可以实现高吞吐量，和容错的实时数据流处理。

他可以接受许多数据源例如Kafka、Flume、Twitter、ZeroMQ或者普通的老的TCP套接字的数据。数据可以使用拥有高级函数例如map、reduce、join、和window的复杂算法表达式进行处理。最终，处理的数据可以被推送到文件系统、数据库和在线仪表盘。实际上，你可以在数据流上应用Spark内置的机器学习算法和图处理算法。

在内部，它的工作原理如下。Spark Streaming接收实时输入数据流，并且将数据分割成batches，which are then processed by the Spark engine to generate the final stream of results in batches.

Spark Streaming 提供一个高级的抽象叫做离散流，或者DStream。它表示一个连续不断的数据流。DStreams既可以通过来自数据源例如Kafka、Flume的数据数据流创建，也可以通过在其他DStreams上应用高级操作创建。在内部，一个DStream被表示成一个RDDs的序列。

本指南向你展示如何使用DStreams开始编写Spark Streaming程序。你可以使用Scala或Java编写Spark Streaming程序，本指南中两者都提供。你将会发现tabs贯穿全文，可以让你在Scala和Java代码片段中选择。

##一个简单的例子## 在我们进入如何编写你自己的Spark Streaming程序的细节之前，让我们快速的看下一个简单的Spark Streaming程序是怎样的。比如说，我们想计算一个通过监听TCP套接字得到的数据服务器上的文本数据中单词的总数。所有你需要做的如下：

首先，我们创建一个JavaStreamingContext对象，他是所有Streaming功能的一个切入点。除了Spark的配置，we specify that any DStream would be processed in 1 second batches.

import org.apache.spark.api.java.function.*;
import org.apache.spark.streaming.*;
import org.apache.spark.streaming.api.java.*;
import scala.Tuple2;
// Create a StreamingContext with a local master
JavaStreamingContext jssc = new JavaStreamingContext("local[2]", "JavaNetworkWordCount", new Duration(1000))

使用这个context，我们通过指定IP地址和数据服务器的端口来创建一个新的DStream。

// Create a DStream that will connect to serverIP:serverPort, like localhost:9999
JavaReceiverInputDStream<String> lines = jssc.socketTextStream("localhost", 9999);

这个DStream lines表示数据的流将会从这个数据服务器接收。流中的每一条记录都是一行文本。然后，我们通过空格将行分割成单词。

// Split each line into words
JavaDStream<String> words = lines.flatMap(
  new FlatMapFunction<String, String>() {
    @Override public Iterable<String> call(String x) {
      return Arrays.asList(x.split(" "));
    }
  });

flatMap是一个DStream操作，它通过使源DStream中的每一条记录生成许多新的记录而创建一个新的DStream。在这个例子中，每一行将会被分割成多个words，words流被表示成words DStream。注意，我们定义使用FlatMapFunction对象转换。正如我们一直在探索，在Java API中有许多这样的转换类来帮助定义DStream转换。

接下俩，我们想要计算这些words的和：

// Count each word in each batch
JavaPairDStream<String, Integer> pairs = words.map(
  new PairFunction<String, String, Integer>() {
    @Override public Tuple2<String, Integer> call(String s) throws Exception {
      return new Tuple2<String, Integer>(s, 1);
    }
  });
JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey(
  new Function2<Integer, Integer, Integer>() {
    @Override public Integer call(Integer i1, Integer i2) throws Exception {
      return i1 + i2;
    }
  });
wordCounts.print();     // Print a few of the counts to the console

使用一个PairFunction，words DStream被进一步mapped(一对一转换)成一个DStream对(word,1)。然后，使用Function2对象， it is reduced to get the frequency of words in each batch of data。最后，wordCounts.print()将会每秒打印一些生成的和。

Note that when these lines are executed, Spark Streaming only sets up the computation it will perform when it is started, and no real processing has started yet. To start the processing after all the transformations have been setup, we finally call

jssc.start();              // Start the computation
jssc.awaitTermination();   // Wait for the computation to terminate

完整的代码可以再Spark Streaming example JavaNetworkWordCount找到。

如果你已经下载并且构建了Spark，你可以像下面这样运行这个例子。你需要首先运行Netcat（一个可以再大多数Unix-like系统上找到的小工具）作为一个数据服务器，通过：

$ nc -lk 9999

然后，在一个不同的终端下，亦可以启动这个例子，通过：

$ ./bin/run-example org.apache.spark.examples.streaming.JavaNetworkWordCount localhost 9999

然后，在运行netcat服务的终端中输入的每一行将会被求和并且每秒打印在屏幕上。他看起来像这样：

<pre> # TERMINAL 1: # Running Netcat $ nc -lk 9999 hello world ... # TERMINAL 2: RUNNING NetworkWordCount or JavaNetworkWordCount $ ./bin/run-example org.apache.spark.examples.streaming.NetworkWordCount localhost 9999 ... ------------------------------------------- Time: 1357008430000 ms ------------------------------------------- (hello,1) (world,1) ... </pre>

你也可以在Spark shell直接使用Spark Streaming：

$ bin/spark-shell

... 并且通过封装已存在的交互式shell SparkContext对象sc来创建你的StreamingContext：

val ssc = new StreamingContext(sc, Seconds(1))

When working with the shell, you may also need to send a ^D to your netcat session to force the pipeline to print the word counts to the console at the sink.

##基础知识## 现在，我们move beyond the simple example，我们详细阐述编写一个streaming应用程序你需要了解的Spark Streaming的基础知识。

###接入### 要编写你自己的Spark Streaming程序，你将需要添加下面的依赖到你的SBT或者Maven项目中:

groupId = org.apache.spark
artifactId = spark-streaming_2.10
version = 1.0.2

对于从像Kafka和Flume这样的数据源获取数据的功能，现在已经出现在Spark Streaming核心API里。你将需要添加相应的attiface spark-streaming-xyz_2.10到依赖。例如，下面是一些常见的：

<pre> Source Artifact Kafka spark-streaming-kafka_2.10 Flume spark-streaming-flume_2.10 Twitter spark-streaming-twitter_2.10 ZeroMQ spark-streaming-zeromq_2.10 MQTT spark-streaming-mqtt_2.10 </pre>

罪行的列表，请参考Apache repository获得所有支持的源和artifacts的列表。

###初始化### 在Java中，要初始化一个Spark Streaming程序，需要创建一个JavaStreamingContext对象，他是整个Spark Streaming 功能的切入点。一个JavaStreamingContext对象可以被创建，使用：

	new JavaStreamingContext(master, appName, batchInterval, [sparkHome], [jars])

master参数是一个标准的Spark集群URL，并且可以是“local”作为本地测试。appName是你的程序的名字，它将会在你的集群的Web ＵＩ中显示。 batchInterval是batches的大小，就像之前解释的。最后，如果运行为分布式模式，需要最后两个参数来部署你的代码到一个集群上，就像Spark programming guide描述的那样。此外，基本的SparkContext可以如同ssc.sparkContext这样访问。

batch internal的设置必须根据你的应用程序的延迟要求和可用的集群资源。查看Performance Tuning获得更对详细信息。

###DStreams### Discretized Stream或者说DStream，是Spark Streaming提供的基本的抽象。它表示连续不断的数据流，或者来自数据源的输入数据流，或者通过转换输入流生成的经过处理的数据流。在内部，它通过一个连续不断的RDDs的序列表示，他是Spark的一个不可变得抽象，分布式数据器。Each RDD in a DStream contains data from a certain interval，就像下面的图表中展示的：

应用在一个DStream上的任何操作转换成在基础的RDDs上面的操作。例如， in the earlier example of converting a stream of lines to words, the flatmap operation is applied on each RDD in the lines DStream to generate the RDDs of the words DStream.下面的图表展示了这个：

这些基础的RDD转换是通过Spark引擎计算的。DStream操作隐藏了大多数的细节并提供开发者方便的高级API。这些操作在后面的章节中有详细讨论。

###输入源### 我们已经在[ quick example]( quick example)看了ssc.socketTextStream(...),它通过一个TCP套接字连接接受文本数据创建了一个DStream。除了套接字，核心Spark Streaming API提供了创建DStream通过文件，和将Akka actors作为输入源。

特别的，对于文件，DStream可以这样创建：

jssc.fileStream(dataDirectory);

Spark Streaming将会监视dataDirectory目录下的任何Hadoop兼容的文件系统，并且处理这个目录下创建的任何文件。

注意：

文件必须有统一的格式
The files must be created in the dataDirectory by atomically moving or renaming them into the data directory.
Once moved the files must not be changed.

For more details on streams from files, Akka actors and sockets, see the API documentations of the relevant functions in StreamingContext for Scala and JavaStreamingContext for Java.

此外，通过源，例如Kafka、Flume和Twitter创建DStream的功能可以通过导入并添加正确的依赖，就像前面的章节中解释的那样。在Kafka的情况下，在添加artifact spark-streaming-kafka_2.10到项目的依赖后，你可以像这样创建一个来自Kafka的DStream：

import org.apache.spark.streaming.kafka.*;
KafkaUtils.createStream(jssc, kafkaParams, ...);

更多关于附加源的细节，查看相应的API文档，此外，你可以实现你自己的源的定制接收者，查看Custom Receiver Guide.

###操作### 有两种DStream操作-转换和输出操作。和RDD转换类似，DStream转换操作针对一个或者多个DStream来创建新的包含转换数据的DStreams。在数据流上应用一系列转换后，输入操作需要调用，它写数据到一个额外的数据槽中，例如一个文件系统或者一个数据库。

####转换#### DStream支持许多转换，在一个普通的Spark RDD上。下面是一些常见的转换：

<pre> Transformation Meaning map(func) Return a new DStream by passing each element of the source DStream through a function func. flatMap(func) Similar to map, but each input item can be mapped to 0 or more output items. filter(func) Return a new DStream by selecting only the records of the source DStream on which func returns true. repartition(numPartitions) Changes the level of parallelism in this DStream by creating more or fewer partitions. union(otherStream) Return a new DStream that contains the union of the elements in the source DStream and otherDStream. count() Return a new DStream of single-element RDDs by counting the number of elements in each RDD of the source DStream. reduce(func) Return a new DStream of single-element RDDs by aggregating the elements in each RDD of the source DStream using a function func (which takes two arguments and returns one). The function should be associative so that it can be computed in parallel. countByValue() When called on a DStream of elements of type K, return a new DStream of (K, Long) pairs where the value of each key is its frequency in each RDD of the source DStream. reduceByKey(func, [numTasks]) When called on a DStream of (K, V) pairs, return a new DStream of (K, V) pairs where the values for each key are aggregated using the given reduce function. Note: By default, this uses Spark's default number of parallel tasks (2 for local mode, and in cluster mode the number is determined by the config property spark.default.parallelism) to do the grouping. You can pass an optional numTasks argument to set a different number of tasks. join(otherStream, [numTasks]) When called on two DStreams of (K, V) and (K, W) pairs, return a new DStream of (K, (V, W)) pairs with all pairs of elements for each key. cogroup(otherStream, [numTasks]) When called on DStream of (K, V) and (K, W) pairs, return a new DStream of (K, Seq[V], Seq[W]) tuples. transform(func) Return a new DStream by applying a RDD-to-RDD function to every RDD of the source DStream. This can be used to do arbitrary RDD operations on the DStream. updateStateByKey(func) Return a new "state" DStream where the state for each key is updated by applying the given function on the previous state of the key and the new values for the key. This can be used to maintain arbitrary state data for each key. </pre>

最后两个转换值得再次解释。

####UpdateStateByKey操作#### updateStateByKey允许你维护任意的状态，同时，可以持续不断的更新新信息。使用它，你需要下面两步：

定义状态-状态可以是任意数据类型
定义状态更新函数-指定一个函数，怎样从之前的状态和新的输入流的值中更新状态

让我们使用一个例子阐述。假设我们想维护一个连续的一个文本流中的单词出现的次数。这里，连续的和是这个state，并且是一个Integer，我们定义update函数，像这样：

import com.google.common.base.Optional;
Function2<List<Integer>, Optional<Integer>, Optional<Integer>> updateFunction =
  new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() {
    @Override public Optional<Integer> call(List<Integer> values, Optional<Integer> state) {
      Integer newSum = ...  // add the new values with the previous running count to get the new count
      return Optional.of(newSum);
    }
  };

下面的应用在一个包含words的DStream上（假设，Pairs DStream包含（word ，1）对在quick example)

update函数将会被每一个word调用，with newValues having a sequence of 1’s (from the (word, 1) pairs) and the runningCount having the previous count.完整的Scala代码，查看例子StatefulNetworkWordCount.

####Transform操作####

####Window操作#### 最后，Spark Streaming还提供了window计算。

####Output操作#### 当一个输出操作被调用，它出发一个流计算，目前，定义了下面的输出操作：

<pre> Output Operation Meaning print() Prints first ten elements of every batch of data in a DStream on the driver. foreachRDD(func) The fundamental output operator. Applies a function, func, to each RDD generated from the stream. This function should have side effects, such as printing output, saving the RDD to external files, or writing it over the network to an external system. saveAsObjectFiles(prefix, [suffix]) Save this DStream's contents as a SequenceFile of serialized objects. The file name at each batch interval is generated based on prefix and suffix: "prefix-TIME_IN_MS[.suffix]". saveAsTextFiles(prefix, [suffix]) Save this DStream's contents as a text files. The file name at each batch interval is generated based on prefix and suffix: "prefix-TIME_IN_MS[.suffix]". saveAsHadoopFiles(prefix, [suffix]) Save this DStream's contents as a Hadoop file. The file name at each batch interval is generated based on prefix and suffix: "prefix-TIME_IN_MS[.suffix]". </pre>

完整的DStream操作的列表可以在API文档得到。对于Scala API，查看DStream和 PairDStreamFunctions,对于Java API，查看JavaDStream和 JavaPairDStream.

###持久化### 类似于RDDs，DStreams同样允许开发者持久化流数据到内存。就是说，在一个DStream上使用persist()将会自动的持久化这个DStream的每一个RDD到内存。如果这个DStream中的数据将会被计算多次（例如，在同样的数据上进行多个操作），这是非常有用的。对于基于window的操作例如reduceByWondow和reduceByKeyAndWindow和基于状态的操作例如updateStateByKey，是默认持久化的。因此，通过基于window的操作生成的DStream是自动持久化到内存的，而不需要开发者去调用persist()方法。

对于数据流来说，它通过network（例如Kafka，Flume，socket等等）接收数据，它的默认的持久化级别是复制数据到两个节点，以便容错。

注意，不想RDDs，DSteam默认的持久化级别是保持数据在内存中序列化。在章节Performance Tuning有更多的讨论。更多关于不同持久化级别的信息可以在 Spark Programming Guide找到。

###RDD Checkpointing### 一个stateful操作是那些在数据的多个batches上的操作。它包括所有基于window的操作和updateStateByKey操作。由于stateful操作依赖于之前数据的batches，他们随着时间连续不断的聚集元数据。要清除这些数据，Spark Streaming支持在存储中间数据到HDFS时进行定期的checkpointing。

启用checkpointing，开发者需要提供RDD将被保存的HDFS路径。通过以下代码完成：

ssc.checkpoint(hdfsPath) // assuming ssc is the StreamingContext or JavaStreamingContext

一个DStream的checkpointing的间隔可以这样设置：

dstream.checkpoint(checkpointInterval)

对于DStream，他必须被checkpointing（即，DStream通过updateStateByKey创建，并且使用相反的函数reduceByKeyAndWindow），DStream的checkpoint间隔默认设置为set to a multiple of the DStream’s sliding interval，例如至少设置10秒。

###Deployment### 和其他任何Spark应用程序一样，Spark Streaming应用程序部署在集群上。请参考 deployment guide获得更多信息。

如果一个正在运行的Spark Streaming应用程序需要升级(包括新的应用代码)，这里有两个可能的技巧：

The upgraded Spark Streaming application is started and run in parallel to the existing application. Once the new one (receiving the same data as the old one) has been warmed up and ready for prime time, the old one be can be brought down. Note that this can be done for data sources that support sending the data to two destinations (i.e., the earlier and upgraded applications).
The existing application is shutdown gracefully (see StreamingContext.stop(...) or JavaStreamingContext.stop(...) for graceful shutdown options) which ensure data that have been received is completely processed before shutdown. Then the upgraded application can be started, which will start processing from the same point where the earlier application left off. Note that this can be done only with input sources that support source-side buffering (like Kafka, and Flume) as data needs to be buffered while the previous application down and the upgraded application is not yet up.

到此，相信大家对“Spark Streaming编程技巧是什么”有了更深的了解，不妨来实际操作一番吧！这里是亿速云网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！

向AI问一下细节

Spark Streaming编程技巧是什么

猜你喜欢

最新资讯

相关推荐

相关标签