Spark Streaming是什么

132次阅读

没有评论

共计 5320 个字符，预计需要花费 14 分钟才能阅读完成。

这篇文章主要讲解了“Spark Streaming 是什么”，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着丸趣 TV 小编的思路慢慢深入，一起来研究和学习“Spark Streaming 是什么”吧！

一：Spark Streaming 概览。

1.1 简单了解 Spark Streaming。

Spark Streaming 是核心 Spark API 的一个扩展。具有可扩展性，高吞吐量，容错性，实时性等特征。

数据从许多来如中摄入数据，如 Kafka, Flume, Twitter, ZeroMQ, Kinesis, or TCP sockets。

也可以使用复杂的算法与高级别的功能像 map，reduce，join 和 window 处理。

最后，也可以将处理过的数据推送到文件系统、数据库。事实上，我们也可以用 Spark 的机器学习和图形处理数据流上的算法。用图形表示如下：
在内部，其工作原理如下。Spark Streaming 接收实时输入的数据流和数据划分批次，然后由 Spark 引擎批处理生成的最终结果流。如图示：

另外，Spark Streaming 提供一个高级抽象，称为离散的流或 DStream，表示连续的流的数据。DStreams 可以被创建从输入的数据流，如 Kafka, Flume, and Kinesis,

或采用其他的 DStreams 高级别操作的输入的数据流。

在内部，DStream 是以 RDDs 的序列来表示。

首先，看看 Maven 的依赖包（spark-streaming_2.10）管理：

  dependency 
 groupId org.apache.spark /groupId 
 artifactId spark-streaming_2.10 /artifactId 
 version 1.6.1 /version 
 /dependency

1.2 eg：从一个数据服务器监听 TCP 套接字接收的文本数据中的单词进行计数

package com.berg.spark.test5.streaming;
import java.util.Arrays;
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.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;
public class SparkStreamingDemo1 {public static void main(String[] args) {SparkConf conf = new SparkConf().setMaster(local[2] ).setAppName( NetworkWordCount 
 conf.set( spark.testing.memory ,  269522560000 
 JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(10));
 System.out.println(jssc:   + jssc);
 //  创建一个 DStream,  将连接  hostname:port,  比如  master:9999
 JavaReceiverInputDStream String  lines = jssc.socketTextStream(master , 9999);
 System.out.println(lines :   + lines);
 JavaDStream String  words = lines.flatMap(new FlatMapFunction String, String () {
 private static final long serialVersionUID = 1L;
 public Iterable String  call(String x) {return Arrays.asList(x.split(  ));
 // Count each word in each batch
 JavaPairDStream String, Integer  pairs = words.mapToPair(new PairFunction String, String, Integer () {public Tuple2 String, Integer  call(String s) {return new Tuple2 String, Integer (s, 1);
 JavaPairDStream String, Integer  wordCounts = pairs.reduceByKey(new Function2 Integer, Integer, Integer () {public Integer call(Integer i1, Integer i2) {
 return i1 + i2;
 // Print the first ten elements of each RDD generated in this DStream to
 // the console
 wordCounts.print();
 jssc.start(); // Start the computation
 jssc.awaitTermination(); // Wait for the computation to terminate}

至于代码如何运行了，首先在 Linux 下终端输入：$ nc -lk 9999

然后在 Eclipse 中运行代码。

随意输入一行文本单词，单词之间用空格隔开，如下：

hadoop@master:~$ nc -lk 9999
berg hello world berg hello

可以在 Eclipse 控制台看到如下结果：

Time: 1465386060000 ms
-------------------------------------------
(hello,2)
(berg,2)
(world,1)

1.3 将 HDFS 目录下的某些文件内容当做输入的数据流。

public class SparkStreamingDemo2 {public static void main(String[] args) {SparkConf conf = new SparkConf().setMaster(local[2] ).setAppName( NetworkWordCount 
 conf.set( spark.testing.memory ,  269522560000 
 JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(10));
 System.out.println(jssc:   + jssc);
 //  创建一个 DStream,  读取 HDFS 上的文件, 作为数据源。JavaDStream String  lines = jssc.textFileStream( hdfs://master:9000/txt/sparkstreaming/ 
 System.out.println(lines :   + lines);
 // Split each line into words
 JavaDStream String  words = lines.flatMap(new FlatMapFunction String, String () {
 private static final long serialVersionUID = 1L;
 public Iterable String  call(String x) {System.out.println(Arrays.asList(x.split(  )).get(0));
 return Arrays.asList(x.split(  ));
 // Count each word in each batch
 JavaPairDStream String, Integer  pairs = words.mapToPair(new PairFunction String, String, Integer () {
 private static final long serialVersionUID = 1L;
 public Tuple2 String, Integer  call(String s) {return new Tuple2 String, Integer (s, 1);
 System.out.println(pairs);
 JavaPairDStream String, Integer  wordCounts = pairs.reduceByKey(new Function2 Integer, Integer, Integer () {public Integer call(Integer i1, Integer i2) {
 return i1 + i2;
 // Print the first ten elements of each RDD generated in this DStream to the console
 wordCounts.print();
 JavaDStream Long  count = wordCounts.count();
 count.print(); //  统计
 DStream Tuple2 String, Integer  dstream = wordCounts.dstream();
 dstream.saveAsTextFiles( hdfs://master:9000/bigdata/spark/xxxx ,  sparkstreaming 
 //wordCounts.dstream().saveAsTextFiles( hdfs://master:9000/bigdata/spark/xxxx ,  sparkstreaming 
 jssc.start(); 
 jssc.awaitTermination(); // Wait for the computation to terminate}

上述代码完成的操作是，一直监听 HDFS 即 hdfs://master:9000/txt/sparkstreaming/ 目录下是否有文件存入，如果有，则统计文件中的单词。。。。

尝试运行程序后，然后往该目录中手动添加一个文件，会在控制台看到对该文件内容中的单词统计后的数据。

注意参数的意义：

public JavaDStream java.lang.String textFileStream(java.lang.String directory)
Create an input stream that monitors a Hadoop-compatible filesystem for
new files and reads them as text
files (using key as LongWritable, value as Text and input format as TextInputFormat).
Files must be written to the monitored directory
by moving them from another location within the same file system.
File names starting with . are ignored.
Parameters:
directory – HDFS directory to monitor for new file
Returns:
(undocumented)

感谢各位的阅读，以上就是“Spark Streaming 是什么”的内容了，经过本文的学习后，相信大家对 Spark Streaming 是什么这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是丸趣 TV，丸趣 TV 小编将为大家推送更多相关知识点的文章，欢迎关注！

正文完