PowerPoint 演示文稿

Size: px

Start display at page:

Download "PowerPoint 演示文稿"

穴明
9 years ago
Views:

2 Apache Spark 与多数据源的结合田

3 目录为什么会用到多个数据源 Spark 的多数据源方案有哪些已有的数据源支持 Spark 在 GrowingIO 的实践分享

4 为什么会用到多个数据源从数据本身来看大数据的特性之一 :Variety 数据的多样性结构化数据与非结构化数据实时数据与离线数据五花八门的数据格式

5 为什么会用到多个数据源从业务需求来看不同的业务场景决定了数据需要被存储到多个地方数据挖掘实时查询多维分析查询聚合汇总统计实时更新

6 为什么会用到多个数据源从软件的发展来看越来越多的针对某一个细分领域的软件技术不断产生 NoSQL:HBase,Cassandra,MongoDB Storage:HDFS,Tachyon Search:ElasticSearch, Lucene MPP: Teradata,GreenPlum, Aster,Gbase QueryEngine:Hive,Phoenix Apache 基金会下的软件中 : BigData:30 个 Database:25 个

7 常见的大数据系统的架构应用计算引擎数据存储 ETL 系统数据 A 数据 B 数据 C 应用计算引擎 A 数据存储 A 计算引擎 B 数据存储 B 计算引擎 C 数据存储 C ETL 系统数据 A 数据 B 数据 C

8 数据存储间如何相互访问方案 1: 冗余存储方案 2: 集中计算问题 : 1 数据一致性 2 存储成本问题 : 1 数据类型转换 2 数据处理效率存储 1 存储 2 存储 3 集中计算存储 1 存储 2 存储 3 数据数据数据数据

9 Spark 的多数据源方案 Spark 在版本中首次发布了 Data Sources API 这套 API 主要提供了一种快速灵活的方法为 Spark 提供访问外部数据源的功能主要目标是让 Spark 各个组件以及外部应用可以方便高效的读写外部数据

10 Spark DataSource API 问题 1: 关于数据类型转换 Spark SQL 中的一套函数式关系查询优化框架 Catalyst Catalyst 提供了一整套数据类型的定义各个数据源实现各自数据类型和 Catalyst 的转换

11 Spark DataSource API 问题 2: 关于数据处理效率 Spark 提供了一个灵活的 api 来兼容不同类型的 DataSource 包括 : 支持全量扫描支持列剪枝支持列剪枝 + 过滤机制支持数据插入

12 在 Spark DataSource API 之前

13 在 Spark DataSource API 之后

14 可以使用的 Spark DataSource Spark 自带 JDBC Mysql PostgreSQL HadoopFs Parquet JSON Orc Spark Packages 网站 Avro CSV RedShift MongoDB Cassandra ElasticSearch 还有其他软件内置的一些 DataSource Apache Phoenix

