建网站视频,互联网行业现状分析,邯郸信息港恋家网,陕西建设网站电子政务大厅简介
POLARDB数据库是阿里云自研的下一代关系型云数据库#xff0c;100%兼容MySQL#xff0c;性能最高是MySQL的6倍#xff0c;但是随着数据量不断增大#xff0c;面临着单条SQL无法分析出结果的现状。X-Pack Spark为数据库提供分析引擎#xff0c;旨在打造数据库闭环100%兼容MySQL性能最高是MySQL的6倍但是随着数据量不断增大面临着单条SQL无法分析出结果的现状。X-Pack Spark为数据库提供分析引擎旨在打造数据库闭环借助X-Pack Spark可以将POLARDB数据归档至列式存储Parquet文件一条SQL完成复杂数据分析并将分析结果回流到业务库提供查询。本文主要介绍如何使用X-Pack Spark数据工作台对POLARDB数据归档。
业务架构
业务需要对多张表出不同纬度按天、按月的报表并对外提供查询服务最大表当前500G数据量还在不断的增加。尝试过spark直接通过jdbc去分析POLARDB一方面比较慢另外一方面每次扫全量的POLARDB数据对在线业务有影响。基于以下几点考虑选择POLARDBSpark的架构
选择POLARDB按天增量归档到spark列存每天增量数据量比较少选择业务低峰期归档对在线查询无影响选择Spark作为报表分析引擎因为Spark很适合做ETL且内置支持数据回流到POLARDB、MongoDB等多种在线库选择Spark离线数仓作为数据的中转站对于分析的结果数据回流到在线库提供查询能够一条Spark SQL完成分析不需要按维度值拆分多条分析SQL前置条件
1. 设置Spark访问POLARDB白名单
Spark集群和POLARDB需在同一个VPC下才能访问目前X-Pack Spark上还不支持一键关联POLARDB数据库需要将Spark集群的IP加到POLARDB白名单中。后续将会开放一键关联POLARDB的功能。 在“HBase控制台”-“集群列表”中找到分析Spark实例在“数据库连接”栏中找到“VSwitch ID”交换机ID如下图 然后在“专有网络VPC控制台”-交换机搜索交换机实例ID查询到IPV4网段。 将Spark集群网络加入到POLARDB白名单进入“控制台”-“集群列表”找到所要关联的POLARDB实例然后在“基本信息”-“访问信息”-“白名单”加入Spark集群所属网段。 2. 创建测试表
POLARDB中已经存在测试表如果没有可登录POLARDB数据库创建测试表下文也以该测试表为例。
CREATE TABLE IF NOT EXISTS test.us_population (state CHAR(2) NOT NULL PRIMARY KEY,city VARCHAR(10),population INTEGER, dt TIMESTAMP );INSERT INTO test.us_population VALUES(NY,New York,8143197, CURRENT_DATE );
INSERT INTO test.us_population VALUES(CA,Los Angeles,3844829, CURRENT_DATE);
INSERT INTO test.us_population VALUES(IL,Chicago,2842518, 2019-04-13);
INSERT INTO test.us_population VALUES(TX,Houston,2016582, 2019-04-14);
INSERT INTO test.us_population VALUES(PA,Philadelphia,1463281, 2019-04-13);
INSERT INTO test.us_population VALUES(AZ,Phoenix,1461575, 2019-04-15);
INSERT INTO test.us_population VALUES(SA,San Antonio,1256509, CURRENT_DATE);
INSERT INTO test.us_population VALUES(SD,San Diego,1255540, CURRENT_DATE);
INSERT INTO test.us_population VALUES(DL,Dallas,1213825, 2019-04-15);
INSERT INTO test.us_population VALUES(SJ,San Jose,912332,2019-04-15);
一、使用交互式工作台归档数据(调试、测试)
创建Spark运行会话
在HBase控制台-会话管理创建会话指定会话名称和执行集群如图 在编辑器中输入Spark启动参数并运行会话以便在交互式查询中使用。
--driver-memory 1G
--driver-cores 1
--executor-cores 1
--executor-memory 2G
--num-executors 1
--name spark_on_polardb
--jars /spark-demo/mysql-connector-java-5.1.34.jar
参数说明
参数说明driver-memoryspark运行driver内存大小driver-coresspark运行driver核数executor-coresspark作业执行器executor核数executor-memory执行器内存jarsspark作业依赖第三方包地址可在资源管理中复制
注上述参数在测试环境中给定偏小大数据量时根据实际集群规格和数据量进行配置
会话运行成功后如下图所示
交互式查询归档数据
创建Spark映射POLARDB表
进入HBase控制台-交互式查询在会话列表中选择上一步创建会话“spark_on_polardb”然后新建查询指定查询名称选择查询类型为“SQL”类型如图
在查询输入框中输入Spark建表语句与POLARDB表进行关联建表语句为
create table spark_polordb
using org.apache.spark.sql.jdbc
options (driver com.mysql.jdbc.Driver,url jdbc:mysql://pc-xxx.rwlb.rds.aliyuncs.com:3306,dbtable test.us_population,user xxx,password xxxxxx
)
参数说明
参数说明spark_polordbspark中表名driverpolardb驱动类名urlpolardb的数据库连接地址dbtable对应polardb表名格式为database.tablenameuserpolardb用户名password连接密码
点击运行查询状态为“success”时表明创建成功。
查询测试
在上步创建查询编辑器中输入查询语句然后运行
SELECT * FROM spark_polordb
查询成功后返回结果如图
创建归档表
X-Pack Spark将POLARDB数据归档至Parquet列式存储格式中一方面能够获取更优的压缩空间另一方面后续分析任务中具有更高的效率。 Spark创建parquet分区表语句如下同样在第一步中交互式查询编辑中输入
CREATE table parquetTable(state CHAR(2), city VARCHAR(10), population int)
USING parquet
PARTITIONED BY(dt timestamp)
参数说明
参数说明parquetTablespark中归档表名USING parquet数据存储格式为parquetPARTITIONED BY按照字段分区类型为timestamp也可以指定为date
建表成功后可以将POLARDB数据写入至Parquet表。
归档数据
将POLARDB数据查询出写入parquet表即可完成数据归档操作语句为
INSERT INTO parquetTable partition(dt) SELECT state, city, population, dt FROM spark_polordb
运行成功后数据归档完成。查询parquet表数据
二、工作流调度周期归档(生产T1归档)
交互式查询主要用来测试调试归档一般需要做t1的操作每天定期把当前的数据做归档这就需要使用工作流的周期调度下面具体介绍如何使用工作流的周期调度实现t1的归档。
归档代码编写
使用工作流之前需要创建对应的Spark作业Spark归档POLARDB可以实现一个完整作业包括以下流程
在Spark中创建POLARDB表映射表(前提POLARDB中表已经存在)创建Spark分区归档表将数据写入归档表
云Spark提供了Spark归档POLARDB的代码DEMO请参考githubSparkArchivePolarDB 具体归档代码需结合实际场景归档不同表设置特定分区和归档条件等。
上传Spark归档作业资源
将打成jar包的spark归档demo代码通过资源管理上传至资源列表jar包下载地址Spark归档工具DEMO下载 自己编写的Spark作业同样需要打成jar包后上传至资源列表后面作业需要运行jar包中归档作业。
创建Spark作业
进入“HBase控制台”-数据工作台-“作业管理”-“创建作业”, 如图
编辑作业内容
作业内容中主要指定了Spark作业运行参数以及具体的归档作业编码类和传入参数等以SparkArchivePolarDB demo为例
--class com.aliyun.spark.polardb.SparkOnPolarDBArchive
--driver-memory 1G
--driver-cores 1
--executor-cores 1
--executor-memory 2G
--num-executors 1
--jars /spark-demo/mysql-connector-java-5.1.34.jar
/spark-demo/spark-examples-0.0.1-SNAPSHOT.jar
pc-xxx.rwlb.rds.aliyuncs.com:3306 test.us_population username passwd sparkTestPolarDB
参数说明
参数说明class指定spark作业运行主类/spark-demo/spark-examples-0.0.1-SNAPSHOT.jarspark作业所属包pc-xxx.rwlb.rds.aliyuncs.com:3306polardb的连接串test.us_population归档polardb表usernamepolardb用户名passwdpolardb连接密码sparkTestPolarDBspark归档表名
其余参数可参见上述章节介绍 作业配置如图
运行作业并查看结果
作业运行后一段时间可以查看到运行状态成功后可在交互式查询中查看归档表数据。 进入交互式工作台使用可参考上述介绍查看归档表数据
配置工作流
进入“HBase控制台”-“数据工作台”-“工作流”选择新建工作流指定工作流名称、描述和执行集群, 然后进入工作流设计工作台拖动Spark作业并进行配置选择上一步配置作业并连线 选择工作流配置-调度属性开启调度状态并设置其实时间和调度周期工作流即将进行周期性调度如图
三、归档方式(产出表的形式)
全量归档
全量归档方式主要用来对原库中历史数据进行归档或者针对数据量比较小的表归档步骤如下
使用Spark的jdbc datasource创建POLARDB的映射表在Spark中创建相同表结构的归档表归档表使用Parquet列式存储能够最大化节约存储空间并加速分析性能通过映射表读取POLARDB数据并写入Spark归档表注意写入时保证字段顺序一致。创建归档表时如果表数据量较大可以创建分区表。分区策略一般分为时间分区和业务分区 时间分区易于使用即将相同时间的数据归档到同一个目录比如选择按年或者按天进行时间分区在分析时限定数据分区即可过滤掉与分析任务无关的数据。业务分区字段需要具有有限的类别比如性别、年龄、部门等。业务分区需要结合具体业务进行考虑分区个数不宜过多spark默认最大分区数为1000。分区方式可以选择静态分区和动态分区默认使用静态分区即写入数据时必须指定写入哪个分区动态分区需要将hive.exec.dynamic.partition.mode设置为nonstrict写入时根据具体分区字段值动态创建分区相同partition key值写入同一个分区。使用示例可参考SparkOnPolarDBArchivedemo
增量归档
业务数据仅增量
在业务表中数据不存在更新和删除的操作仅仅是向数据表中增量写入这种情况下只需要在数据表中记录数据入库时间或者其他标记记录新增数据在Spark中使用工作流周期调度传入增量数据条件定期将新增数据归档只Spark中即可。
业务数据更新
针对业务数据存在更新的数据如果原表中无法辨别更新的数据目前只能通过全量归档的方式每次对全量数据进行一次归档将原归档表数据进行overwrite如果存在更新数据标记如update_time字段由于Spark目前不支持ACID无法使用merge..into功能直接更新已有数据增量更新归档步骤如下
设置更新增量数据选择条件(归档表全量归档时已创建)如update_time大于某个日期抽取增量更新的数据写入spark临时表将历史数据归档表与增量更新数据表进行left out join并过滤出增量表字段为空的数据表示历史数据中未参与增量更新的数据然后与增量更新的数据进行union合并写入Spark临时表将临时表数据覆盖写入到归档表中作为新的归档数据参与后续业务分析。
Spark更新增量归档目前只能使用join关联方式遍历所有数据完成数据更新但好处是尽量避免影响在线库POLARDB的数据访问每次只读取更新和增量的部分数据将计算工作放在廉价的Spark集群中。 使用示例可参考SparkOnPolarDBIncrement 另一种方式如果在业务侧需要保留多个版本更新的数据可以直接将更新和增量的数据追加到归档表中然后在业务侧通过最新时间判断出有效的数据可以避免每次更新时复杂计算过程。
业务数据更新删除
业务表中如果存在delete目前Spark没有较好的办法进行支持需要在业务库记录删除的关键字段信息与归档表进行join过滤掉join到的数据然后覆写到归档表中达到delete的效果。
总结
在进行实际数据开发中往往需要多个Spark作业配合完成数据归档以及分析工作单个工作流中支持配置多个作业并按序执行同时配合交互式工作台进行数据验证减少很多开发中不便。目前工作台仍在不断优化中在使用中遇到不便之处可随时提出建议便于简化您的数据开发工作。后续X-Pack Spark将提供一键归档功能敬请期待。
原文链接 本文为云栖社区原创内容未经允许不得转载。
