这篇博客帮你开始使用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
结语
关于Spark Streaming with HBase的示例叙述就结束了。在以下参考部分，你可以找到更多信息。
Getting Started with Apache Spark: From Inception to Production Book