6.5 CountOnce

假设HDFS只存储⼀个标号为ID的Block，每份数据保存2个备份，这样就有2个机器存储了相同的数据。其中ID是⼩于10亿的整数。若有⼀个数据块丢失，则需要找到哪个是丢失的数据块。

在某个时间，如果得到⼀个数据块ID的列表，能否快速地找到这个表中仅出现⼀次的ID？即快速找出出现故障的数据块的ID。

问题阐述：已知⼀个数组，数组中只有⼀个数据是出现⼀遍的，其他数据都是出现两遍，将出现⼀次的数据找出。

1.实例描述

输⼊为Block ID。

1、2、2、3、3、1、5、7、11……

输出为：

100

2.设计思路

利⽤异或运算将列表中的所有ID异或，之后得到的值即为所求ID。先将每个分区的数据异或，然后将结果进⾏异或运算。

3.代码⽰例

CountOnce的代码⽰例如下：

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
*/
object NumOnce {
def computeOneNum（args： Array[String]） {
val spark = new SparkContext（"local[1]"， "NumOnce"，
System.getenv（"SPARK_HOME"）， SparkContext.jarOfClass（this.getClass））
val data = spark.textFile（"data"）
/*每个分区分别对数据进⾏异或运算，最后在reduceByKey阶段，将各分区异或运算的结果再做异或运算
合并。偶数次出现的数字，异或运算之后为0，奇数次出现的数字，异或后为数字本⾝*/
val result = data.mapPartitions（iter => {
var temp = iter.next（）
while（iter.hasNext） {
temp = temp^iter.next（）
}
Seq（（1， temp））.iterator
}）.reduceByKey（_^_）.collect（）
println（"num appear once is ："+result（0））
}
}

4.应⽤场景

数据块损坏检索。例如，每个数据块有两个副本，有⼀个数据块损坏，需要找到那个数据块。

6.6 倾斜连接

并⾏计算中，我们总希望分配的每⼀个任务（task）都能以相似的粒度来切分，且完成时间相差不⼤。但是由于集群中的硬件和应⽤的类型不同、切分的数据⼤⼩不⼀，总会导致部分任务极⼤地拖慢了整个任务的完成时间。硬件不同暂且不论，下⾯举例说明不同应⽤类型的情况，如Page Rank或者Data Mining中的⼀些计算，它的每条记录消耗的成本不太⼀样，这⾥只讨论关于关系型运算的Join连接的数据倾斜状况。

数据倾斜原因如下。

1）业务数据本⾝的特性。

2）Key分布不均匀。

3）建表时考虑不周。

4）某些SQL语句本⾝就有数据倾斜。

数据倾斜表现如下。

任务进度⻓时间维持，查看任务监控⻚⾯，由于其处理的数据量与其他任务差异过⼤，会发现只有少量（1个或⼏个）任务未完成。

1.实例描述

输⼊：

表A（数据倾斜），表B

输出：

表C（A，B连接后的表）

2.设计思路

数据倾斜有很多解决⽅案，本例简要介绍⼀种实现⽅式。假设表A和表B连接，表A数据倾斜，只有⼀个Key倾斜。⾸先对A进⾏采样，统计出最倾斜的Key。将A表分隔为A1只有倾斜Key，A2不包含倾斜Key，然后分别与B连接。

3.代码⽰例

