中仑建设网站,98证书兼职网,专业网站建设品牌策划方案,山西建设网站目录
1. DataX介绍
2. DataX支持的常用数据源类型
3. 设计理念
4. DataX框架设计 4.1. Reader 4.2. Writer 4.3. Framework
5. DataX的运行流程
6. DataX与Sqoop对比
7. 部署
8. 配置详解
9. 案例 同步MySql到HDFS
9.1. 整体结构
9.2. mySqlReader
9.2.1. …目录
1. DataX介绍
2. DataX支持的常用数据源类型
3. 设计理念
4. DataX框架设计 4.1. Reader 4.2. Writer 4.3. Framework
5. DataX的运行流程
6. DataX与Sqoop对比
7. 部署
8. 配置详解
9. 案例 同步MySql到HDFS
9.1. 整体结构
9.2. mySqlReader
9.2.1. 使用tableMode
9.2.2. 使用QuerySQLMode
9.3. HDFSWriter
9.3.1. 在Hive中建表的时候指定表( NULL DEFINED AS )
9.3.2. 修改源码 点击参考
9.4. setting
9.5. 完整配置示例
9.6. 模拟传输
9.6.1. 启动hadoop和Hive创建Hive表
9.6.2. 启动传输任务
9.6.3. 查看数据
9.7. DataX传参
10. DataX参数优化
10.1. 速度控制
10.2. 内存调整 1. DataX介绍 DataX 是阿里巴巴开源的一个异构数据源离线同步工具致力于实现包括关系型数据库(MySQL、Oracle等)、HDFS、Hive、ODPS、HBase、FTP等各种异构数据源之间稳定高效的数据同步功能 源码地址: 点击进入 组件地址: 点击下载
2. DataX支持的常用数据源类型
类型数据源关系型数据库MySqlOracleSQLServerPostgreSQLNoSql数据存储HBase 0.94 / 1.1Phoenix 4.x / 5.xMongoDBHive无结构化数据存储TxtFileFTPHDFSElasticSearch 支持读不支持写
3. 设计理念 为了解决异构数据源同步问题DataX将复杂的网状的同步链路变成了星型数据链路DataX作为中间传输载体负责连接各种数据源。当需要接入一个新的数据源的时候只需要将此数据源对接到DataX便能跟已有的数据源做到无缝数据同步。 对于使用者,只需要学习DataX的数据源配置方式就可以将数据源里面的数据进行传输
4. DataX框架设计 DataX本身作为离线数据同步框架采用Framework plugin架构构建。将数据源读取和写入抽象成为Reader/Writer插件纳入到整个同步框架中 4.1. Reader 数据采集模块,负责采集数据源的数据,将数据发送给Framework 4.2. Writer 数据写入模块,负责不断向Framework取数据, 并将数据写入到目的端 4.3. Framework 用于连接reader和writer,作为两者的数据传输通道,并处理缓冲,控流,并发,数据转换等核心技术问题
5. DataX的运行流程 DataX采集全量数据的原理是通过sql查询的方式获取数据,查询语句会被切割,例如 通过时间查询,按照时间维度将数据切分成多个Task, DataX在传输数据的时候将启动TaskGroup,每个TaskGroup负责一定的并发度运行其所得的Task,单个TaskGroup的并发的固定为5,总TaskGroup数量与配置的总并发度有关: TaskGroup数量 总并发度 / 5
6. DataX与Sqoop对比
功能DataXSqoop运行模式单进程多线程MR分布式不支持,可以通过调度系统规避支持控流有需要定制开发统计信息已有一些统计,上报需要定制没有, 分布式数据数据不方便数据校验在core部分与校验功能没有, 分布式收集数据不方便监控需要定制需要定制
7. 部署
下载DataX安装包到服务器解压并测试运行
python datax_home/bin/datax.py datax_home/job/job.json 8. 配置详解 可以使用如下命名查看DataX配置文件模板
python bin/datax.py -r mysqlreader -w hdfswriter 配置文件模板如下json最外层是一个jobjob包含setting和content两部分其中setting用于对整个job进行配置content用户配置数据源和目的地
Reader和Writer的具体参数可参考官方文档点击查看 9. 案例 同步MySql到HDFS
9.1. 整体结构
{job: {content: [{reader: {},writer: {}}],setting: {speed: {channel: 1}}}
}
9.2. mySqlReader
9.2.1. 使用tableMode
{name: mysqlreader, //Reader名称 , 固定写法parameter: {username: root, //数据库用户密码password: 123456,connection: [{jdbcUrl: [jdbc:mysql://node1:3306/gmall],//数据库jdbc urltable: [tb] //数据库jdbc url}],column: [id, name, age], //同步的字段 [*]表示所有字段where: id0, //while过滤条件splitPk: //分片字段,如果没有这个字段,或者值为空,则只有一个Task}
}
9.2.2. 使用QuerySQLMode
{name: mysqlreader, //Reader名称 , 固定写法parameter: {username: root, //数据库用户密码password: 123456,connection: [{jdbcUrl: [jdbc:mysql://node1:3306/gmall],//数据库jdbc urltable: [select * from tb] //数据库jdbc url}]}
}
9.3. HDFSWriter
{name: hdfswriter, //Writer名称 , 固定写法parameter: {column: [{ //列信息,包括列名和类型 类型为Hive表字段类型,目前不支持decimal,binary,arrays,maps,struictsname: id,type: bigint}, {name: name,type: string}, {name: age,type: bigint}],defaultFS: hdfs://node2:8020, //HDFS文件系统namenode节点地址,不支持传HA的集群名path: /mydatax, //HDFS文件系统目标路径fileName: tb, //HDFS文件名前缀fileType: text, //HDFS文件类型compress: gzip, //HDFS压缩类型 text文件支持压缩gzip bzip2; orc文件支持压缩NONE SNAPPYfieldDelimiter: \t, //HDFS的分隔符writeMode: append //数据写入的模式 append 追加 ; nonConflict: 若写入目录有同名(前缀相同文件),报错}
}
注意: HFDS Writer并未提供nullFormat参数也就是用户并不能自定义null值写到HFDS文件中的存储格式。默认情况下HFDS Writer会将null值存储为空字符串而Hive默认的null值存储格式为\N。所以后期将DataX同步的文件导入Hive表就会出现问题。
解决方法有两种,任意一种都可以
9.3.1. 在Hive中建表的时候指定表( NULL DEFINED AS )
9.3.2. 修改源码 点击参考
9.4. setting
{speed: { //传输速度配置channel: 1 //并发数},errorLimit: { // 容错比例配置record: 1, //错误条数上限,超出则任务失败percentage: 0.02 //错误比例上限,超出则任务失败}
}
9.5. 完整配置示例
{job: {content: [{reader: {name: mysqlreader,parameter: {username: root,password: 123456,connection: [{jdbcUrl: [jdbc:mysql://node1:3306/yangxp],table: [tb]}],column: [id, name, age],where: id0,splitPk: }},writer: {name: hdfswriter,parameter: {column: [{name: id,type: bigint}, {name: name,type: string}, {name: age,type: bigint}],defaultFS: hdfs://node2:8020,path: /mydatax,fileName: tb,fileType: text,compress: gzip,fieldDelimiter: \t,writeMode: append}}}],setting: {speed: {channel: 1},errorLimit: {record: 1,percentage: 0.02}}}
}
9.6. 模拟传输
9.6.1. 启动hadoop和Hive创建Hive表
DROP TABLE IF EXISTS tb;
CREATE EXTERNAL TABLE tb
(id STRING COMMENT 编号,name STRING COMMENT 姓名,age STRING COMMENT 年龄
) COMMENT 年龄表ROW FORMAT DELIMITED FIELDS TERMINATED BY \tNULL DEFINED AS LOCATION /mydatax/;9.6.2. 启动传输任务
将配置内容写入mysql_to_hive.json配置文件到目录datax_home/job/下并启动任务
python datax_home/bin/datax.py datax_home/job/mysql_to_hive.json 9.6.3. 查看数据
数据被压缩后上传到Hive,直接在看会乱码 可以借助hive客户端查看表数据,或者使用hdfs的zcat查看
hdfs dfs -cat /mydatax/tb__af7eac41_a69e_4976_bb68_a98cd5d3a689.gz|zcat 9.7. DataX传参 通常情况下离线数据同步任务需要每日定时重复执行故HDFS上的目标路径通常会包含一层日期以对每日同步的数据加以区分也就是说每日同步数据的目标路径不是固定不变的因此DataX配置文件中HDFS Writer的path参数的值应该是动态的。为实现这一效果就需要使用DataX传参的功能。 DataX传参的用法如下在JSON配置文件中使用${param}引用参数在提交任务时使用-p-Dparamvalue传入参数值具体示例如下。
{name: hdfswriter,parameter: {...path: /mydatax/${dt},...}
} python datax_home/bin/datax.py -p-Ddt2023-03-04 datax_home/job/mysql_to_hdfs.json
10. DataX参数优化
10.1. 速度控制 DataX3.0提供了包括通道(并发)、记录流、字节流三种流控模式可以随意控制你的作业速度让你的作业在数据库可以承受的范围内达到最佳的同步速度。
参数说明job.setting.speed.channel 并发数 job.setting.speed.record 总record限速 job.setting.speed.byte 总byte限速 core.transport.channel.speed.record 单个channel的record限速默认为1000010000条/s core.transport.channel.speed.byte 单个channel的byte限速默认值1024*10241M/s
注意事项 1.若配置了总record限速则必须配置单个channel的record限速 2.若配置了总byte限速则必须配置单个channe的byte限速 3.若配置了总record限速和总byte限速channel并发数参数就会失效。因为配置了总record限速和总byte限速之后实际channel并发数是通过计算得到的
计算公式为: min(总byte限速/单个channel的byte限速总record限速/单个channel的record限速)
配置示例:
{core: {transport: {channel: {speed: {byte: 1048576 //单个channel byte限速1M/s}}}},job: {setting: {speed: {byte : 5242880 //总byte限速5M/s}},...}
}10.2. 内存调整 当提升DataX Job内Channel并发数时内存的占用会显著增加因为DataX作为数据交换通道在内存中会缓存较多的数据。例如Channel中会有一个Buffer作为临时的数据交换的缓冲区而在部分Reader和Writer的中也会存在一些Buffer为了防止OOM等错误需调大JVM的堆内存。 建议将内存设置为4G或者8G这个也可以根据实际情况来调整。 调整JVM xms xmx参数的两种方式一种是直接更改datax.py脚本另一种是在启动的时候加上对应的参数如下
python datax_home/bin/datax.py --jvm-Xms8G -Xmx8G datax_home/job/mysql_to_hdfs..json
