About Me 亚信科技 - 田毅 Spark 社区 Contributor 北京 SparkMeetup 的发起人主要关注 SparkSQL 与 Spark Streaming

Size: px

Start display at page:

Download "About Me 亚信科技 - 田毅 Spark 社区 Contributor 北京 SparkMeetup 的发起人主要关注 SparkSQL 与 Spark Streaming"

驿抄松
5 years ago
Views:

1 Spark 应用案例分析

2 About Me 亚信科技 - 田毅 Spark 社区 Contributor 北京 SparkMeetup 的发起人主要关注 SparkSQL 与 Spark Streaming

3 目录 Spark 的优势和收益 Spark 与现有 Hadoop 生态的互操作性 Spark 实践分享使用 Spark 的建议

4 Spark 的优势和收益

5 Spark 的优势和收益统一的数据处理架构混合使用多种计算类型操作兼容大量数据源类型降低开发复杂度丰富的 API 可读性强高性能的执行引擎

6 统一的数据处理架构应用层 User Application API 层 SQL 分析计算 STREAMING 流计算 MLlib 机器学习 GraphX 图计算核心层 SPARK-CORE SparkSQL: 批处理关系查询交互式查询 SparkStreaming: 流式计算 MLlib: 以机器学习为代表的迭代型计算 GraphX: 图计算多种数据处理场景的统一实现

7 统一的数据处理架构 Spark HDFS JDBC HBase Cassan dra 通过 RDD 良好的扩展性, 可以方便的扩展 Spark 支持的数据源目前数据源方面已经支持 HDFS, JDBC, 数据格式支持 JSON, Parquet 等. Spark 将在 1.2 以及以后的版本中提供统一的数据源 API 以支持用户自定义数据源的扩展. 因此, 后续 Spark 将可以构建多数据源的统一数据处理平台

8 统一的数据处理架构传统方式 : 复杂的批量数据处理 : HDFS + MR + Hive 基于历史数据的交互式查询 : HDFS + Impala 基于实时数据流的数据处理 : Storm 存在如下问题 : 数据交互难资源争抢人员技能分散使用 Spark 带来的好处人员组织简单数据交互方便资源统一调配

9 降低开发复杂度 Spark 统一计算模型 RDD+DAG 对 Spark 用户提供丰富的 API 对 Spark 本身的运行机制更容易理解代码量的大大减少 Hadoop 2470k Hive 925k Impala 2320k Spark 291k

10 降低开发复杂度提升代码可读性 Spark: val lines = ssc.textfilestream(args(0)) val words = lines.flatmap(_.split(" ")) val wordcounts = words.map(x => (x, 1)).reduceByKey(_ + _) MapReduce: public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>{ private final static IntWritable one = new IntWritable(1); private Text word = new Text(); public void map(object key, Text value, Context context) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasmoretokens()) { word.set(itr.nexttoken()); context.write(word, one); } } }

11 高性能的执行引擎天下武功, 唯快不破和 MR 相比, Spark 的性能优势主要体现在两个地方 : 高效的事件触发机制 + 多线程的执行机制, 使得 Task 的执行非常高效, 启动时间达到亚秒级 DAG 的编程范式大大的减少了数据分析计算过程中的持久化落地动作

12 高性能的执行引擎 Spark 颠覆 MapReduce 保持的排序记录 Yahoo 2013 年的记录海量数据的 SORT 性能 10% 的机器数量 12% 的 CPU 核数 3 倍集群速度

13 Spark 与现有 Hadoop 生态的互操作性

14 Spark 与 Hadoop 版本 Spark 支持多个 Hadoop 版本的实现众多 Hadoop 商业版本提供商也纷纷收录 Spark 到最新的发行版本 Cloudera, Hontonworks, MapR, Pivotal

15 Spark 与 Yarn 目前 Spark 提供了 yarn-cluster 和 yarn-client 两种模式支持 Spark 程序在 yarn 上运行实际应用中, 比较常用 : yarn-client 模式 : 有交互需求的常驻应用使用 spark-shell 来执行 scala 脚本使用 thrift-server 来执行 hql 脚本 yarn-cluster 模式 : 无交互需求的临时应用使用 spark-submit 来执行 spark 应用程序将 Spark 运行在 Yarn 上的好处可以与 MR 良好的协调使用资源可以通过 Hadoop 的安全机制运行在 Kerberized cluster

16 Spark 与 Hive Spark1.0 之前, 通过 Shark 实现 Hive 的大部分功能从 Spark 1.0 开始, Spark 增加了一个重要的组件, Spark SQL Spark SQL 目前主要通过 HiveContext 实现了 : 访问 Hive 仓库中所有数据的能力大部分 HQL 语法的解析和执行 Hive on Spark(HIVE-7292) 目前正在开发之中, 很快就可以成为 hive 的第三个执行引擎 ( 前两个是 mr 和 tez)

17 Spark 实践分享

