SparkStreaming教程

概要

Spark流是对于Spark核心API的拓展，从而支持对于实时数据流的可拓展，高吞吐量和容错性流处理。数据可以由多个源取得，例如：Kafka，Flume，Twitter，ZeroMQ，Kinesis或者TCP接口，同时可以使用由如map，reduce，join和window这样的高层接口描述的复杂算法进行处理。最终，处理过的数据可以被推送到文件系统，数据库和HDFS。

image

在内部，其按如下方式运行。Spark Streaming接收到实时数据流同时将其划分为分批，这些数据的分批将会被Spark的引擎所处理从而生成同样按批次形式的最终流。

image

Spark Streaming提供了被称为离散化流或者DStream的高层抽象，这个高层抽象用于表示数据的连续流。
创建DStream的两种方式：
1. 由Kafka，Flume取得的数据作为输入数据流。
2. 在其他DStream进行的高层操作。
在内部，DStream被表达为RDDs的一个序列。
这个指南会指引你如何利用DStreams编写Spark Streaming的程序。你可以使用诸如Scala，Java或者Python来编写Spark Streaming的程序。文中的标签可以让你在不同编程语言间切换。
注意：少量的API在Python中要么是不可用的，要么是和其他有差异的。在本文中，这些点将会被高亮显示。
为方便起见，Spark Streaming 直接缩写为SS形式。

一个简单的例子

在深入了解如何编写你自己的SS程序之前，让我们先迅速浏览下基本的SS程序是什么样的。假设我们想统计文本数据中单词个数(数据来自于监听一个TCP接口的数据服务器)。你只需要这样做：
第一步，加载入StreamingContext，这个是所有流功能函数的主要访问点，我们使用两个执行线程和1s的批次间隔来创建本地的StreamingContext：

maven

<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <spark.version>2.2.1</spark.version>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
  </properties>

  <dependencies>
    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>4.11</version>
      <scope>test</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-core_2.11</artifactId>
      <version>${spark.version}</version>
    </dependency>

    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-streaming_2.11</artifactId>
      <version>${spark.version}</version>
    </dependency>

    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-streaming-kafka_2.11</artifactId>
      <version>1.6.2</version>
    </dependency>

    <dependency>
      <groupId>org.scala-lang</groupId>
      <artifactId>scala-library</artifactId>
      <version>2.11.8</version>
    </dependency>

  </dependencies>

package com.hx.test

/**
 * fileName : Test11StreamingWordCount 
 * Created by 970655147 on 2016-02-12 13:21.
 */
object Test11StreamingWordCount {

  // 基于sparkStreaming的wordCount
  // 环境windows7 + spark1.2 + jdk1.7 + scala2.10.4
  // 1\. 启动netcat [nc -l -p 9999]
  // 2\. 启动当前程序
  // 3\. netcat命令行中输入数据
  // 4\. 回到console, 查看结果[10s 之内]
  // *******************************************
  // 每一个print() 打印一次
  // -------------------------------------------
  // Time: 1455278620000 ms
  // -------------------------------------------
  // Another Infomation !
  // *******************************************
  // inputText : sdf sdf lkj lkj lkj lkj
  // MappedRDD[23] at count at Test11StreamingWordCount.scala:39
  // 2
  // (sdf,2), (lkj,4)
  def main(args :Array[String]) = {

    // Create a StreamingContext with a local master
    // Spark Streaming needs at least two working thread
    val sc = new StreamingContext("local[2]", "NetworkWordCount", Seconds(10) )
    // Create a DStream that will connect to serverIP:serverPort, like localhost:9999
    val lines = sc.socketTextStream("192.168.47.141", 9999)
     // Split each line into words
    // 以空格把收到的每一行数据分割成单词
    val words = lines.flatMap(_.split(" "))
    // 在本批次内计单词的数目
    val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)
    // 打印每个RDD中的前10个元素到控制台
    wordCounts.print()

    sc.start()
    sc.awaitTermination()

  }

}

words DStream 进一步被映射成(一对一的转换)(word,1)对的DStream形式，这个“对”形式的DStream将会被reduced（一个Spark操作）以取得数据各个批次中单词的统计。最后，wordCounts.print()打印处每一秒所获得的少量计数值。

注意：这么多行代码被执行后，SS仅仅设置了其若开始运行将要进行的运算，但是并没有开始真正意义上的处理。在所有的转换都部署完毕后，我们需要调用下面两个操作来真正启动处理：

