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……输出为1002.设计思路利⽤异或运算将列表中的所有ID异或之后得到的值即为所求ID。先将每个分区的数据异或然后将结果进⾏异或运算。3.代码⽰例CountOnce的代码⽰例如下import org.apache.spark.SparkContext import org.apache.spark.SparkContext._ */ object NumOnce { def computeOneNumargs Array[String] { val spark new SparkContextlocal[1] NumOnce System.getenvSPARK_HOME SparkContext.jarOfClassthis.getClass val data spark.textFiledata /*每个分区分别对数据进⾏异或运算最后在reduceByKey阶段将各分区异或运算的结果再做异或运算 合并。偶数次出现的数字异或运算之后为0奇数次出现的数字异或后为数字本⾝*/ val result data.mapPartitionsiter { var temp iter.next whileiter.hasNext { temp temp^iter.next } Seq1 temp.iterator }.reduceByKey_^_.collect printlnnum appear once is result0 } }4.应⽤场景数据块损坏检索。例如每个数据块有两个副本有⼀个数据块损坏需要找到那个数据块。6.6 倾斜连接并⾏计算中我们总希望分配的每⼀个任务task都能以相似的粒度来切分且完成时间相差不⼤。但是由于集群中的硬件和应⽤的类型不同、切分的数据⼤⼩不⼀总会导致部分任务极⼤地拖慢了整个任务的完成时间。硬件不同暂且不论下⾯举例说明不同应⽤类型的情况如Page Rank或者Data Mining中的⼀些计算它的每条记录消耗的成本不太⼀样这⾥只讨论关于关系型运算的Join连接的数据倾斜状况。数据倾斜原因如下。1业务数据本⾝的特性。2Key分布不均匀。3建表时考虑不周。4某些SQL语句本⾝就有数据倾斜。数据倾斜表现如下。任务进度⻓时间维持查看任务监控⻚⾯由于其处理的数据量与其他任务差异过⼤会发现只有少量1个或⼏个任务未完成。1.实例描述输⼊表A数据倾斜表B输出表CAB连接后的表2.设计思路数据倾斜有很多解决⽅案本例简要介绍⼀种实现⽅式。假设表A和表B连接表A数据倾斜只有⼀个Key倾斜。⾸先对A进⾏采样统计出最倾斜的Key。将A表分隔为A1只有倾斜KeyA2不包含倾斜Key然后分别与B连接。3.代码⽰例倾斜连接的代码⽰例如下import org.apache.spark.SparkContext import org.apache.spark.SparkContext._ */ object SkewJoin { def mainargs Array[String] { val skewedTable left.execute val spark new SparkContextlocal TopK System.getenvSPARK_HOME SparkContext.jarOfClassthis.getClass /*存在数据倾斜的数据表*/ val skewTable spark.textFileskewTable /*与skewTable连接的表*/ val Table spark.textFileTable 对数据倾斜的表 ⾏ 样 假 有 个 倾斜 重 获 倾斜 ⼤的/*对数据倾斜的表进⾏采样假设只有⼀个key倾斜最严重获取倾斜最⼤的key*/ val sample skewTable.samplefalse 0.3 9.groupByKey.collect val maxrowKey sample.maprows rows._2.size rows._1.maxByrows rows._1._2 /*将倾斜的表拆分为两个RDD⼀个为只含有倾斜key的表⼀个为不含有倾斜key的表*/ /*分别与原表连接*/ val maxKeySkewedTable skewTable.filterrow { buildSideKeyGeneratorrow maxrowKey } val mainSkewedTable skewTable.filterrow { buildSideKeyGeneratorrow maxrowKey } /*分别与原表连接*/ val maxKeyJoinedRdd maxKeySkewedTable.joinTable val mainJoinedRdd mainSkewedTable.joinTable /*将结果合并*/ sc.unionmaxKeyJoinedRdd 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 fishery2.设计思路通过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 createSockethandleMessage String Unit { websocket WebSocket.openws//localhost 8080/1.0/marketDataWs.onTextMessagem { handleMessagem } subscriptions.foreachlisting websocket.sendText{\subscribe\{ listing }} } var websocket WebSocket _ } classPriceEchoextendsPriceWebSocketClient{ createSocketprintln }为了能够让Spark正确挂接到WebSocket并不断接收消息可以通过实现⼀个接收器Receiver达到这个⽬的。由于接收的数据符合通⽤的⽹络协议所以通过继承NetworkReceiver类实现接收器。⽤户需要创建⼀个块⽣成器block generator并将接收到的消息附加到块⽣成器中。classPriceReceiverextendsNetworkReceiver[String]withPriceWebSocketClient{ lazy val blockGenerator new BlockGeneratorStorageLevel.MEMORY_ONLY_SERprotected override def onStart { blockGenerator.startcreateSocketm blockGenerator m }protected override def onStop { blockGenerator.stopwebsocket.shutdown}}到⽬前为⽌获取的流数据是以JSON格式存储的⽂本字符串通过抽取数据中重要的部分进⽽⽤其创建case类这样数据处理将变得更容易。创建⼀个PriceUpdate case类。import scala.util.parsing.json.JSON import scala.collection.JavaConversions import java.util.TreeMap case class PriceUpdateid String price Double lastPrice Double object PriceUpdate{ val lastPrices JavaConversions.asMapnew TreeMap[StringDouble] def applytext String PriceUpdate { valid price getIdAndPriceFromJSONtext val lastPrice Double lastPrices.getOrElseid price lastPrices.putid price PriceUpdateid price lastPrice } /*此⽅法解析与处理JSON数据格式暂不赘述*/ def getIdAndPriceFromJSONtext String // snip - simple JSON processing } /*这时还不能找到⾦融序列属性不能获取之前的价格信息。 同时需要更新接收器类为下⾯的情况解析输⼊数据*/ import spark.streaming.dstream.NetworkReceiver import spark.storage.StorageLevel class PriceReceiver extends NetworkReceiver[PriceUpdate]withPriceWebSocketClient{ lazy val blockGenerator new BlockGeneratorStorageLevel.MEMORY_ONLY_SER protected override def onStart { blockGenerator.start createSocketm { val priceUpdate PriceUpdatem 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 StreamingContextlocal datastream Seconds15 C:/software/spark-1.0.0 Listtarget/scala-2.10.3/spark-data-stream_2.10.3- 1.0.jar /* 创建并注册输⼊流*/ ssc.registerInputStreamstream /*启动流数据处理引擎*/ ssc.start上⾯这段代码初始化了流数据处理的上下⽂并配置了应⽤。local表⽰在本地执⾏datastream是应⽤的名称Seconds15定义批处理的时间⽚C:/software/spark-1.0.0定义Spark的路径Listtarget/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 mainargs Array[String] { import Listings._ val ssc new StreamingContextlocal datastream Seconds15 C:/software/spark-0.7.3 Listtarget/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.registerInputStreamstream stream.mappu listingNamespu.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.mappu listingSectorspu.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 DoubleInt pair DoubleInt { ifpair._1 0 reduced._1 pair._1 reduced._2 pair._2 elsereduced._1 pair._1 reduced._2 - pair._2 } val invReduce reduced DoubleInt pair DoubleInt { ifpair._1 0 reduced._1 pair._1 reduced._2 - pair._2 elsereduced._1 pair._1 reduced._2 pair._2 } val windowedPriceChanges sectorPriceChanges.reduceByKeyAndWindowreduce invReduce Seconds5*60 Seconds15现在通过上⽂介绍的函数已经可以构建⼀个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 mainargs Array[String] { import Listings._ import System._ val ssc new StreamingContextlocal datastream Seconds15 C:/software/spark-0.7.3 Listtarget/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.checkpointspark ssc.registerInputStreamstream val reduce reduced DoubleInt pair DoubleInt { ifpair._1 0 reduced._1 pair._1 reduced._2 pair._2 elsereduced._1 pair._1 reduced._2 - pair._2 } val invReduce reduced DoubleInt pair DoubleInt { ifpair._1 0 reduced._1 pair._1 reduced._2 - pair._2 elsereduced._1 pair._1 reduced._2 pair._2 } val sectorPriceChanges stream.mappu listingSectorspu.id pu.price - pu.lastPrice 1 val windowedPriceChanges sectorPriceChanges.reduceByKeyAndWindowreduce invReduce Seconds5*60 Seconds15 val positivePriceChanges windowedPriceChanges.filter{case _ _ count count 0} val priceChangesToSector positivePriceChanges.map{casesector value count value * count sector} val sortedSectors priceChangesToSector.transformrdd rdd.sortByKey false.map_._2 sortedSectors.foreachrdd { println|---------------------------------------------- |Positive Trending Time %d ms |---------------------------------------------- |.stripMargin.formatcurrentTimeMillis rdd.take5.mapsectorCodes _.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