Spark介绍与应用案例分析.pdf

spark技术研究与实践分享 about me 田毅 亚信科技-田毅 spark社区contributor 北京sparkmeetup的发起人 主要关注sparksql与spark streaming 提纲 为什么选择spark spark实践分享 使用spark的建议 为什么选择spark 多种计算场景的结合 多数据源的计算中心 活跃的社区支持 多种计算场景的结合 sql 批处理 streaming 流处理 mllib 机器学习 graphx 图计算 spark core （sparkcontextrdddag） user application 用户应用 随着信息技术的发展, 越来越多的企业面临着复杂计算场景的考验 1 机器学习的不断发展和应用 2 信息时效性决定了流处理技术的重要性 3 传统业务人员操作熟练的sql编写能力 多种计算场景的结合 假设场景：与新浪微博合作，通过一个消息队列实时接收微博信息，根据指定关 键字过滤消息 传统方案：使用storm读取消息队列内容，设定bolt进行关键字过滤 微 博 消 息 队 列 storm 消 息 队 列 多种计算场景的结合 假设场景：与新浪微博合作，通过一个消息队列实时接收微博信息，根据指定关 键字过滤消息，再通过实时配置sql对微博进行统计分析，生成实时报表 传统方案： 使用storm读取消息队列内容，设定bolt进行关键字过滤，将结果写入hdfs 使用hive或者impala实现sql统计分析 微 博 消 息 队 列 storm h d f s hive/ impala 多种计算场景的结合 假设场景：与新浪微博合作，通过一个消息队列实时接收微博信息，根据指定关键字过滤消息，通过机 器学习，对关键字不断进行调整，再通过实时配置sql对微博进行统计分析，生成实时报表； 传统方案： 使用storm读取消息队列内容，设定bolt进行关键字过滤，将结果写入hdfs 使用hive或者impala实现sql统计分析 使用mahout实现机器学习算法，将训练后的算法模型回传给storm 微 博 消 息 队 列 storm h d f s hive/ impala mahout 多种计算场景的结合 spark方案： 优势： 1 同一套架构，学习成本较低 2 资源可统一规划 3 流计算与machine learning数据交互简单 微 博 消 息 队 列 spark streaming mllib 消 息 队 列 商业产品：databricks cloud 开源产品：zeppelin /nflabs/ zeppelin 多数据源的计算中心 对于大多数公司来说, 数据会根据应用场景被存储到多种数据源 以我们熟悉的电信行业举例: 但是, 这些数据单独应用只能满足企业内部若干独立的应用场景 想要真正的从数据中获得最大的价值, 必须让所有数据关联到一起进行计算分析 数据类型 举例 应用场景 存储方式 单据类数据 通信详单, 账单 随机查询 hbase, cassandra 日志类数据 信令数据, 应用 日志 汇总分析 hdfs 关系类数据 用户资料, 订购 关系 实时更新, 关联 查询 rdbms 多数据源的计算中心 hdfs hbase rdbms mr etl 计算引擎 复杂的数据同步流程 极大消耗网络带宽和存储资源 多数据源的计算中心 spark 1.1.0 通过扩展rdd实现外部数据访问 hdfs hbase rdbms spark hadoopfilerdd 使用rdd扩展存在的问题: 只能全量获取, 网络压力大,无用传输太多 引用cheng liandatabricks在meetup上的slide 多数据源的计算中心 baserelation prunedscan tablescan prunedfilteredscan spark 1.2.0 external datasource api 全量扫描 指定列扫描 根据filter指定列扫描 (parquet and orc) 尽可能将列过滤与行过滤在server端进行, 降低传输大小 可以更好的利用数据源的特性 引用cheng liandatabricks在meetup上的slide 多数据源的计算中心 spark hbase user_bill 查询所有开通gprs用户中,各项增值业务的用 户数,平均年龄,总花费 select t2.businame, count(t2.user), avg(age), sum(t2.fee) from ( select businame, user, sum(fee) as fee from user_bill where busitype in (sms,gprs) group by businame, user ) t1 left outer join ( select user, age from user_info where user_usegprs=1 ) t2 on t1.user = t2.user where t2.user is not null group by t2.businame db2 user_info db2relation requiredcolumns=(user,age) filter=(“user_usegprs=1”) sql resultset rdd hbaserelation rdd rowset filter requiredcolumns filter 多数据源的计算中心 企业级数据计算中心 spark hbase rmdbs json parquet cassandra hdfs 活跃的社区支持 2013年6月 2014年6月 2014年12月 contributor数 量 68 255 368 参与贡献的公 司 17 50 未统计 代码行数 63000 175000 239000 活跃的社区支持 活跃的邮件列表: 日均70+的邮件 jira问题收集: 日平均报告/解决 issue 15个 github pull request: 日平均merge pull request 10个 活跃的社区支持 11月刚刚推出的模块维护人制度, 确保每个模块都有至少2个committer专门跟踪维护 - spark core public api: matei, patrick, reynold - job scheduler: matei, kay, patrick - shuffle and network: reynold, aaron, matei - block manager: reynold, aaron - yarn: tom, andrew or - python: josh, matei - mllib: xiangrui, matei - sql: michael, reynold - streaming: td, matei - graphx: ankur, joey, reynold spark实践分享 使用spark实现信令数据的实时营销 使用spark实现广告竞价效果实时反馈平台 使用spark实现信令数据的实时营销 场景描述: 输入数据 用户信令数据: 每秒钟5w条 业务需求 沉淀: 形成用户实时位置信息和行为轨迹 匹配: 合适的业务, 对用户进行主动营销 需要支持多业务的扩展 输出数据 用户实时位置信息 用户历史行为轨迹 需要进行业务营销的用户信息 难点 大量的数据查询更新 业务架构 消 息 队 列 主流程 消 息 队 列 业务流程1 业务流程2 业务流程3 数据清洗 数据增强 数据筛选 业务判断 数据筛选 业务判断 数据筛选 业务判断 数据沉淀 技术架构1.0 外围 系统 消息队列 增量 数据 增量 数据 增量 数据 spark streaming hbase 关联数据 预处理 消息队列 业务流程 业务流程 增量 数据 hdfs 关联数 据1 关联数 据2 关联数 据n 输出 数据 输出 数据 输出 数据 技术架构1.0实践 优化： spark优化 kafka接收数据优化：多topic，多dstream，repatition task并行数量优化 hbase优化 预建多分区 balance 表现：集群处理吞吐能力无法满足要求 分析 : 集群规模较小, spark最大并发任务数不到300, 同时hbase操作平均不到150 hbase单次request处理时间1-2ms, 每秒处理700(单线程) 总计处理100000每秒 实际需求: 5w * 4 = 200000每秒 结论：机器数量有限时，有限的任务并行度会限制hbase的吞吐能力 技术架构2.0 外围 系统 消息队列 增量 数据 增量 数据 增量 数据 消息队列 hdfs 动态数据 只读数据 增量 数据 广 播 变 量 rdd file spark streaming 预处理 业务流程 业务流程 输出 数据 输出 数据 输出 数据 实现信令数据的实时处理 预处理流程业务: 沉淀用户实时位置信息和行为轨迹 在流数据上增加用户历史位置信息 实现方式: val cogroup = leftrdd.cogroup(rightrdd).map (lout, rout) cogroup.cache() cogroup.flatmap(_._1) cogroup.flatmap(_._2) 流入数据 用户历史 输出数据 用户历史 更新 cogroup 实现信令数据的实时处理 子流程业务(举例): 判断是否校园用户(根据校园基站列表) 判断是否营销对象(根据用户资料表与营销规则表) 对比更新营销结果表(避免重复营销) 数据通过kafka发给营销系统 实现方法: inputdstream.foreachrdd(rdd= rdd. . registerastable(”inputtable“) val temprs = sql(“xxxxx”) temprs . . registerastable(”temptable“) val result = sql(“xxxxx”) ) 流入数据 关联表1 关联表2 临时数据 sql sql 输出数据 使用spark实现广告竞价效果实时反馈平台 广 告 位 目标网站 xx 网站 dsp 平台 广告 交易 平台 浏览网页 点击广告 跳转 竞价请求 出价 浏览记录 购买记录 竞价成功 报表 系统 使用spark实现广告竞价效果实时反馈平台 业务需求1： 1 实时收集所有出价记录，竞价成功记录，浏览记录和购买记录 2 按广告位统计： 最近2000次的竞价成功次数 最近2000次的平均成功价格 最近2000次的点击比率 3 将实时统计结果反馈到竞价模块对竞价策略进行调整 业务需求2： 1 按广告主统计: 出价次数 花费金额 转化率等等指标 2 将实时统计结果更新到报表模块展示 技术难点： 1 数据量较大，每秒消息数量在3-5万 2 不按照常用的时间窗口统计，而按照竞价次数统计 spark streaming 技术架构 dsp 平 台 hdfs 日志1 spark streaming 日志1预处理 日志2 日志3 日志4 日志2预处理 日志3预处理 日志4预处理 spark streaming 竞价统计反馈 报表统计 消息队列 技术架构 dsp 平 台 hdfs 日志1 spark streaming 日志1预处理 日志2 日志3 日志4 日志2预处理 日志3预处理 日志4预处理 竞价统计反馈 报表统计 union &cache 预处理 数据清洗： dstream.filter: 清洗非法格式数据 dstream.map: 清洗不使用的数据字段 数据聚合： dstream.reducebykey: 对数据进行统计 聚合维度: 广告位 广告主 格式转换： dstream.map: 将数据转换格式为统一格式 数据清洗 数据聚合 格式转换 数据获取 竞价统计反馈 数据获取 数据聚合 数据输出 数据获取： dstream.filter: 按需获取需要的数据 数据聚合： dstream.updatestatebykey: 对每个广告位的状态 (统计信息)进行更新 其中state可以是自定义的class 数据输出： dstream.mappartition: 将数据输出到指定的接口 （http或者jdbc） sparkstreaming实施中的问题 hdfs 文件正在生成时文件后缀问题 java.io.filenotfoundexception: file does not exist: / user/streaming/tmp/test/bidinput/2bid.gz._copying_ 产生原因: sparkstreaming读取目录时没有过滤正在拷贝的文件 patch: spark-4314 sparkstreaming实施中的问题 fileinputdstream只能读取单级目录 对于这样的目录层级, 无法使用sparkstreaming读取 -data |-20141201 |-20141202 |-20141203 |-20141204 patch: spark-3586 sparksql相关patch hashouterjoin优化spark-4483 通过单表遍历的方式, 替换原有两边hashmap的join方式 100万 join 1万 性能对比, 性能提升16%, 内存消耗减少70% master: 耗时: 12671 ms 耗时: 9021 ms 耗时: 9200 ms current mem usage:787788984 after patch： 耗时: 10382 ms 耗时: 7543 ms 耗时: 7469 ms current mem usage:208145728 sparksql相关patch broadcasthashouterjoin优化spark-4485 通过broadcast实现小表在map端实现outerjoin 性能对比, 性能提升7倍 original: left outer join : 15439 ms right outer join : 9707 ms optimized: left outer join : 1992 ms right outer join : 1