DStreams转换（Transformation）
和RDD类似，转换中允许输入DStream中的数据被修改。DStream支持很多Spark RDD上的转换。常用的转换如下。
| 转换 | 含义 |
|---|---|
| map(func) | 将源DStream中的每个元素传给函数func，返回新的DStream。 |
| flatMap(func) | 和map类似，但是每个输入条目可以映射到0或多个输出条目。 |
| filter(func) | 选择源DStream中经过func处理后返回true的元素，返回新的DStream。 |
| repartition(numPartitions) | 改变DStream的并行级别，可创建更多或者更少的分区。 |
| union(otherStream) | 返回新的DStream，包含源DStream和otherDStream元素的union。 |
| count() | 通过对每个RDD的元素计数返回单元素RDDs的DStream。 |
| reduce(func) | 使用函数func（两个参数和一个返回值），通过对每个RDD元素进行聚合返回单元素RDDs的DStream。函数应该是可结合和可交换的，以便进行并行计算。 |
| countByValue() | 在元素类型为K的DStream上调用时，返回一个(K, Long)对的DStream，每个key的value是每个RDD中这个key出现的频率。 |
| reduceByKey(func, [numTasks]) | 在(K, V)对的DStream上调用时，返回一个新(K, V)对的DStream，每个key的value是使用给定reduce函数进行聚合的结果。注意：默认情况，使用Spark的默认并行任务数量（本地模式为2，集群模式数量由spark.default.parallelism配置项决定）进行分组。可以传递可选参数numTasks任务数量。 |
| join(otherStream, [numTasks]) | 当在(K, V)对的DStream和(K, W)对的DStream上调用时，返回(K, (V, W))对的DStream。 |
| cogroup(otherStream, [numTasks]) | 当在(K, V)对的DStream和(K, W)对的DStream上调用时，返回(K, Seq[V], Seq[W])元组的DStream。 |
| transform(func) | 对源DStream的每个RDD应用一个RDD-to-RDD的函数，返回一个新DStream。可用于在DStream上进行任意RDD操作。 |
| updateStateByKey(func) | 返回新的"state" DStream，通过对key的前一个state和新值应用给定方法更新每个key的state。可用于维护每个key的任意state数据。 |
其中几个转换需要详细说明。
UpdateStateByKey操作
当连续使用新信息更新state时，updateStateByKey操作允许用户维护任意状态。使用这个操作，必须包含以下两个步骤。
- 定义state - state可以是任意数据类型。
- 定义state更新函数 - 用一个函数指定如何使用前一个state和来自输入流的新值更新state。
在每个批次，Spark会对所有存在的key应用state更新函数，不管这些key是否有新数据。如果更新函数返回None，则key-value对会消除。
用一个例子说明。维护每个单词的运行计数，运行计数是一个state并且是个整数。定义更新函数如下：
def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
val newCount = ... // add the new values with the previous running count to get the new count
Some(newCount)
}
应用在包含单词对（pair DStream包含(word, 1)，参见示例）的DStream上。
val runningCounts = pairs.updateStateByKey[Int](updateFunction _)
更新函数会在每个单词上进行调用，newValues是1's的序列（来自(word, 1)对），runningCount是之前的计数。
注意，使用updateStateByKey要求配置检查点目录，之后详细讨论。
Transform操作
transform操作（以及其变化transformWith）可以在DStream上应用任意RDD-to-RDD函数。可用于使用任意没有暴露给DStream API的RDD操作。例如，对数据流中每个批次的数据和另一个数据集进行join的功能没有直接暴露给DStream API。但是，可以简单地使用transform完成。这增加了很多可能性。例如，通过对输入数据流和预先计算的垃圾信息（也可能是Spark生成的）进行join，完成实时数据清理，然后基于结果进行筛选。
val spamInfoRDD = ssc.sparkContext.newAPIHadoopRDD(...) // RDD containing spam information
val cleanedDStream = wordCounts.transform { rdd =>
rdd.join(spamInfoRDD).filter(...) // join data stream with spam information to do data cleaning
...
}
注意，在每个批次时间间隔中，提供的函数都会被调用。可用于完成随时间变化的RDD操作，也就是说，RDD操作，分区数量和广播变量等可以在批次之间改变。
Window操作
Spark Streaming也提供了windowed计算，可在一个滑动window数据上应用转换。下图进行说明。
image.png
如上图所示，每次在源DStream滑动window，落在window中的源RDDs会被合并和操作用于产生windowed DStream的RDDs。在这个例子中，操作应用在后面3个时间单位的数据上，以2个时间单位进行滑动。任何window操作都需要指定两个参数。
- window长度 - window持续时间（上图是3）。
- 滑动间隔 - window操作执行的间隔（上图是2）。
这两个参数必须是源DStream批时间间隔的倍数。
下面用一个示例说明window操作。扩展之前的示例，生成过去30s的单词计数，每10s一次。为实现这个功能，必须对pair DStream过去30s的数据应用reduceByKey。这里使用reduceByKeyAndWindow操作。
// Reduce last 30 seconds of data, every 10 seconds
val windowedWordCounts = pairs.reduceByKeyAndWindow((a:Int,b:Int) => (a + b), Seconds(30), Seconds(10))
一些通用的window操作如下。所有操作都需要两个参数，windowLength和slideInterval。
| 转换 | 含义 |
|---|---|
| window(windowLength, slideInterval) | 返回新的DStream，基于源DStream的windowed批次进行计算。 |
| countByWindow(windowLength, slideInterval) | 返回数据流中元素的滑动窗口计数。 |
| reduceByWindow(func, windowLength, slideInterval) | 返回单元素数据流，使用func函数，聚合滑动时间间隔的数据流元素进行创建。 |
| reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks]) | 当在(K, V)对DStream上调用时，返回一个新的(K, V)对DStream，在滑动window的批次上，使用给定reduce函数func通过聚合得到每个key的value。注意：默认情况下，使用Spark的默认并行任务数量（本地模式是2，集群模式根据配置属性spark.default.parallelism确定）。可传递可选参数numTasks设置不同的任务数量。 |
| reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks]) | 上面方法的另一个版本，每个window的reduce value使用前一个window的reduce values增量计算。通过reduce进入滑动window的数据完成这个操作，然后逆向reduce离开window的旧数据。举个例子，加减keys的数量作为window slides。但是，只适用于"可逆向reduce函数"，也就是说，有对应"逆向reduce"函数的那些reduce函数（即invFunc参数）。和上面的方法类似，reduce任务的数量可以通过可选参数进行配置。注意，使用这个操作必须启用检查点。 |
| countByValueAndWindow(windowLength, slideInterval, [numTasks]) | 当在(K, V)对DStream上调用时，返回一个新的(K, Long)对DStream，每个key的value是这个key在滑动window中的频次。和reduceByKeyAndWindow类似，reduce任务的数量可以通过可选参数进行配置。 |
Join操作
最后，值得强调的是，可以很容易地在Spark Streaming中执行不同类型的join。
Stream-stream joins
数据流可以很容易地与其它数据流进行join。
val stream1: DStream[String, String] = ...
val stream2: DStream[String, String] = ...
val joinedStream = stream1.join(stream2)
这里，在每个批时间间隔中，stream1生成的RDD都会和stream2生成deRDD进行join。也可以使用leftOuterJoin，rightOuterJoin和fullOuterJoin。另外，在流的window之间进行join非常有用。
val windowedStream1 = stream1.window(Seconds(20))
val windowedStream2 = stream2.window(Minutes(1))
val joinedStream = windowedStream1.join(windowedStream2)
Stream-dataset joins
之前解释DStream.transform操作时进行了说明。这里是另外一个windowed数据流和数据集之间进行join的示例。
val dataset: RDD[String, String] = ...
val windowedStream = stream.window(Seconds(20))...
val joinedStream = windowedStream.transform { rdd => rdd.join(dataset) }
实际上，可以动态修改要进行join的数据集。提供给transform的函数会在每个批时间间隔进行评估，然后dataset指向的当前数据集。
完整的DStream转换列表在API文件中。Scala请参考DStream和PairDStreamFunctions。Java请参考avaDStream和JavaPairDStream。Python请参考DStream。
DStreams输出操作
输出操作允许DStream的数据写入到外部系统中，如数据库或者文件系统。由于输出操作实际上允许转换数据被外部系统消费，所以输出操作出发了所有DStream转换的额实际执行（类似RDDs的action）。目前，定义了如下输出操作：
| 输出操作 | 含义 |
|---|---|
| print() | 在运行streaming应用程序的驱动节点上输出DStream中每个批次的前10个元素。 对于开发和调试非常有用。Python API要调用pprint()。 |
| saveAsTextFiles(prefix, [suffix]) | 将DStream的内容存储为文件。每个批次时间间隔的文件名字基于prefix和suffix生成："prefix-TIME_IN_MS[.suffix]"。 |
| saveAsObjectFiles(prefix, [suffix]) | 将DStream的内容存储为Java对象序列化的SequenceFiles。每个批次时间间隔的文件名字基于prefix和suffix生成："prefix-TIME_IN_MS[.suffix]"。Python API不支持。 |
| saveAsHadoopFiles(prefix, [suffix]) | 将DStream的内容存储为Hadoop文件。每个批次时间间隔的文件名字基于prefix和suffix生成："prefix-TIME_IN_MS[.suffix]"。Python API不支持。 |
| foreachRDD(func) | 最通用的输出操作符，在stream生成的每个RDD上应用函数func。这个函数应该讲每个RDD的数据发送到外部系统，如保存RDD到文件或者通过网络写入到数据库。注意，函数func在运行streaming应用程序的驱动进程中执行，并且函数中通常会有RDD actions出发streaming RDDs的计算。 |
使用foreachRDD的设计模式
dstream.foreachRDD是一个强大的原语，允许数据发送到外部系统。但是，理解如何正确高效地使用这个原语非常重要。避免一些常见错误的方式如下。
通常写数据到外部系统要求创建连接对象（如TCP连接到远程服务）并使用连接发送数据到远程系统。为达到这个目的，开发者可能大一地在Spark driver创建连接对象，然后尝试在Spark worker中使用连接来保存RDDs中的记录。例如：
dstream.foreachRDD { rdd =>
val connection = createNewConnection() // executed at the driver
rdd.foreach { record =>
connection.send(record) // executed at the worker
}
}
这种做法是错误，因为这要求连接对象要序列化并且发送到worker中去。这样的连接对象很少跨机器进行传递。这个错误可能会显示为序列化错误（连接对象未进行序列化），初始化错误（连接对象需要在worker节点初始化）等等。正确的方法是在worker节点创建连接对象。
但是，这会导致另外一个常见错误 - 为每条记录创建一个新的连接。例如，
dstream.foreachRDD { rdd =>
rdd.foreach { record =>
val connection = createNewConnection()
connection.send(record)
connection.close()
}
}
创建连接对象需要时间和资源开销。因此，为每条记录创建和销毁连接对象会引发不必要的高开销并会显著降低系统的吞吐量。好的解决防范是使用rdd.foreachPartition - 创建一个连接对象，然后使用它发送一个RDD分区中的所有记录。
dstream.foreachRDD { rdd =>
rdd.foreachPartition { partitionOfRecords =>
val connection = createNewConnection()
partitionOfRecords.foreach(record => connection.send(record))
connection.close()
}
}
这种方式将连接创建的开销平摊到多条记录当中了。
最后，可以通过在多个RDDs/批次之间重用连接对象进一步优化。可以维护一个连接对象的静态池，重用连接对象将RDDs的多个批次发送到外部系统，进一步减少开销。
dstream.foreachRDD { rdd =>
rdd.foreachPartition { partitionOfRecords =>
// ConnectionPool is a static, lazily initialized pool of connections
val connection = ConnectionPool.getConnection()
partitionOfRecords.foreach(record => connection.send(record))
ConnectionPool.returnConnection(connection) // return to the pool for future reuse
}
}
注意，池子中的连接应该是按照需要懒惰创建的并且如果一定时间不用会超时。这实现了向外部系统发送数据最有效的方式。
其它需要记住的点：
- DStream通过输出操作懒惰执行，就想RDDs通过RDD action懒惰执行。DStream输出操作中的RDD actions会强制接收数据的处理。因此，如果应用程序没有任何输出操作或者有
dstream.foreachRDD()这种不带任何RDD action的输出操作，什么也不会执行。系统会简单地接收数据并丢弃数据。 - 默认地，输出操作只会执行一次。会按照在应用程序的定义顺序执行。
DataFrame和SQL操作
可对流数据使用DataFrames and SQL操作。需要使用SparkContext创建一个SparkSession。必须这样做，才能从驱动程序错误中恢复重启。通过创建SparkSession的懒实例化单例完成。下面是一个示例。修改了之前的示例，使用DataFrames和SQL生成单词计数。每个RDD会转换为DataFrame，作为临时表注册，然后使用SQL查询。
/** DataFrame operations inside your streaming program */
val words: DStream[String] = ...
words.foreachRDD { rdd =>
// Get the singleton instance of SparkSession
val spark = SparkSession.builder.config(rdd.sparkContext.getConf).getOrCreate()
import spark.implicits._
// Convert RDD[String] to DataFrame
val wordsDataFrame = rdd.toDF("word")
// Create a temporary view
wordsDataFrame.createOrReplaceTempView("words")
// Do word count on DataFrame using SQL and print it
val wordCountsDataFrame =
spark.sql("select word, count(*) as total from words group by word")
wordCountsDataFrame.show()
}
完整代码参见source code。
也可以从另外一个线程在表上执行查询（异步运行StreamingContext）。只要保证设置StreamingContext记住查询需要的足量数据即可。否则StreamingContext，无法识别异步查询，会在查询完成前删除旧的流数据。例如，如果想查询上一个批次的数据，但是你的查询可能需要运行5分钟，则调用streamingContext.remember(Minutes(5))。
关于DataFrame的更多信息参见DataFrames and SQL。
MLib操作
可以使用MLlib提供的机器学习算法。首先，streaming机器学习算法（如Streaming Linear Regression， Streaming KMeans等），可以同时从流数据中学习并在流数据上应用模型。除了这些，对于更大一类的机器学习算法，可以离线学习模型（如使用历史数据），然后将模型应用于在线流数据。具体参见MLlib。
网友评论