Lenovo Data Platform

Size: px

Start display at page:

Download "Lenovo Data Platform"

政宇殷
5 years ago
Views:

1 扩展 Spark 引擎支持 MPP 计算场景替换大规模企业级传统数据仓库张成松 May 12, 2017

2 个人简介 -- 张成松联想大数据研发总监联想大数据平台首席架构师 ( 从无到有, 全球化大数据平台,9 IDC 节点 ) 互联网 & 开源技术爱好者非专业驴友 2

3 演讲内容一个故事引出的业务场景企业级数据仓库 + 大数据, 如何碰撞出火花? 来点干货, 如何扩展 Spark 支持 MPP 计算场景? Spark MPP 在联想大数据企业级分析平台中的应用 3

4 一个故事引出的业务场景 4

5 业务在增长, 数据越来越多数据查询数据分析, 越来越慢, 经常跑个半小时也出不来 The Story of Data Analysis T+1 的数据处理太慢, 我想下一秒就知道业务有没有问题 Oracle SQL Server 数据库, 存不下了, 维护成本和费用也越来越高我不懂大数据, 数据大了就是大数据吧? 除了传统业务数据, 智能 sensor 穿戴式设备数据, 非结构化数据越来越多 5

6 传统企业级数据仓库向大数据平台转型解决传统数据应用问题解决大数据计算问题解决数据实时性处理问题历史数据迁移与传统数据仓库 & 应用工具无缝集成适配原有业务处理逻辑, 支持数据 CRUD 支持存储过程 PB 级数据计算结构化 & 非结构化数据存储应用大数据场景数据分析大数据环境下的数据治理实时数据采集流式计算实时计算交互探索式多维数据分析 6

7 企业传统数据仓库 + 大数据, 如何碰撞出火花? 7

8 企业级大数据平台实施方案两套架构混合架构一套架构, 数据共享 MPP+Hadoop 变种 Hadoop, 融合 MPP Hadoop/Spark for MPP 8

9 方案一 : MPP+Hadoop Business Layer OLTP OLAP Data Warehouse Batch calculate Data Mining ORDBMS MPP-DB hadoop ecosystem Spark Postgres XL OODB Hive HDFS 两套系统, 数据独立存储, 集群间数据同步 9

10 方案二 : 变种 Hadoop, 融合 MPP Business Layer OLTP OLAP Data Warehouse Batch calculate Data Mining Calculate Scheduler ORDBMS MPP-DB hadoop ecosystem Spark OODB Postgres -XL Hive HDFS 两套架构融合, 数据各自存储, 抽象存取 & 计算服务, 按需选择计算引擎 10

11 方案三 : Hadoop/Spark for MPP Business DB Business Layer MPP- DB OLTP OLAP Data Warehouse Batch calculate Data Mining ORDBM S Spark & Spark MPP Hive OODB Postgre s XL HDFS 一套系统, 同时支持大数据计算和传统数据仓库, 导入历史数据后, 数据统一存储 11

12 For 方案 3, 如何扩展 Spark 支持 MPP 计算场景? 12

13 Spark MPP : 系统架构 WEB-UI Client API JDBC ODBC SHELL SQL Server 语法解析 SQL PARSE Spark MPP SQL Engine Oracle 语法解析 Transaction Mgr. SQL PARSE 事务并发控制存储过程语法解析变量函数游标多租户 & 资源管理用户权限管理 SQL ANALYZER SQL ANALYZER 分布式 CRUD 逻辑执行 SQL OPTIMIZER SQL OPTIMIZER 并行执行计算资源管理 SPARK 执行引擎数据安全控制 HDFS 13

