Spark-Streaming高级实践课件

SparkStreaming高级实践王富平2018年3月Spark-Streaming简介1SPARK是什么？批处理框架•MapReduce的进化版高性能•内存加速、分布式计算生态丰富机器学习库(mlib)、图计算(graghX)、Sql、SparkStreamingSpark-Streaming简介1SPARKStreaming是什么？实时处理框架•基于批处理理念设计的实时处理框架高性能•内存加速、分布式计算生态丰富机器学习库(mlib)、图计算(graghX)、SqlSpark-Streaming简介1基于批处理理念设计的实时处理框架基于spark先天优势，与各类型数据库、数据源结合较好朋友圈强大，spark的sql、mlib等组件，在sparkstreaming可以直接试用Spark-Streaming简介1WordCount例子Spark-Streaming高级特性—Window特性2WordCount需求升级：每10s统计一次单词在前1分钟内出现次数需求特点：数据复用•当前批次计算需要使用前一批次数据。聚合操作•词频计算，是对前一分钟内数据进行聚合操作Spark-Streaming高级特性—Window特性2Windowwidth长度该设置为多少？Slidinginternal该如何设置？Batchinternal该如何设置？BatchInternal•分批时间间隔Windowwidth•窗口长度•窗口长度必须是”batchinternal”整数倍SlidingInternal•窗口步进间隔•窗口步进间隔必须是”batchinternal”整数倍Spark-Streaming高级特性—Window特性2函数名参数描述windowwindowDuration：窗口长度创建窗口计算流windowwindowDuration：窗口长度slideDuration：滑动间隔创建窗口计算流创建WINDOW数据流Spark-Streaming高级特性—Window特性2函数名参数描述countByWindowwindowDuration：窗口长度slideDuration：滑动间隔返回窗口内记录数countByValueAndWindowwindowDuration：窗口长度slideDuration：滑动间隔返回窗口内不同value的记录数WINDOW相关计算函数Spark-Streaming高级特性—Window特性2计算窗口期内：请求失败率Spark-Streaming高级特性—Sql特性3Sparkstream数据封装成DataFrame（DataSet）之后，天然可以使用sparksql特性注册临时表valrecordsDF=spark.createDataFrame((1to100).map(i=>Record(i,s"val_$i")))recordsDF.createOrReplaceTempView("records")sql("SELECT*FROMrecordsrJOINsrcsONr.key=s.key").show()表joinsql("SELECT*FROMrecordsrJOINsrcsONr.key=s.key").show()Spark-Streaming高级特性—Sql特性3定义UDF加载jar包udfvalsqlContext=newHiveContext(sc)sqlContext.sql(“createtemporaryfunctiongetCateas'com.yihaodian.udf.hive.GetUrlCate'USINGJAR'SearchUDF-1.1-SNAPSHOT.jar‘”)动态定义UDFspark.udf.register("upper",(in:String)=>in.toUpperCase)Spark-Streaming高级特性—checkpoint机制4Stream通过使用CheckPoint保存处理状态甚至当前处理数据，一旦任务失败后，可以利用checkPoint数据进行恢复checkpoint特点：Metadatacheckpointing•保存stream任务配置•保存stream处理逻辑•保存stream当前未完成数据处理任务Datacheckpointing•保存inputstream输入数据，当任务异常退出恢复后，数据重新处理•保存状态数据，比如reduceByWindow、updateStateByKey需要使用上一批次的数据Spark-Streaming高级特性—checkpoint机制3sparkstreaming从1.2开始提供了数据的零丢失，想享受这个特性，需要满足如下条件•可靠的sources和可靠的receivers•应用metadatacheckpoint•WAL（writeaheadlog）也就是datacheckpoint无WALSpark-Streaming高级特性—checkpoint机制4加入WALSpark-Streaming高级特性—checkpoint机制3Wal缺点•降低了receivers的性能，因为数据还要存储到HDFS等分布式文件系统•只能保证At-Least-Once，不能保证exactly-onceKafkadirectAPI•为了WAL的性能损失和exactly-once，sparkstreaming1.3中使用KafkadirectAPI。非常巧妙，Sparkdriver计算下个batch的offsets，指导executor消费对应的topics和partitions。消费Kafka消息，就像消费文件系统文件一样。NDCG指标计算3NDCG是什么NormalizedDiscountedCumulativeGain，即NDCG，常用作搜索排序的评价指标，理想情况下排序越靠前的搜索结果，点击概率越大，即得分越高(gain)。CG=排序结果的得分求和，discounted是根据排名，对每个结果得分*排名权重，权重=1/log(1+排名)，排名越靠前的权重越高。首先我们计算理想DCG（称之为IDCG），再根据用户点击结果，计算真实的DCG，NDCG=DCG/IDCG，值越接近1，则代表搜索结果越好。DCG计算公式如下NDCG指标计算4在苏宁易购搜索关键词“苹果”，取第一排4个结果为例子NDCG指标计算4

结果1结果2结果3结果4实际点击次数0010理想点击次数1000IDCGDCG点击结果计算得出IDCG=1，DCG=0.5，NDCG=DCG/IDCG=0.5NDCG在sparkstreaming实现4数据调研86%的搜索行为在5分钟内完成90%的搜索行为在10分钟内完成时间窗口计算•对前15分钟数据的整体分析去重•一次搜索只计算一次NDCG在sparkstreaming实现4试用SparkStreaming框架的Window特性，实现时间窗口计算。时间窗口为15分钟，步进5分钟，意味着每5分钟计算一次每次计算，只对在区间［15分钟前，10分钟前］发起的搜索行为进行NDCG计算，这样就不会造成重复计算NDCG性能保障—容量预估5容量预估不是一个静态工作流量日志在不断增长，而系统处理能力是有限的

大促活动会造成额外的数据高峰。NDCG性能保障—容量预估5容量调节KafkaTopic分区数量SparkStreaming并发数SparkStreaming

cpu、内存NDCG性能保障—多维计算优化6NDCG指标为例子，目前支持4个维度组合的计算：大区、城市、渠道、搜索词，为了支持4个维度任意组合，需要进行15次计算，在HBas