18 Spark 实践分享批量处理离线数据多维数据分析查询信令数据的实时处理使用 mllib 预测用户行为

19 在企业数据仓库中使用 Shark 模型分层技术分层 ST 层 ( 指标, 报表展示数据 ) RDBMS DW 层 ( 不同维度关联汇总 ) DWD 层 ( 汇总数据 ) ODS 层 ( 明细数据 ) Hadoop (Hive) 原始数据 (DB2, Oracle, CSV)

20 在企业数据仓库中使用 Shark 问题 : 部分程序迁移到 hive 后, 执行速度变慢使用 Shark 的目的 : 提速小数据量的模型计算步骤较多的模型计算使用 Shark 的方式 : 以 Standalone 方式运行, 使用主机的 20% 计算资源 MR 使用主机 80% 计算资源切换使用 Hiveserver 与 Sharkserver 来实现不同的计算方式 Applicaiton SharkServer HiveServer Spark MapReduce HDFS

21 在企业数据仓库中使用 Shark TCL 名 Hive Shark 时间减少 dw_new_comp_town_ds.tcl % co_dtdsm_gd.tcl % co_dtdym_new.tcl % nb_atvdt_td.tcl % pr_stcdr.tcl % ac_dwprtacctdm.tcl % dw_new_chuanka_user_town_ds.tcl % ac_dwgrpptdmnew.tcl % dw_new_chuanka_lt_town_ds.tcl % nb_dwmmsdm.tcl % nb_wland.tcl % cr_24houroutdm.tcl % 说明 : hive 使用 80% 资源 Shark 使用 20% 资源提速效果根据不同应用特点不尽相同相对流程较复杂的应用优化效果更好如果部分临时表使用内存表的方式效果更佳

22 在企业数据仓库中使用 Shark 存在的问题 [ 现象 ]: 某些 SQL 的运行长时间不结束, 并且不断的有 FullGC 出现, 最终出现 OutOfMemory 的现象任务失败 [ 分析 ]: spark 在 agg 和 cogroup 时, 通过 spill 机制, shuffle 数据可临时写入磁盘 ; shark 重写了 CoGroupRDD 类, 使用 hashmap 进行数据存储, 但不具备 spill 功能 [ 结论 ]: 在处理大数据关联的场景时, 在同一个 Group 下数据量过大, shark 无法支撑解决办法之一, 选择 shark 与 hive 并存, 通过 yarn 实现资源共享解决办法之二, 增大 Reduce Task 的数量 SparkSQL 目前也存在类似的问题关键是针对不同场景下数据的分布规律, 无法用一种方法同时解决性能与稳定性这两个问题

23 多维数据分析查询市场部策划了一个营销活动, 为了在有限的成本下提升营销效果, 怎样精确定位客户群, 准确选择目标客户? 根据业务经验, 筛选标签创建客户群, 潜在终端营销用户, 进行营销

24 多维数据分析查询表名周期列数 SPARK_DATE 日周期 284 SPARK_MONTH 月周期 1748 Client ThriC Server Spark Cluster 最终这条查询时间平均使用了 2 秒

25 多维数据分析查询问题 -> 解决思路 : 1 数据和 task 分布不均, 跨节点访问造成部分 task 慢尽量增加 hdfs 上的备份数, 减少数据跨节点访问 2 Spark 没有创建足够多的 Task 并发执行增加 HDFS 文件 block 数, 增加文件数 3 磁盘 IO 耗时较长使用 RCFile 格式 +GZ 压缩存储 4 并发查询较多时, 查询耗时变长需要增加 ThriftServer 的内存, 避免大量 GC 5 序列化的 CPU 消耗较大选用 kyro 序列化方式结论 : 从性能角度看, 使用 SparkSQL 作为多维数据分析查询与搜索引擎相比不是最佳的实践, 性能结果只能勉强可用从系统整体角度看, 如果在整体业务中, 这部分不是核心功能, 使用 SparkSQL 也是一个很好的选择

26 多维数据分析查询扩展性能测试并发 4 节点 30 并发 8 节点 sql1 sql2 sql3 sql4 sql5 sql6 sql7 sql8 sql9 sql10 sql11 sql12 sql13 sql14 sql15 sql16 sql17 sql18 sql19 sql20 sql21 sql22 sql23

27 多维数据分析查询并发性能测试节点 1 并发 8 节点 5 并发 8 节点 10 并发 8 节点 20 并发 8 节点 30 并发 5 0 sql1 sql2 sql3 sql4 sql5 sql6 sql7 sql8 sql9 sql10 sql11 sql12

28 实现信令数据的实时处理场景描述 : 输入数据用户信令数据 : 每秒钟 5w 条业务需求沉淀 : 形成用户实时位置信息和行为轨迹匹配 : 合适的业务, 对用户进行主动营销需要支持多业务的扩展输出数据用户实时位置信息用户历史行为轨迹需要进行业务营销的用户信息