倾斜连接的代码⽰例如下：

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
*/
object SkewJoin {
def main（args ： Array[String]） {
val skewedTable = left.execute（）
val spark = new SparkContext（"local"， "TopK"，
System.getenv（"SPARK_HOME"）， SparkContext.jarOfClass（this.getClass））
/*存在数据倾斜的数据表*/
val skewTable = spark.textFile（"skewTable"）
/*与skewTable连接的表*/
val Table = spark.textFile（"Table"）
对数据倾斜的表 ⾏ 样 假 有 个 倾斜 重 获 倾斜 ⼤的/*对数据倾斜的表进⾏采样，假设只有⼀个key倾斜最严重，获取倾斜最⼤的key*/
val sample = skewTable.sample（false， 0.3， 9）.groupByKey（）.collect（）
val maxrowKey = sample.map（rows => （rows._2.size， rows._1））.maxBy（rows =>
rows._1）._2）
/*将倾斜的表拆分为两个RDD，⼀个为只含有倾斜key的表，⼀个为不含有倾斜key的表*/
/*分别与原表连接*/
val maxKeySkewedTable = skewTable.filter（row => {
buildSideKeyGenerator（row） == maxrowKey
}）
val mainSkewedTable = skewTable.filter（row => {
！（buildSideKeyGenerator（row） == maxrowKey）
}）
/*分别与原表连接*/
val maxKeyJoinedRdd = maxKeySkewedTable.join（Table）
val mainJoinedRdd = mainSkewedTable.join（Table）
/*将结果合并*/
sc.union（maxKeyJoinedRdd， mainJoinedRdd）
}
}

4.应⽤场景

在⼤数据分析平台中，经常遇到数据倾斜问题，读者可以参照相应的思路处理数据倾斜的处理。SQL on Hadoop系统中也需要处理数据倾斜问题。

6.7 股票趋势预测

本例将介绍如何使⽤Spark构建实时数据分析应⽤[1]，以分析股票价格趋势。

本例假设已预先连接了Spark Streaming。读者可以阅读介绍BDAS的章节预先了解相关概念。

第⼀步，需要获取数据流，本例使⽤JSON/WebSocket格式呈现6种实时市场⾦融信息。

第⼆步，需要知道如何使⽤获取到的数据流，本例不涉及专业的⾦融知识，但可以在这个应⽤中通过价格改变规律，预测价格趋势。

1.实例描述

本例通过使⽤scalawebsocket库（Github⽹址为https://github.com/pbuda/scalawebsocket）访问WebSocket。scalawebsocket库只⽀持Scala 2.10。获取⽹上的⾦融数据流。

输⼊为：股票名和相应价格。

-------------------------------------------
Time： 1375194945000 ms
-------------------------------------------
Croda International PLC - 24.82 - 24.82
ASOS PLC - 47.485 - 47.485
Arian Silver Corp - 0.0435 - 0.0435Medicx Fund Ltd - 0.7975 - 0.7975
Supergroup PLC - 10.73 - 10.73
Diageo PLC - 20.07 - 20.075
Barclays PLC - 2.891 - 2.8925
QinetiQ Group PLC - 1.874 - 1.874
CSR PLC - 5.7 - 5.7
United Utilities Group PLC - 7.23 - 7.23
输出为：处于增⻓趋势的股票名称。
----------------------------------------------
Positive Trending （Time： 1375269240035 ms）
----------------------------------------------
Real estate
Telecommunication
Graphics， publishing & printing media
Environmental services & recycling
Agriculture & fishery

2.设计思路

通过Spark Streaming的时间窗⼝，增加新数据，减少旧数据。本例中的reduce函数⽤于对所有价格改变求和（有正向的改变和负向的改变）。之后希望看到正向的价格改变数量是否⼤于负向的价格改变数量，这⾥通过改变正向数据将计数器加1，改变负向的数据将计数器减1进⾏统计，从⽽统计出股票的趋势。

3.代码⽰例

通过本书的BDAS章节，假设读者已经对Spark和Spark Streaming有了初步了解，下⾯将介绍整个应⽤的设计与开发。

（1）接收流数据

为了在Spark中处理流数据，需要创建⼀个StreamingContext对象（Spark Streaming中的上下⽂对象），作为流处理的上下⽂。之后注册⼀个输⼊流（InputDStream），它会初始化⼀个接收器（Receiver）对象（Spark默认提供了许多类型的接收器，如Twitter、Akka Actor、ZeroMQ等）。由于默认没有⽹⻚套接字（WebSocket）的实现，所以本例将⾃定义这个类，获取⽹⻚流数据。