14 Spark MPP : 实现数据 CRUD 原理 HDFS 原始存储结构 Spark MPP 方案 HDFS 存储结构上次事务, 数据合并后的存储目录新的事务, 变化数据存储目录数据变化时记录增加将新增记录直接写入新的 delta_$transid_x 目录内 bucket 文件 ; 记录更新将更新内容做为新记录, 写入新的 delta_$transid_x 目录内 bucket 文件 ; 记录删除将删除内容做为 null 记录, 写入新的 delta_$transid_x 目录内 bucket 文件 ; 每条记录增加的信息 : { 操作类型, 原始事务 id, rowid, 当前的事务 ID } 周期性 Compaction 新增文件数量大小监控, 当达到一定数据时自动合并定时控制, 周期性合并仅对特定格式的 ORC 表有事务处理功能, 事务处理的主要操作为增删改 ; 操作时开启事务模式,SET transaction.type=sparkmpp; ORC 表必须是分桶表, 在表属性里需要加上 TBLPROPERTIES ( transactional = true ), 以标识这是一个要用作事务操作的表 ; 14

15 Spark MPP :insert update delete CREATE TABLE user_info (name STRING, age INT, sex STRING) CLUSTERED BY (name) INTO 1 BUCKETS STORED AS ORC TBLPROPERTIES ("transactional"="true"); Sample: 1. insert into user_info values( Lilei, 6, M ) ;-- 事务 id:1 2. update user_info set age = 7 where name = Jim ;-- 事务 id:2 3. delete user_info where name = Lucy ;-- 事务 id:3 数据存储 : hdfs://warehouse/user_info/base_0/bucket_0001 Name Age Sex extra_info 数据操作 Lucy 5 F { } Jim 6 M { } 数据合并后 hdfs://warehouse/user_info/base_3/bucket_0001 Name Age Sex extra_info { } Jim 7 M { } Lilei 6 M { } 新增数据存储 : hdfs://warehouse/user_info/delta_1/bucket_0001 Name Age Sex extra_info Lilei 6 M { } hdfs://warehouse/user_info/delta_2/bucket_0001 Name Age Sex extra_info Jim 7 M { } hdfs://warehouse/user_info/delta_3/bucket_0001 Name Age Sex extra_info { } 构造新记录, 写入文件查询出来最大 transaction_id 的整条数据, 修改字段后, 写入文件构造新记录 ( 字段 null), 写入文件 15

16 Spark MPP :select 实现逻辑原始文件存储 : hdfs://warehouse/user_info/base_1/bucket_0001 Name Age Sex extra_info Lucy 5 F { } Jim 6 M { } 新增文件存储 : hdfs://warehouse/user_info/delta_2/bucket_0001 Name Age Sex extra_info Lilei 6 M { } hdfs://warehouse/user_info/delta_3/bucket_0001 Name Age Sex extra_info Jim 7 M { } hdfs://warehouse/user_info/delta_4/bucket_0001 Name Age Sex extra_info { } 数据查询 (select) 处理逻辑 : 1. rowsets = sort([ tid asc, rowid asc, max(ctid) ], data ) -- 从 base 和所有 delta 文件中, 查询符合 where 条件的, transaction_id 最大的数据集 ; 2. foreach row in rowsets -- 构造返回结果集 2.1 if row.optype = insert or update : read_lines += row -- 如果操作标志为 inser or update, 把记录写入返回结果集 ; 2.2 if row.optype = delete : continue -- 如果操作标志为 delete, 继续 ; 3. return read_lines -- 返回构造的结果集 ; 16

17 Spark MPP : 测试结果 Insert 测试 : Update 测试 : Delete 测试 : 17

Spark MPP : 存储过程 &SQL 语句执行逻辑选择 SQL 解析器 SQL SERVER SQL 解析过程语法分析 (SqlParse) 词法分析 (Lexer) g4 语法规则 Antlr4 SQL Server 语法文件 Executor Engine( 迭代遍历 TreeNode) SQL 执行过程用户输入 SQL MYSQL AST( 抽象语法树 )

