Spark Streaming 入门指南

2022-05-26 14:47:44

这篇博客帮你开始使用Apache Spark Streaming和HBase。Spark Streaming是核心Spark API的一个扩展，它能够处理连续数据流。

Spark Streaming是什么？

首先，Spark Streaming是什么？数据流是数据连续到来的无限序列。Streaming划分连续流动的输入数据成离散单元以便处理。流处理是对流数据的低延迟处理和分析。Spark Streaming是核心Spark API的一个扩展，能够允许对实时数据的可扩展，高吞吐量，容错流处理。Spark Streaming用于需要快速处理实时到来的大数据量的应用情形。

实时应用情形的实例有以下场合：

网络站点监控，网络监控
欺诈侦测
网络点击
广告
物联网传感器

Spark Streaming支持的数据源有：HDFS directories, TCP sockets, Kafka, Flume, Twitter等等。Data Streams能被Spark的核心APIS，DataFrames SQL，或机器学习APIs，来处理，还能够持久化到文件系统，HDFS，数据库，或任何提供Hadoop OutputFormat的数据源。

Spark Streaming是怎么运行的

Spark Streaming划分一个数据流成X秒一组的数据，它被称为Dstreams，本质上是RDDs的一个序列。你的Spark应用程序使用Spark APIs处理RDDs，RDD操作的处理结果成批地返回。

一个Streaming应用实例的架构

Spark Streaming实例代码完成了以下操作：

读流数据
处理流数据
将处理过的数据写入到HBase Table

另外的Spark实例代码会执行下面操作：

读取由streaming代码写入的HBase Table数据
计算每日的汇总统计结果
将统计结果写入到HBase Table Column Family stats（HBase Table统计列族）

样本数据集

油量泵传感器数据来自于目录中一个逗号分隔值（csv）文件。Spark Streaming监控那个目录，并处理目录中创建的任何文件。（正如之前提到的，Spark Streaming支持不同的流数据源；为了简单化，这个例子使用了文件。）下面是csv文件样本数据的一个例子：

我们使用Scala 样本类（case class）定义对应于传感器数据csv文件的传感器数据库模式，使用一个parseSensor函数去解析在传感器样本类（case class）中的逗号分隔值。

HBase Table Schema

流数据的HBase Table Schema如下：

泵名称日期和时间戳的复合行键。
列中的Column Family数据对应于输入数据域，Column Family告警列对应于警报值筛选器。注意，经过一定时间后，数据和告警列集会被设置成终止值。

日常统计数据汇总归纳的Schema列举如下：

泵名称和日期的复合行键。
统计列族（Column Family stats）
最小列，最大列，平均列

下面的函数转换一个Sensor 对象成为一个HBase Put对象，用于往HBase插入一行。

写入一个HBase Table的配置

你可以使用Spark中的TableOutputFormat 类来写一个HBase Table，类似于你怎样从MapReduce 往HBase Table中写数据。下面我们使用TableOutputFormat类来设置要往HBase中写的配置。

Spark Streaming例子代码

这些是Spark Streaming代码的基本步骤：

1.初始化一个Spark StreamingContext对象；

2.对DStreams应用转换和输出操作；

3.使用streamingContext.start()来开始接收数据和处理；

4.使用streamingContext.awaitTermination()来等待处理过程停止；

我们会使用实际的应用程序代码完成上面的每一步。

初始化StreamingContext

首先，我们创建一个StreamingContext，它是流功能函数主入口点，是有一个两秒批间隔。

val sparkConf = new SparkConf().setAppName("HBaseStream")
//  create a StreamingContext, the main entry point for all streaming functionality
val ssc = new StreamingContext(sparkConf, Seconds(2))

下一步，我们使用StreamingContext textFileStream(目录) 方法创建输入流，它用于监控Hadoop-兼容（Hadoop-compatible）的文件系统，文件系统用于新文件，输入流也处理在这个目录下任何文件创建的操作。

// create a DStream that represents streaming data from a directory source
val linesDStream = ssc.textFileStream("/user/user01/stream")

那个linesDStream代表数据流，每条记录是文本中的一行。本质上，DStream 是RDDs的一个序列，代表每批次间隔的一个RDD。

对DStreams应用转换和输出操作

接着，我们解析数据行成Sensor对象，对linesDStream使用映射操作.

// parse each line of data in linesDStream  into sensor objects
val sensorDStream = linesDStream.map(Sensor.parseSensor)