本例通过使⽤scalawebsocket库（Github⽹址为https://github.com/pbuda/scalawebsocket）访问WebSocket。scalawebsocket库只⽀持Scala 2.10。

读者可以选⽤⽀持Scala 2.10的Spark版本。

通过下⾯的代码实现⼀个简单的trait，进⽽使⽤WebSocket（它产⽣所有可⽤的股票序列）。

import scalawebsocket.WebSocket
trait PriceWebSocketClient {
import Listings._
def createSocket（handleMessage： String => Unit） = {
websocket = WebSocket（）.open（"ws：//localhost：
8080/1.0/marketDataWs"）.onTextMessage（m => {
handleMessage（m）
}）
subscriptions.foreach（listing => websocket.sendText（"{\"subscribe\"：{" +
listing + "}}"））
}
var websocket： WebSocket = _
}
classPriceEchoextendsPriceWebSocketClient{
createSocket（println）
}

为了能够让Spark正确挂接到WebSocket，并不断接收消息，可以通过实现⼀个接收器（Receiver）达到这个⽬的。由于接收的数据符合通⽤的⽹络协议，所以通过继承NetworkReceiver类实现接收器。⽤户需要创建⼀个块⽣成器（block generator），并将接收到的消息附加到块⽣成器中。

classPriceReceiverextendsNetworkReceiver[String]withPriceWebSocketClient{ lazy val blockGenerator = new BlockGenerator（StorageLevel.MEMORY_ONLY_SER）

protected override def onStart（） { blockGenerator.start

createSocket（m => blockGenerator += m） }

protected override def onStop（） { blockGenerator.stop

websocket.shutdown

}}

到⽬前为⽌，获取的流数据是以JSON格式存储的⽂本字符串，通过抽取数据中重要的部分进⽽⽤其创建case类，这样数据处理将变得更容易。创建⼀个PriceUpdate case类。

import scala.util.parsing.json.JSON
import scala.collection.JavaConversions
import java.util.TreeMap
case class PriceUpdate（id： String， price： Double， lastPrice： Double）
object PriceUpdate{
val lastPrices = JavaConversions.asMap（new TreeMap[String，Double]）
def apply（text： String）： PriceUpdate = {
val（id， price） = getIdAndPriceFromJSON（text）
val lastPrice： Double = lastPrices.getOrElse（id， price）
lastPrices.put（id， price）
PriceUpdate（id， price， lastPrice）
}
/*此⽅法解析与处理JSON数据格式，暂不赘述*/
def getIdAndPriceFromJSON（text： String） = // snip - simple JSON processing
}
/*这时，还不能找到⾦融序列属性，不能获取之前的价格信息。
同时，需要更新接收器类为下⾯的情况，解析输⼊数据*/
import spark.streaming.dstream.NetworkReceiver
import spark.storage.StorageLevel
class PriceReceiver extends NetworkReceiver[PriceUpdate]withPriceWebSocketClient{
（ ）lazy val blockGenerator = new BlockGenerator（StorageLevel.MEMORY_ONLY_SER）
protected override def onStart（） {
blockGenerator.start
createSocket（m => {
val priceUpdate = PriceUpdate（m）
blockGenerator += priceUpdate
}）
}
protected override def onStop（） {
blockGenerator.stop
websocket.shutdown
}}

还需要实现⼀个输⼊流（InputDStream），这个输⼊流需要实现getReceiver⽅法，当外部调⽤这个⽅法时，返回⼀个初始化好的价格接收器。

object streamextendsNetworkInputDStream[PriceUpdate]（ssc） { override def getReceiver（）： NetworkReceiver[PriceUpdate] = { newPriceReceiver（）

}}

将之前的程序加⼊Spark Streaming的主⼲程序中。