ssc.start() # 开始计算
ssc.awaitTermination() # 等待计算终止

如果你已经下载并编译了Spark，就可以按如下讲解来运行这个例子。首先你需要运行Netcat(大多数类Unix系统都有的工具)作为数据服务器：

$ nc -l 9999

运行netcat终端上的任何键入的行将会被计算并打印到屏幕上。

# TERMINAL 1:
# Running Netcat
$ nc -l 9999
hello world

# TERMINAL 2: RUNNING network_wordcount.py
$ ./bin/spark-submit examples/src/main/python/streaming/network_wordcount.py localhost 9999
(hello,1)
(world,1)

基本概念

下来，我们超越简单历程的局限，阐述SS的基本知识。

Linking

和Spark相似，SS通过Maven中心也可用。为了编写你自己的SS程序，你需要将下面的依赖加进你的SBT或者Maven项目。

# Maven:
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming_2.10</artifactId>
    <version>1.6.1</version>
</dependency>

为了从像Kafka，Flume这些没有出现在SS核心API的源来摄取数据，你需要将相应的artifact: spark-streaming-xyz_2.10 加进依赖中。下面列举一些常见的：

Source  Artifact
Kafka   spark-streaming-kafka_2.10
Flume   spark-streaming-flume_2.10
Kinesis spark-streaming-kinesis-asl_2.10
Twitter spark-streaming-twitter_2.10
ZeroMQ  spark-streaming-zeromq_2.10
MQTT    spark-streaming-mqtt_2.10 

对于最新的列表，请参阅 Maven仓库 以获得完整的支持源和artifacts。

初始化StreamingContext

为初始化一个SS程序，必须创建一个StreaingContext对象，此对象为所有Spark Streaming功能函数的主要入口。
1. 可以由一个SparkContext对象来创建一个StreamingContext对象

# appName: 在集群UI上显示的应用标签。
# master: 一个 Spark, Mesos, or YARN集群URL, or 一个特定的
#   在"loacl[*]"字符串以运行于本地。
sc = SparkContext(master, appName)
ssc = StreamingContext(sc, 1)

在实际中，运行在集群上时，你或许不希望在程序中将 master 写死，而是使用 spark-submit来发射应用并在这里接收。但是，为了本地测试和单元测试，你可以写入”local[*]”来在进程中启动SS(监测本地系统中核的数量)。
在定义了一个context后，你需要做：
1. 通过创建DStream来定义输入源
2. 通过对DStream使用转换和输出操作来定义流计算
3. 使用streamingContext.start()来接收并处理数据
4. 使用streamingContext.awaitTermination()等待处理的停止(手动或者因为任何出错).
5. 处理进程可以使用streamingContext.stop()来手动停止。

离散数据流(DStreams)

离散数据流或者DStream是SS提供的基本抽象。其表现数据的连续流，这个输入数据流可以来自于源，也可以来自于转换输入流产生的已处理数据流。内部而言，一个DStream以一系列连续的RDDs所展现，这些RDD是Spark对于不变的，分布式数据集的抽象。一个DStream中的每个RDD都包含来自一定间隔的数据，如下图：

image

在DStream上使用的任何操作都会转换为针对底层RDD的操作。例如：之前那个将行的流转变为词流的例子中，flatMap操作应用于行DStream的每个RDD上 从而产生words DStream的RDD。如下图：

image

这些底层的RDD转换是通过Spark引擎计算的。DStream操作隐藏了大多数细节，同时为了方便为开发者提供了一个高层的API。这里的一些操作会在下文中详述。

输入DStream和接收器

输入DStream是用于表示来自于源流的输入数据流的。在上边讲到的例子中，lines就是这样一个输入DStream，因为其表示了来自于netcat服务器的数据流。除了文件流（下文中会讲到）之外的所有输入流都会和接收器对象(Scala文档，Java文档)相关联，其接收来自源的数据并为了之后的处理将数据保存在Spark的内存中。
SS提供了两种类型的内置流源：
1. 基本源：StreamingContext API直接可用的源。例如：文件系统，套接字连接。
2. 高级源：类似Kafka，Flume等等这样的源可以通过额外的工具类来使用。这要求链接到额外的依赖(如linking节所讲)。

基本源