18 Spark MPP : 存储过程 &SQL 语句执行逻辑选择 SQL 解析器 SQL SERVER SQL 解析过程语法分析 (SqlParse) 词法分析 (Lexer) g4 语法规则 Antlr4 SQL Server 语法文件 Executor Engine( 迭代遍历 TreeNode) SQL 执行过程用户输入 SQL MYSQL AST( 抽象语法树 ) Visitor(rewrite/ transform) 执行 Node 预定义规则函数游标存储过程定义变量 Rebulid Resultset ORACLE 语法树 (TreeNode) 提交 DDL/DML SQL Spark 可以识别的 SQL 语句, DDL,DML. check/optimize SPARK Resultset xml 自定义格式可执行语法树 (TreeNode) Hive meta 库 18

19 Spark MPP : 创建存储过程 drop PROCEDURE test_a_out GO CREATE PROCEDURE VARCHAR(50) = 'default_value' int = 27 AS 支持 Input/output/return 参数传递 = 'zhong' INSERT INTO b1(name) VALUES(@name) = 'deyin' UPDATE b1 SET name where name = 'zhong' DELETE from b1 where name return go 语法校验 g4 文件语法解析构造语法树 insert root proc sql_clauses update varchar(50) = 'nm_value' varchar(50) = OUTPUT, 1 go 存储过程是否存在? ( 名称 /MD5) optimize 数据持久化优化后的执行语法树 root cache 载入内存持久化存储 TreeNode proc sql_clauses DB insert update 19

20 Spark MPP : 执行存储过程 varchar(50) = 'nm_value' varchar(50) = OUTPUT, 1 go 语法校验 cache DB Produce Name load 语法树执行 SP 函数体执行过程中的变量结果状态等获取可执行的语法树 root 收集执行结果, 并返回恢复现场 Insert 节点 update 节点保存现场 proc sql_clauses insert update 20

21 Spark MPP : 变量变量隔离 : GO 存储过程函数体与外部变量系统变量表变量变量临时表 ADD DECLARE SELECT PRINT QUERY VALUE EXECUTE PROCEDURE 保存变量现场标量变量容器存储过程游标 UPDATE VALUE 变量存储区域 SET DELETE GO Deallocate 恢复变量现场 EXEC END 变量拷贝区域 21

Spark MPP : 函数 Spark Executor select len( abc ); Spark 内置函数语法解析 Spark Driver Function Node UDF 用户输入 SQL g4 语法规则 Scalar Function Others Function 选择 Calculator Transform 函数名参数转换 Calculator Check

22 Spark MPP : 函数 Spark Executor select len( abc ); Spark 内置函数语法解析 Spark Driver Function Node UDF 用户输入 SQL g4 语法规则 Scalar Function Others Function 选择 Calculator Transform 函数名参数转换 Calculator Check Arguments print len( abc ); Do computing 注 : 语法解析后, 识别 TreeNode 为 Function: 1 Spark 内置函数,Spark Driver 提交 Executor 执行, 直接返回结果 ; 2 非 Spark 内置函数, 或与 Spark 内置函数冲突, 在 Spark Driver 中选择函数计算器, 通过 UDF 形式, 调用函数的计算器 ; Return Result 22

23 Spark MPP : 游标 Examples: int DECLARE cursor_name SCROLL CURSOR FOR select id,name,age from tmp.tb_cursor OPEN cursor_name FETCH next FROM @age insert into = 0 BEGIN FETCH next FROM @age % 2 <> 0 begin insert into end end close cursor_name deallocate cursor_name go select * from tmp.tb_cursor_result Declare Open Fetch Close Deallocate 1. Global/Local/Variable 2. Save to Variable Container 1. Find cusor by name 2. Execute cursor(select statement) 3. Save Resultset to cursor 1. Find cusor by name 2. Move pointer in cursor result 3. Fetch row from resultset 4. Assign value to into variables 1. Set the cursor status to CLOSE 2. Clear cursor resultset 1. Delete cursor from Variable Container 23