/*创建Spark Streaming上下⽂*/
val ssc = new StreamingContext（"local"， "datastream"， Seconds（15），
"C:/software/spark-1.0.0"， List（"target/scala-2.10.3=/spark-data-stream_2.10.3-
1.0.jar"））
/* 创建并注册输⼊流*/
ssc.registerInputStream（stream）
/*启动流数据处理引擎*/
ssc.start（）

上⾯这段代码初始化了流数据处理的上下⽂，并配置了应⽤。

local表⽰在本地执⾏，datastream是应⽤的名称，Seconds（15）定义批处理的时间⽚，"C:/software/spark-1.0.0"定义Spark的路径，List

（"target/scala-2.10.3=/spark-data-stream_2.10.3-1.0.jar"）定义需要的Jar包。

项⽬结构要为SBT的项⽬格式，之后在根⽬录运⾏SBT package run即可运⾏，这样将会编译，打包程序⽣成target/scala-2.10.3=/sparkdata-stream_2.10.3-1.0.jar，然后Spark应⽤使⽤Jar中的类。

下⾯代码为到⽬前为⽌应⽤的完整代码。

override def main（args： Array[String]） {
import Listings._
val ssc = new StreamingContext（"local"， "datastream"， Seconds（15），
"C:/software/spark-0.7.3"， List（"target/scala-2.9.3/spark-data-stream_2.9.3-
1.0.jar"））
object stream extends NetworkInputDStream[PriceUpdate]（ssc） {
override de fgetReceiver（）： NetworkReceiver[PriceUpdate] = {
newPriceReceiver（）
}}
ssc.registerInputStream（stream）
stream.map（pu => listingNames（pu.id） + " - " + pu.lastPrice + " - " +
pu.price）.print（）
ssc.start（）
}
控制台将会产⽣以下输出。
-------------------------------------------
Time： 1375194945000 ms
-------------------------------------------
Croda International PLC - 24.82 - 24.82
ASOS PLC - 47.485 - 47.485
Arian Silver Corp - 0.0435 - 0.0435
Medicx Fund Ltd - 0.7975 - 0.7975
Supergroup PLC - 10.73 - 10.73
Diageo PLC - 20.07 - 20.075
Barclays PLC - 2.891 - 2.8925
QinetiQ Group PLC - 1.874 - 1.874
CSR PLC - 5.7 - 5.7
United Utilities Group PLC - 7.23 - 7.23
……

（2）处理流数据

通过上⽂的初始化和数据接收，已经可以源源不断地获取数据了。下⾯介绍如何处理和分析数据。

下⾯程序可将数据转化为类股，改变价格和频度的序列。第⼀次处理时，将每个数据项转化为（类股，价格改变，1）的元组。通过下⾯的代码完成这个过程。

val sectorPriceChanges = stream.map（pu => （listingSectors（pu.id）， （pu.price -

pu.lastPrice， 1）））

现在就可以使⽤reduceByKeyAndWindow函数了，这个函数允许⽤户使⽤滑动窗⼝处理数据，时间窗⼝内的数据将会使⽤reduce函数处理，使⽤Key-Value对中的Key作为reduce的关键字，这⾥将使⽤⼀个reduce函数和反向reduce函数。

这样每次在时间窗⼝内迭代时，Spark都对新数据进⾏reduce处理，需要丢弃的旧数据不再使⽤reduce处理。

Spark需要做的就是撤销之前最左侧旧数据对整个reduce数据结果的改变，增加右侧新的reduce数据对整个reduce数据结果产⽣新的改变。

需要写⼀个reduce和inverse reduce函数。在本例中，reduce函数⽤于对所有价格改变求和（有正向的改变和负向的改变）。为了看到正向的价格改变数量⼤于负向的价格改变数量，这⾥可以通过改变正向数据，将计数器加1，改变负向的数据，将计数器减1达到这个效果。代码如下。

