Spark平台在电信运营商的应用实践资料.pdf

Spark平台在电信运营商的应用实践 亚信⼤大数据平台 ⽥田毅 ⺫⽬目录 ★项⺫⽬目实践分享 ★基于Spark改造⽤用户标签分析查询平台 ★基于Spark Streaming改造内容识别处理平台 ★⼀一些⼼心得分享 ★如何⽤用好External DataSource API ★⾼高效的在Spark Streaming中引⽤用外部数据 ○基于Spark改造⽤用户标签分析查询平台 3 TCL脚本 ○基于Spark改造⽤用户标签分析查询平台 • 改造前的设计 4 数据库 指标表 通信数据 上⺴⽹网数据 数据清洗指标计算标签计算 标签表接⼝口表 ⽤用户 数据探索客户群计算 SQL ○基于Spark改造⽤用户标签分析查询平台 • 改造前的问题 • 1 标签数量越来越⼤大，数据库负载过⾼高，扩展成本⾼高 • 2 标签表的列数随着标签数量增加不断增多，部分现场达到2000+，只能通过分表的⽅方式解决，查 询时需要Join操作 • 3 标签与指标的计算⽆无法摆脱SQL的约束，⽆无法快速集成机器学习的算法 TCL脚本 ○基于Spark改造⽤用户标签分析查询平台 • 第⼀一次改造设计: ⼩小试⽜牛⼑刀 6 数据库 指标表 通信数据 上⺴⽹网数据 数据清洗指标计算标签计算 标签表接⼝口表 ⽤用户 数据探索客户群计算 SparkSQL HDFS ○基于Spark改造⽤用户标签分析查询平台 • 改造后的好处 • 1 使⽤用SparkSQL ＋ Parquet的⽅方案，有效保证了查询效率 • 2 原有系统基本不⽤用太⼤大改造 • 3 查询系统具备平⾏行扩展能⼒力 • 未解决的问题 • 1 标签与指标的计算⽆无法摆脱SQL的约束，⽆无法快速集成机器学习的算法 • 产⽣生出来的新问题 • 1 增加了从数据库倒出数据，加载到HDFS的额外步骤 • 2 增加了从⽂文本数据转化为Parquet格式的额外步骤 SparkSQL ○基于Spark改造⽤用户标签分析查询平台 • 第⼆二次改造设计：⼤大⼑刀阔斧 8 HDFS 指标表 通信数据 上⺴⽹网数据 数据清洗指标计算标签计算 标签表接⼝口表 ⽤用户 数据探索客户群计算 SparkSQL ○基于Spark改造⽤用户标签分析查询平台 • 改造后的好处 • 1 通过SparkSQL替换掉了原有的数据库，整个系统的扩展性进⼀一步增强 • 2 两套SparkSQL可以根据各⾃自忙闲时的不同，共享整个系统的计算资源 • 遗留的问题 • 1 没有摆脱标签分析算法对于SQL的依赖 • 2 系统前端仍然依赖ETL系统对数据进⾏行抽取加载 • 怎么破？ ○基于Spark改造⽤用户标签分析查询平台 • Spark 1.3.0 发布了 • External Datasource API进⼀一步增强 • DataFrame提供了丰富多样的数据源⽀支持 • DataFrame提供了⼀一整套⽤用于操纵数据的DSL 外部系统 Spark Applications Based on DF ○基于Spark改造⽤用户标签分析查询平台 • 第三次改造设计：更进⼀一步 11 HDFS 指标表 数据表1 数据表2 指标计算 标签计算 标签表 ⽤用户 数据探索客户群计算 SparkSQL ExtDataSour ce 按需抽取数 据 原有SQL转化 为DF的API ○基于Spark改造⽤用户标签分析查询平台 • 改造后的好处 • 1 终于摆脱了对SQL的依赖，为后续引⼊入复杂算法分析打下基础 • 2 利⽤用External Datasource API可以根据计算需求从源表抽取指定的数据 • 3 基于DF的处理程序代码量仅有原程序的1/10，可读性⼤大⼤大提⾼高 • 遗留的问题 • 1 如何控制对源数据库的压⼒力问题 == 时间窗 • 2 Ext DS的实现对于不同的数据库类型需要进⾏行细致的优化 基于Spark Streaming改造内容识别平台 内容识别平台——功能介绍 产品⺫⽬目标：通过对上⺴⽹网⽇日志的分析还原⽤用户上⺴⽹网时的场景 ⽤用户上⺴⽹网记录 ⼊入⼝口识别： APP还是浏览器 应⽤用识别： 微博，，UC URL内容识别： 新闻，体育 MapReduce Job Map Task 基于Spark Streaming改造内容识别平台 改造前的设计 上⺴⽹网数据输出⺫⽬目录输⼊入⺫⽬目录 Distribute Cache 数据分析 后续系统 标签分析 ⽇日志查询 营销活动 识别规则 识别规则 基于Spark Streaming改造内容识别平台 • 改造前的问题 • 1 数据处理延迟较⾼高 • 2 需要频繁加载规则数据到内存 • 3 数据源逐渐变为实时接⼝口 Spark Streaming DStream.map(…) 基于Spark Streaming改造内容识别平台 改造前的设计 上⺴⽹网数据 Kafka Broadcast 数据分析 后续系统 标签分析 ⽇日志查询 营销活动 Kafka input topic output topic 识别规则 将原有的HDFS 改为Kafka接⼝口 MR引擎换为 Spark Streaming 原有MR的Map处理逻辑 迁移到DStream的map ⽅方法 规则数据改为通过 ⼲⼴广播发布到所有的 Executor 基于Spark Streaming改造内容识别平台 • 改造后的好处 • 1 数据分析的代码逻辑⼏几乎没有修改, 兼容了原有的HDFS⽂文件接⼝口 • 2 规则数据只需要⼀一次加载，可以⻓长期保存在executor的内存中 • 3 通过kafka ＋ spark streaming实现了流式处理的要求 • 4 数据处理延迟从原有的分钟级别降低到秒级 • 改造过程的经验 • 1 序列化问题 == 使⽤用Kryo序列化需要注意先注册 • 2 流处理框架和业务逻辑两部分代码建议完全隔离 • 3 业务逻辑可以保持java的实现⽅方式，通过反射等⽅方式调⽤用业务逻辑的代码 如何⽤用好External DataSource API • External Datasource API 是Spark 1.2.0版本中⼀一个重要的feature • 赋予Spark平台⾼高效灵活访问外部数据源的能⼒力 • 我们⽇日常的使⽤用中也经常会遇到数据存在于多个数据源之中的场景 • 如何使⽤用Ext DS的API来实现对多个数据源的⽀支持呢？ • 让我们⽤用HBase作为外部数据源举个例⼦子 18 如何⽤用好External DataSource API • ⺴⽹网上已经有很多HBase的Ext DS的实现，如： • • • 我们简要的分析⼀一下实现HBase的Ext DS的⼏几个要点 Task Task Task Task Executor Executor Region Region Region Region RegionServer RegionServer 如何才能达到这样的实现呢？我们先来看看External DS的设计原理 如何⽤用好External DataSource API Optimizer Filter Pushdown Column Pruning 解析为Ext Relation Analyzer Resolve relation Strategy Build PhysicalRDD 将Filter尽可能push 到Ext Relation，同 时进⾏行Column Pruning SQL Spark Plan 根据Ext Relation实 现，调⽤用buildScan ⽣生成RDD 如何⽤用好External DataSource API DAGScheduler Submit JobSubmit StageSubmit Task External RDD Job rdd.partitionsrdd. getPreferredLocations Executor Executor Executor pute ⾼高效的在Spark Streaming中引⽤用外部数据 • 实现如何分partition • 接⼝口定义 RDD.scala /** * Implemented by subclasses to return the set of partitions in this RDD. This method will only * be called once, so it is safe to implement a time-consuming computation in it. */ protected def getPartitions: Array[Partition] • 实现思路： • 根据createRelation⽅方法中传⼊入的参数确定HBase的连接⽅方式和表名 • 通过调⽤用HBase的API获取该表中所有Region的列表 • 根据buildScan中传⼊入的Array[Filter]对Region进⾏行⼀一定的过滤 • 根据Region列表⽣生成⼀一个 Array[Partition] • 这样做的好处： • 每个Partition只处理⼀一个Region的数据，为后⾯面的getPreferredLocations做基础 22 ⾼高效的在Spark Streaming中引⽤用外部数据 • 实现getPreferredLocations⽅方法 • 接⼝口定义 RDD.scala /** * Optionally overridden by subclasses to specify placement preferences. */ protected def getPreferredLocations(split: Partition): Seq[String] = Nil • 实现思路： • 根据split中包含的Region信息，来确定这个Region在哪个节点 • 这样做的好处： • Task在调度的时候可以优先被调度到Region所在的机器上执⾏行，减少⺴⽹网络传输 23 ⾼高效的在Spark Streaming中引⽤用外部数据 • 实现compute⽅方法 • 接⼝口定义 RDD.scala /** * :: DeveloperApi :: * Implemented by subclasses to compute a given partition. */ @DeveloperApi def compute(split: Partition, context: TaskContext): Iterator[T] • 实现思路： • 根据split中包含的Region信息、filter、requiredColumns调⽤用HBase的API进⾏行查询 24 如何⽤用好External DataSource API • 实现Ext DS的⼏几个要点 • 1 继承RelationProvider实现⼀一个createRelation⽅方法，来获取所有参数 • 2 继承BaseRelation实现⼀一个根据⾃自定义的schema⽣生成⽅方法 • 3 根据数据源类型选择实现不同级别的buildScan（还有InsertableRelation可以实现插⼊入） • 4 在buildScan中根据push down的过滤条件⽣生成RDD • 5 RDD中实现如何分partition • 6 RDD中实现getPreferredLocations⽅方法 • 7 RDD中实现compute⽅方法 ⾼高效的在Spark Streaming中引⽤用外部数据 • 为什么要引⽤用外部数据 • 流式处理中输⼊入数据包含的信息有限 • ⼤大量的使⽤用场景需要关联外部数据进⾏行逻辑判断处理 • Spark Streaming⾃自⾝身提供的⼀一些⽅方法 • 只读数据：Broadcast • 更新数据：UpdateStateByKey • 存在的问题 • 。