29 实现信令数据的实时处理消息队列 Spark Streaming 消息队列增量数据预处理业务流程外围系统增量数据增量数据业务流程 Hbase 每秒支持 15W Req 每条记录需要交互查询 4-8 次无法满足要求增量数据 HBase 关联数据

30 实现信令数据的实时处理消息队列 Spark Streaming 消息队列增量数据预处理业务流程外围系统增量数据增量数据增量数据关联更新 HDFS 业务流程广播变量动态数据只读数据

31 实现信令数据的实时处理预处理流程业务 : 沉淀用户实时位置信息和行为轨迹在流数据上增加用户历史位置信息实现方式 : val cogroup = leftrdd.cogroup(rightrdd).map { (lout, rout) } cogroup.cache() cogroup.flatmap(_._1) cogroup.flatmap(_._2) 流入数据输出数据 COGroup 用户历史用户历史更新

32 实现信令数据的实时处理子流程业务 ( 举例 ): 判断是否校园用户 ( 根据校园基站列表 ) 判断是否营销对象 ( 根据用户资料表与营销规则表 ) 对比更新营销结果表 ( 避免重复营销 ) 数据通过 Kafka 发给营销系统实现方法 : inputdstream.foreachrdd(rdd=>{ }) rdd.. registerastable( inputtable") val temprs = sql( xxxxx ) temprs.. registerastable( temptable") val result = sql( xxxxx ) 为什么使用 SparkSQL 这部分逻辑变化较快, 使用 SQL 便于维护人员修改 ( 会写 scala 的人太少了 ) 流入数据关联表 1 SQL 临时数据 SQL 输出数据关联表 2

33 使用 MLlib 进行用户行为预测 AD Exchange 广告竞价平台流量过滤数据处理平台更新模型活动匹配训练效果评估出价更新训练集效果跟踪收集日志

34 使用 Mllib 进行用户行为预测业务背景 : 为. 举例 : 实现 : 通过 mllib 中的逻辑回归算法根据用户历史行为记录预测用户未来的行根据用户历史点击广告的记录, 判断用户是否会点击指定的广告 LogisticRegressionWithSGD 算法样本 : 用户标签, 广告标签, 广告位标签用户是否点击广告性能 : 阿里云 Yarn 集群 25 个 Executor 2Core 2G 内存 5W 个维度条数据, 每次计算耗时 243 秒

35 使用 Spark 的建议

36 使用 Spark 的建议 SQL on Spark 目前虽然有多个方案实现, Shark, SparkSQL, Hive on Spark 但是都不太具备单独运行在生产系统中的能力解决办法 : 如果业务不是很庞大 ( 包含的 SQL 不是很多 ), 推荐使用 Spark 的 API 组织一个 Spark 应用来实现业务逻辑使用 Hive 作为备选方案

37 使用 Spark 的建议使用 Spark 进行流计算适合场景 : 允许延迟在 2-5 秒及以上的业务对吞吐量要求较高, 且增长迅速的业务 Spark 进行流计算的优势可以直接写 SQL 处理数据相对方便的使用 Hive 或 hdfs 中的数据使用到外部数据常用方法外部数据量远大于流数据时 : 使用 Hbase, redis, memcache 等外部系统进行逐条存储, 查询外部数据不是很大, 且需要读写操作直接使用 Spark 读取并关联外部数据以批处理的方式进行查询和更新外部数据较小, 且只读使用广播变量的方式将外部数据提前广播到各个节点

38 使用 Spark 的建议使用 Spark 要考虑到 Scala, Akka 等新元素带来的学习成本使用 Spark 建议先从最适合 Spark 特点的业务开始尝试当前 Spark 上的问题还很多, 有点类似 hadoop 0.20.x 版本的阶段使用 Spark 需要很好的和社区进行关注与互动

39 使用 Spark 的建议如果想了解更多的 Spark 实践分享, 请关注 Spark 在各地的 meetup 小组活动目前已经有北京, 上海, 杭州, 深圳四个地方

40 Q & A

41 @InfoQ infoqchina

About Me 田亚信科技 - 田毅 Spark 社区 Contributor 北京 SparkMeetup 的发起人主要关注 SparkSQL 与 Spark Streaming

About Me 田亚信科技 - 田毅 Spark 社区 Contributor 北京 SparkMeetup 的发起人主要关注 SparkSQL 与 Spark Streaming Spark 技术研究与实践分享 About Me 田毅 tianyi.asiainfo@gmail.com @ 亚信科技 - 田毅 Spark 社区 Contributor 北京 SparkMeetup 的发起人主要关注 SparkSQL 与 Spark Streaming 提纲为什么选择 Spark Spark 实践分享使用 Spark

About Me 亚信科技 - 田毅 Spark 社区 Contributor 北京 SparkMeetup 的发起人 主要关注 SparkSQL 与 Spark Streaming

About Me 亚信科技 - 田毅 Spark 社区 Contributor 北京 SparkMeetup 的发起人主要关注 SparkSQL 与 Spark Streaming