24 Spark MPP : 测试结果 24

25 优化 Spark MPP, 不断提升性能实现了数据的 CRUD 操作, 支持存储过程游标函数变量等逻辑, 仅仅是 Spark 架构实现 MPP 功能的第一步为了提升执行效率和性能, 还需要从不同层面对 Spark MPP 引擎进行优化并发执行去冗余列执行计划优化消除无效节点复杂事务合并相似节点, 减少 INPUT DATA 更多 DB 类型支持系统函数 SQL 引擎优化规则优化计算成本优化存储过程优化代码生成 25

26 联想大数据企业级分析平台 (LEAP) 26

27 联想大数据企业级分析平台解决方案 (LEAP) 业务分析套件 Nash 数据能力开放平台 Gauss Big Data as a Service 资源开放算法武库数据工厂 Riemann Bayes Fourier 大数据计算平台 Descartes 大数据技术整合与深度优化数据采集转换套件 Euclid 数据资产管理平台 Euler 系统运维监控中心 Shannon 27

28 联想大数据企业级分析平台解决方案 (LEAP) 业务层数字化营销智能供应链舆情分析预防性维护客户洞察产品管理财务管理服务层分析层整合层获取层 8 实时能交力互查式询分析能交互力分析套件算法武库大数据平台分布式存储系统终端采集业务组件搜索与查询可视化分析第三方工具算法模型业务模型挖掘工具机器学习分析挖掘能力 2 网络爬虫 1 应用开发与可视化能力 3 4 MPP HDFS 数据工厂分布式计算框架统一资源调度管理采集与转换能力数据采集转换批量导入数据产品数据 API 数据审查数据服务管理 No SQL 内存计算流计算图计算流式采集数据资产管理元数据数据模型数据标准数据质量数据安全与隐私数据生命周期系统运维监控大数据整合与计算能力数据产品化能力数据资产管理能力监控告警一安全管理站式运维能力用户管理性能管理容灾管理集群管理自动化部署源数据音频视频传感器文本社交网络设备日志 CRM SCM ERP 28

29 联想大数据具有强大的研发和实施能力三个研发中心 200+ 名大数据研发工程师 60+ 名大数据平台运维工程师北京成都香港 40+ 名数据科学家来自中科院清华北大牛津港大港科大以及美国澳洲等著名学府的博士和硕士人才博士与海归比例超过 80% 在顶级期刊和会议中发表学术论文近百篇, 国内和美国专利数十余项 6 年! 300+ 研发人员持续投入 20+ 名大数据领域专家数据专家 : 平台架构数据架构数据标准数据治理数据管控等业务专家 : 制造业零售业政府医疗行业能源与公共事业通讯业金融业等 30 30

30 全球部署的超大规模集群,PB 级与复杂业务生态的实战锤炼法兰克福埃森全球化多中心部署, 2000 台服务器, 3000 名操作用户旧金山总容量 12PB, 数据总量 10PB 日新增数据 30TB 日处理数据 4.3 PB 99.5% 的全球高可用性芝加哥弗吉尼亚新加坡香港北京天津在实践中充分验证系统的高可靠性企业数据本地化收集和存储完全合规各国数据保护和隐私保护法律联想大数据平台, 支持 2 亿用户的数据计算, 每天 2 万个以上的计算任务, Lenovo ID 和乐商店支持了 1 亿以上的用户数, 以及多个支持百万以上用户数的应用和服务每隔 5 秒, 联想全球的新增设备信息会被实时提取和计算, 实时优化全球业务流程 31

untitled

untitled OO 1 SQL Server 2000 2 SQL Server 2000 3 SQL Server 2000 DDL 1 2 3 DML 1 INSERT 2 DELETE 3 UPDATE SELECT DCL 1 SQL Server 2 3 GRANT REVOKE 1 2 1 2 3 4 5 6 1 SQL Server 2000 SQL Server SQL / Microsoft SQL