val reduce = （reduced： （Double，Int）， pair： （Double，Int）） => {
if（pair._1 > 0） （reduced._1 + pair._1， reduced._2 + pair._2）
else（reduced._1 + pair._1， reduced._2 - pair._2）
}
val invReduce = （reduced： （Double，Int）， pair： （Double，Int）） => {
if（pair._1 > 0） （reduced._1 + pair._1， reduced._2 - pair._2）
else（reduced._1 + pair._1， reduced._2 + pair._2）
}
val windowedPriceChanges = sectorPriceChanges.reduceByKeyAndWindow（reduce，
invReduce， Seconds（5*60）， Seconds（15））

现在通过上⽂介绍的函数，已经可以构建⼀个reduce流处理应⽤，这个应⽤能够感知价格波动和趋势。由于只希望呈现出正向波动最剧烈的⼀些类股。可以过滤流数据，保留下正向波动的类股，然后将数据元组Key-Value的Key改变为可以排序，统计出波动最⼤的类股的属性。本例假设正向波动剧烈与否的权重为价格改变⼤⼩乘以价格改变计数器值，将Value中的两个值组合计算出的结果作为新的Key。最后，将数据按照新的Key打印出最⼤的5个类股。
 

import scala.collection.immutable.List
import spark.SparkContext._
import spark.streaming._
import spark.streaming.StreamingContext._
import spark.streaming.dstream._
object DataStreamextendsApp{
val reportHeader = """----------------------------------------------
Positive Trending
=================""".stripMargin
override def main（args： Array[String]） {
import Listings._
import System._
val ssc = new StreamingContext（"local"， "datastream"， Seconds（15），
"C:/software/spark-0.7.3"， List（"target/scala-2.9.3/spark-data-stream_2.9.3-
1.0.jar"））
object stream extends NetworkInputDStream[PriceUpdate]（ssc） {
override def getReceiver（）： NetworkReceiver[PriceUpdate] = {
newPriceReceiver（）
}}
ssc.checkpoint（"spark"）
ssc.registerInputStream（stream）
val reduce = （reduced： （Double，Int）， pair： （Double，Int）） => {
if（pair._1 > 0） （reduced._1 + pair._1， reduced._2 + pair._2）
else（reduced._1 + pair._1， reduced._2 - pair._2）
}
val invReduce = （reduced： （Double，Int）， pair： （Double，Int）） => {
if（pair._1 > 0） （reduced._1 + pair._1， reduced._2 - pair._2）
else（reduced._1 + pair._1， reduced._2 + pair._2）
}
val sectorPriceChanges = stream.map（pu => （listingSectors（pu.id）， （pu.price -
pu.lastPrice， 1）））
val windowedPriceChanges = sectorPriceChanges.reduceByKeyAndWindow（reduce，
invReduce， Seconds（5*60）， Seconds（15））
val positivePriceChanges = windowedPriceChanges.filter{case （_， （_， count））
=> count > 0}
val priceChangesToSector = positivePriceChanges.map{case（sector， （value，
count）） => （value * count， sector）}
val sortedSectors = priceChangesToSector.transform（rdd => rdd.sortByKey
（false））.map（_._2）
sortedSectors.foreach（rdd => {
println（"""|----------------------------------------------
|Positive Trending （Time： %d ms）
|----------------------------------------------
|""".stripMargin.format（currentTimeMillis + rdd.take（5）.map（sectorCodes
（_））.mkString（"\n"））
}）
ssc.start（）
}}

运⾏上述⽰例，将会打印出下⾯的⽇志，这样就构建出了预测类股趋势的Spark应⽤。
 

Positive Trending （Time： 1375269240035 ms）
----------------------------------------------
Real estate
Telecommunication
Graphics， publishing & printing media
Environmental services & recycling
Agriculture & fishery----------------------------------------------
Positive Trending （Time： 1375269255035 ms）
----------------------------------------------
Real estate
Graphics， publishing & printing media
Environmental services & recycling
Agriculture & fishery
Electrical appliances & components
----------------------------------------------
Positive Trending （Time： 1375269270034 ms）
----------------------------------------------
Environmental services & recycling
Agriculture & fishery
Electrical appliances & components
Vehicles
Precious metals & precious stones