如上讲解，我们已经通过一个简单的例子了解了如何从文本数据来创建一个DStream（文本数据来自于TCP套接字的连接）。除了套接字，StreamingContext API也提供了以文件作为输入源来创建DStream的途径。
1. 文件流：从兼容HDFS API的任何文件系统文件中读取的数据，可按如下方式创建DStream：

streamingContext.textFileStream(dataDirectory)

SS将监听目录”dataDirectory”，同时处理任何创建在此目录中的文件（不包括此目录下嵌套目录中的文件）。
注意：
1. 文件中数据格式必须一致。
2. 文件必须以原子方式移动或者重命名进入目录的方式来创建。
3. 一旦移动完毕，文件就不能在改动了。所以即使继续加入文件内容，新加入数据也不会被读取。

高级源

这一类源需要非Spark库的外部接口，其中一些需要复杂的依赖（如Kafka和Flume）。因此，为了尽可能减少版本之间的依赖冲突问题，从这些源中创建DStreams的功能函数被移至单独的库从而可以在需要的时候显式链接。

注意：这些高级源不能在Spark shell上使用，因此基于这些源的应用程序不能在shell上测试。

对DStream的转换

和RDD一样，使用转换可以修改从输入DStream获取的数据。DStreams支持许多用在一般Spark RDD上的转换。其中一些常用的如下：
map(func)：将源DStream中的每个元素通过一个函数func从而得到新的DStreams。
flatMap(func)：和map类似，但是每个输入的项可以被映射为0或更多项。
filter(func)：选择源DStream中函数func判为true的记录作为新DStreams
repartition(numPartitions)：通过创建更多或者更少的partition来改变此DStream的并行级别。
union(otherStream)：联合源DStreams和其他DStreams来得到新DStream
count：统计源DStreams中每个RDD所含元素的个数得到单元素RDD的新DStreams。
reduce(func)：通过函数func(两个参数一个输出)来整合源DStreams中每个RDD元素得到单元素RDD的DStreams。这个函数需要关联从而可以被并行计算。
countByValue：对于DStreams中元素类型为K调用此函数，得到包含(K,Long)对的新DStream，其中Long值表明相应的K在源DStream中每个RDD出现的频率。
reduceByKey(func, [numTasks])：对(K,V)对的DStream调用此函数，返回同样（K,V)对的新DStream，但是新DStream中的对应V为使用reduce函数整合而来。Note：默认情况下，这个操作使用Spark默认数量的并行任务（本地模式为2，集群模式中的数量取决于配置参数spark.default.parallelism）。你也可以传入可选的参数numTaska来设置不同数量的任务。
join(otherStream,[numTasks])：两DStream分别为(K,V)和(K,W)对，返回(K,(V,W))对的新DStream。
cogroup(otherStream,[numTasks])：两DStream分别为(K,V)和(K,W)对，返回(K,(Seq[V],Seq[W])对新DStreams
transform(func)：将RDD到RDD映射的函数func作用于源DStream中每个RDD上得到新DStream。这个可用于在DStream的RDD上做任意操作。
updateStateByKey(func)：得到”状态”DStream，其中每个key状态的更新是通过将给定函数用于此key的上一个状态和新值而得到。这个可用于保存每个key值的任意状态数据。
这其中的一些转换操作值得详细讨论。

UpdateStateByKey

updateStateByKey操作使得我们可以在用新信息进行更新时保持任意的状态。为使用这个功能，你需要做下面两步：
1. 定义状态，状态可以是一个任意的数据类型。
2. 定义状态更新函数，用此函数阐明如何使用之前的状态和来自输入流的新值对状态进行更新。

注意，updateStateByKey操作，要求必须开启Checkpoint机制。且不能是本地文件系统，必须是HDFS

import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2018/7/24.
  */
object WordCount {
  def main(args:Array[String]){
    val conf=new SparkConf().setAppName("updateStateByKeyPro")
      .setMaster("local[2]")//spark://hadoop:7077
    val ssc=new StreamingContext(conf,Seconds(10))

    //开启checkpoint
    ssc.checkpoint("hdfs://192.168.47.244:8020/input")

    //连接nc(netcat)服务,接收数据源，产生Dtream 对象
    val lines=ssc.socketTextStream("192.168.47.141",9999)

    //分隔单词，并将分隔后的每个单词出现次数记录为１
    val pairs=lines.flatMap(_.split(" "))
      .map(word=>(word,1))
    //调用updateStateByKey算子，统计单词在全局中出现的次数
    val result=pairs.updateStateByKey((values:Seq[Int],state:Option[Int])=>{
      //创建一个变量，用于记录单词出现次数
      var newValue=state.getOrElse(0) //getOrElse相当于if....else.....
      for(value <- values){
        newValue +=value //将单词出现次数累计相加
      }
      Option(newValue)
    })
    //直接输出结果
    result.print()

    ssc.start() //开启实时计算
    ssc.awaitTermination() //等待应用停止
  }

}

基于DStream的输出操作

输出操作使得DStream数据可以被推送到外部的系统，如：数据库或者文件系统。因为输出操作的确允许外部系统来使用转换的数据，故而其触发所有DStream转换的真实执行(对于RDD同样如此)。现在，下边是定义的输出操作：
print()：在运行流程序的驱动结点上打印DStream中每一批次数据的最开始10个元素。这用于开发和调试。在Python API中，同样的操作叫pprint()。
saveAsTextFiles(prefix, [suffix])：以text文件形式存储这个DStream的内容。每一批次的存储文件名基于参数中的prefix和suffix。”prefix-Time_IN_MS[.suffix]”.
saveAsObjectFiles(prefix, [suffix])：Python API中不可用。译者是写Python的，所以..。
saveAsHadoopFiles(prefix, [suffix])：同上。
foreachRDD(func)：这是最通用的输出操作，即将函数func用于产生于stream的每一个RDD。其中参数传入的函数func应该实现将每一个RDD中数据推送到外部系统，如将RDD存入文件或者通过网络将其写入数据库。

使用foreachRDD的设计模式

dstream.foreachRDD是强大的原始操作，其允许数据被传送到外部系统。然而，理解如何正确有效地使用这个原始显得至关重要。需要避开一些如下所示的常见错误。
通常将数据写入外部系统需要创建一个连接对象(如连接到远程服务器TCP)，同时用此连接对象将数据传送到远程系统。为达到这个目的，开发者会尝试在Spark驱动上创建一个连接，然后在Spark工作结点使用从而存储RDD中的记录。例如(在Scala)：

def sendRecord(rdd):
    connection = createNewConnection() # executed at the driver
    rdd.foreach(lambda record: connection.send(record)
    connection.close()

dstream.foreachRDD(sendRecord)

上述代码是不对的，因为这需要连接对象序列化再由驱动送至工作结点。这样的连接对象极少在机器之间转移。这个问题会表现为序列化错误（连接对象不能序列化），初始化错误（连接对象需要在工作结点初始化）等等。正确的解决方法是在工作结点创建一个连接对象。
然而，这又会引出另一个常见的问题 - 为每个记录创建一个新的连接。例如：
一个RDD里面包含多条数据

def sendRecord(record):
    connection = createNewConnection()
    connection.send(record)
    connection.close()
dstream.foreachRDD(lambda rdd: rdd.foreach(sendRecord))

通常，创建一个连接对象有时间和资源上的花费。因此，为每一个record都创建和销毁一个连接对象会引起不必要的高花费，同时会显著减少整个系统的吞吐量。更好的解决方案是使用rdd.foreachPartition - 创建一个单独的连接对象并使用此连接为一个RDD分区中所有records做传送工作。<即针对分区创建连接>

def sendPartition(iter):
    connection = createNewConnection()
    for record in iter:
        connection.send(record)
    connection.close()

dstream.foreachRDD(lambda rdd: rdd.foreachPartition(sendPartition))

为分区创建连接就避免了为records创建带来的高创建量。
最后，可以在多RDDs/批次间重复使用连接对象来进行进一步的优化。使用者可以创建一个连接对象状态池，从而在多batch里的RDD被推送到外部系统后重复使用，这个可以进一步减少开销。

def sendPartition(iter):
    # ConnectionPool is a static, lazily initialized pool of connections
    connection = ConnectionPool.getConnection()
    for record in iter:
        connection.send(record)
    # return to the pool for future reuse
    ConnectionPool.returnConnection(connection)
dstream.foreachRDD(lambda rdd: rdd.foreachPartition(sendPartition))

注意： 池中的连接应该按需懒创建并在一段时间无用后超时。这个操作高效地实现了将数据传送到外部系统。
其他需要记住的点
DStreams被输出操作延迟执行，同样的思想体现在RDD操作对RDDs的延迟执行。具体而言，DStream中的RDD操作输出操作迫使对接收数据的处理。因此，如果你的应用没有任何输出操作或者像dstream.foreachRDD()这样的其中没有任何RDD操作的输出操作，那么没有什么会执行。这个系统将简单地接收数据之后丢弃。

总而言之，dstream.foreachRDD()循环的使一个dstream中的所有rdd（一个批次一个rdd），每个rdd可能有几个分区Partition，每个分区又是一个记录的集合，每个分区创建一个数据库连接，写入该分区的数据。

访问MySql

import java.sql.{PreparedStatement, Connection, DriverManager}
import java.util.concurrent.atomic.AtomicInteger
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.storage.StorageLevel

object NetworkWordCountStateful {
  def main(args: Array[String]): Unit = {
    //定义状态更新函数
    val updateFunc = (values: Seq[Int], state: Option[Int]) => {
      val currentCount = values.foldLeft(0)(_ + _)
      val previousCount = state.getOrElse(0)
      Some(currentCount + previousCount)
    }
    // Create a StreamingContext with a local master
    // Spark Streaming needs at least two working thread
    val sc = new StreamingContext("local[2]", "NetworkWordCount", Seconds(10) )
    // Create a DStream that will connect to serverIP:serverPort, like localhost:9999
    val lines = sc.socketTextStream("192.168.47.245", 9999)
     sc.checkpoint("hdfs://192.168.47.244:8020/input")
 //设置检查点，检查点具有容错机制

    val words = lines.flatMap(_.split(" "))
    val wordDstream = words.map(x => (x, 1))
    val stateDstream = wordDstream.updateStateByKey[Int](updateFunc)
    stateDstream.print()

    //下面是新增的语句，把DStream保存到MySQL数据库中
    stateDstream.foreachRDD(rdd => {
      //内部函数
      def func(records: Iterator[(String,Int)]) {
        var conn: Connection = null
        var stmt: PreparedStatement = null
        try {
          val url = "jdbc:mysql://hadoop01:3306/storm"
          val user = "root"
          val password = "root"  //笔者设置的数据库密码是hadoop，请改成你自己的mysql数据库密码
          conn = DriverManager.getConnection(url, user, password)
          records.foreach(p => {
            val sql = "insert into wordcount(word,count) values (?,?)"
            stmt = conn.prepareStatement(sql);
            stmt.setString(1, p._1.trim)
            stmt.setInt(2,p._2.toInt)
            stmt.executeUpdate()
          })
        } catch {
          case e: Exception => e.printStackTrace()
        } finally {
          if (stmt != null) {
            stmt.close()
          }
          if (conn != null) {
            conn.close()
          }
        }
      }

      val repartitionedRDD = rdd.repartition(3)
      repartitionedRDD.foreachPartition(func)
    })

    sc.start()
    sc.awaitTermination()
  }

}

一个RDD里面包含多条数据

消费kafka数据

 <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka_2.11</artifactId>
            <version>1.6.2</version>
        </dependency>

        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>2.11.8</version>
        </dependency>

package com.neusoft

import kafka.serializer.StringDecoder
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._
import org.apache.spark.SparkConf

object WordCount {
  def main(args: Array[String]) {
    //定义状态更新函数
    val updateFunc = (values: Seq[Int], state: Option[Int]) => {
      val currentCount = values.foldLeft(0)(_ + _)
      val previousCount = state.getOrElse(0)
      Some(currentCount + previousCount)
    }

    val sparkConf = new SparkConf().setAppName("DirectKafkaWordCount").setMaster("local[2]")
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    ssc.checkpoint("hdfs://192.168.121.128:8020/demo1")
    //kafka的topic集合，即可以订阅多个topic，args传参的时候用,隔开
    val topicsSet = Set("test")
    //设置kafka参数，定义brokers集合
    val kafkaParams = Map[String, String]("metadata.broker.list" -> "192.168.121.128:9092")
    val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
      ssc, kafkaParams, topicsSet)
    print("---------:" +messages)

    val lines = messages.map(_._2)
    val words = lines.flatMap(_.split(" "))
//    val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)
//    wordCounts.print()

    val stateDstream = words.map(x => (x, 1)).updateStateByKey[Int](updateFunc)
    stateDstream.print()

    ssc.start()
    ssc.awaitTermination()
  }
}