映射操作使用Sensor.parseSensor函数对linesDStream中的RDDs进行操作，获得了RDDs of Sensor对象。

接下来，我们使用DStream的 foreachRDD方法去对这个DStream中的每个RDD进行处理。我们筛选那个传感器对象到适合的低压力值去创建告警，通过转换它们成Put对象来将传感器和告警数据写入到HBase中，接着使用PairRDDFunctions 的saveAsHadoopDataset方法，输出RDD到支持Hadoop的任何存储系统,那个存储系统使用一个Hadoop Configuration对象（查看上面的HBase的Hadoop Configuration）。

// for each RDD. performs function on each RDD in DStream
sensorRDD.foreachRDD { rdd =>
        // filter sensor data for low psi
     val alertRDD = rdd.filter(sensor => sensor.psi < 5.0)
      // convert sensor data to put object and write to HBase  Table CF data
      rdd.map(Sensor.convertToPut).saveAsHadoopDataset(jobConfig)
     // convert alert to put object write to HBase  Table CF alerts
     rdd.map(Sensor.convertToPutAlert).saveAsHadoopDataset(jobConfig)
}

那个传感器RDD对象被转换成Put对象，然后写入HBase。

开始接收数据

为了开始接收数据，我们必须在StreamingContext中显式地调用start()，然后调用awaitTermination去等待流计算直到它终止。

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

从Spark中读取和写回HBase

现在我们想要读取HBase传感器表数据，计算每日汇总统计数据，将统计数据写到统计列族。

下面的代码读取HBase传感器表压力列数据，使用StatCounter计算这个数据的统计值，将统计数据写入到传感器统计列族。

    // configure HBase for reading
    val conf = HBaseConfiguration.create()
    conf.set(TableInputFormat.INPUT_TABLE, HBaseSensorStream.tableName)
    // scan data column family psi column
    conf.set(TableInputFormat.SCAN_COLUMNS, "data:psi")
// Load an RDD of (row key, row Result) tuples from the table
    val hBaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat],
      classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
      classOf[org.apache.hadoop.hbase.client.Result])
    // transform (row key, row Result) tuples into an RDD of Results
    val resultRDD = hBaseRDD.map(tuple => tuple._2)
    // transform into an RDD of (RowKey, ColumnValue)s , with Time removed from row key
    val keyValueRDD = resultRDD.
              map(result => (Bytes.toString(result.getRow()).
              split(" ")(0), Bytes.toDouble(result.value)))
    // group by rowkey , get statistics for column value
    val keyStatsRDD = keyValueRDD.
             groupByKey().
             mapValues(list => StatCounter(list))
    // convert rowkey, stats to put and write to hbase table stats column family
    keyStatsRDD.map { case (k, v) => convertToPut(k, v) }.saveAsHadoopDataset(jobConfig)

下面这个图显示了newAPIHadoopRDD的输出是一个RDD行键值，成对出现。那个PairRDDFunctions的 saveAsHadoopDataset方法保存Put对象到HBase中。

软件

这个示例运行在MapR Sandbox上，包含了Spark。

你可以从下面网址下载代码和数据来运行这个例子：https://github.com/caroljmcdonald/SparkStreamingHBaseExample

运行程序

你可以运行那些代码作为一个独立应用程序，正如Getting Started with Spark on MapR Sandbox示例中所描述的。

这里是步骤总结：

使用用户名user01，密码mapr登录MapR Sandbox, 正如Getting Started with Spark on MapR Sandbox所描述的。
使用Maven编译应用程序。
使用scp命令拷贝jar文件和数据文件到你的sandbox根目录/user/user01。

执行流app：

/opt/mapr/spark/spark-<version>/bin/spark-submit --driver-class-path `hbase classpath`
--class examples.HBaseSensorStream sparkstreamhbaseapp-1.0.jar

拷贝流数据文件到流目录：
```
cp sensordata.csv /user/user01/stream/
```

读数据计算一列的统计数据：

/opt/mapr/spark/spark-<version>/bin/spark-submit --driver-class-path `hbase classpath`
--class examples.HBaseReadWrite sparkstreamhbaseapp-1.0.jar

计算整行的统计数据：

/opt/mapr/spark/spark-<version>/bin/spark-submit --driver-class-path `hbase classpath`
--class examples.HBaseReadRowWriteStats sparkstreamhbaseapp-1.0.jar