15 JDBC 支持列剪枝支持条件过滤支持数据插入 sqlcontext.read.jdbc( jdbc:postgresql://testhost:7531/testdb, testtable, idfield, 索引列 10000, 起始 index , 结束 index 10, partition 数量 new Properties ).registertemptable("testtable") Driver Executor Executor Executor JDBC Server 适合场景 : 配置数据的加载, 处理问题 : 随着数据量增加, 对 JDBC Server 的压力骤增

16 HadoopFs 获取文件列表, 生成多个 Task 支持列剪枝支持条件过滤支持数据插入已经支持 Parquet,JSON, sqlcontext.read.parquet("hdfs://testfs/testpath").registertemptable("test") Driver Executor Executor Executor HDFS 适合场景 : 离线处理中输入输出数据, 临时数据问题 : 实时场景下, 无法增量写入数据, 连续写入会产生大量碎片文件

17 支持列剪枝支持条件过滤支持数据插入 EsSparkSQL.esDF(hc,indexName,esQuery).registerTempTable( testtable ) ElasticSearch Driver 获取 Query 需要的 Nodes 和 Shards Executor Executor Executor 数据 : 单条 20K 主机 :16C 32G 性能 :40k~50k 条 / 秒 / 每台 ES Node ES Node ES Node 适合场景 : 存储 doc 数据, 随机数据搜索问题 : 原生程序使用 HTTP 方式进行数据加载, 吞吐量很低需要修改为 Traffic 方式

18 Apache Phoenix 支持列剪枝支持条件过滤支持数据插入 sqlcontext.read.format("org.apache.phoenix.spark").options(map( table -> table, zkurl -> zookeeperurl)).load.registertemptable( testtable ) Driver 获取执行计划, 分解成多个 Task Executor Executor Executor 数据 : 单条 100b 5 字段主机 :8C 16G 性能 :80k-100k 条 / 秒 / 每台数据 : 单条 2K 20 字段主机 :8C 16G 性能 :30k-50k 条 / 秒 / 每台 Region 适合场景 : Server 实时处理中输入数据, 统计结果数据问题 : 需要根据需求仔细设计 Schema Region Server Region Server

19 MongoDB 支持列剪枝支持条件过滤支持数据插入 sqlcontext.read.format( com.stratio.datasource.mongodb ).options(options).load.registertemptable( testtable ) Driver 获取 Target 的 Sahrds Executor Executor Executor MongoDB Shard MongoDB Shard MongoDB Shard

20 Cassantra 支持列剪枝支持条件过滤支持数据插入 CassandraSQLContext val cscontext = new CassandraSQLContext(sc) sccontext. cassandrasql( xxxxxx ) Datastax 官方推出的 DataSource

21 我们的公司 GrowingIO 用我们的产品帮助企业大规模驱动业务增长大幅度增加分析效率用数据增长营收智能数据标记 Data 商业运营智能 Intelligence

22 整体架构 GrowingIO 的 Spark 实践 Web UI Query Service HBase 离线应用 Elastic Search 实时应用 Kafka

23 GrowingIO 的 Spark 实践实时计算 Kafka Json Table JDBC Table Spark Streaming 业务逻辑 SQL HBase Table ES Table HBase Elastic Search PostgreSQL 配置库

24 GrowingIO 的 Spark 实践离线计算 HBase HBase Table ES Table Spark Server 业务逻辑 SQL HBase Table Hive Table HBase Elastic Search HDFS PostgreSQL 配置库

25 坑 1 Elastic Search Elastic Search 数据查询当 Mapping 数据的列大于 Source 中的列时, 报 Index Out of Bound Exception 修改 RowValueReader 的 addtobuffer 方法 Elastic 数据加载默认通过 HTTP 的接口加载数据, 性能极差修改为 Transport 方式加载使得性能提升 2-3 倍性能优化需要详细设计 Index, 尽量减少每次查询的数据量

26 坑 2 Phoenix Spark 1.5 支持 DecimalType GenericMutableRow => InternalRow PHOENIX-2279 Limit 与 Union 相关的 BUG Maven 中 Hadoop 版本兼容性 Region Split 导致缓存中的 Region 信息失效 Phoenix JDBC Driver has been closed(yarn 资源控制 ) 读取数据时 Partition 数量过少导致读取速度慢

27 GrowingIO 的 Spark 实践总结 : 使用 Data Source API 带来好处存储成本降低 : 数据只存储一份开发成本降低 : 无需关心外部数据源的连接代码, 直接开发业务逻辑使用 Data Source API 需要注意的地方外部数据源设计优化 Elastic Search 的 Index 设计 HBase 的 Schema 设计 (Phoenix Bucket) 外部数据源的性能瓶颈执行顺序 Cache 外部 DataSource 的 BUG

28 GrowingIO 的 Spark 实践使用 Spark Streaming Kafka Direct API 模式数据分布不均的问题 Receiver 模式在计算量较大时需要通过 shuffle 将数据分发到多个节点计算占用 Yarn 资源的问题 Receiver 模式需要消耗 Yarn 的计算资源人工控制 offset 根据应用场景手工控制 offset 更加简单和安全下游系统采用容忍重复的设计原则保证 Streaming 程序可以随时重启运行

29 GrowingIO 的 Spark 实践优化 Spark Streaming 的调度机制原有调度机制下的问题定时器定时创建 job, 无视当前是否有 Job 正在运行所有的 Job 进行统一的流量控制当限速较大时, 会发生 Job 的积压当限速较小时, 处理性能又不高优化后的调度机制 JobScheduler 中每次创建 Job 后, 等待 Job 结束后再创建下一个 Job 优化后不会再有 Job 的积压, 限速可以设置的相对激进一些推荐的限速设置 : 最大处理能力的 80% Spark 1.5 中的 BackPressure

30 GrowingIO 的 Spark 实践 Spark Server 多个离线应用共享资源支持任务优先级设定 Spark App1 Spark Server Spark App2 Spark App3 S D K Spark Context Spark Cluster Spark AppN

31 感谢聆听

合集

Ver 1.0 版本目录第一章当大数据遇上 SSD 01 第二章广东移动运用 Hadoop 创新应用 04 第三章第四章第五章第六章第七章第八章第九章第十章如何利用大数据分析提升垃圾短信过滤效果广东电信用大数据重构室内网优大数据提升