0
0

详解DataX及使用

DataX概述

简介

DataX 是阿里巴巴开源的一个异构数据源离线同步工具，致力于实现包括关系型数据库(MySQL、Oracle等)、HDFS、Hive、ODPS、HBase、FTP等各种异构数据源之间稳定高效的数据同步功能。

支持数据源

在这里插入图片描述

DataX架构原理

设计理念

为了解决异构数据源同步问题，DataX将复杂的网状的同步链路变成了星型数据链路，DataX作为中间传输载体负责连接各种数据源。当需要接入一个新的数据源的时候，只需要将此数据源对接到DataX，便能跟已有的数据源做到无缝数据同步。
在这里插入图片描述

框架设计

DataX本身作为离线数据同步框架，采用Framework + Plugin架构构建。将数据源读取和写入抽象成为Reader/Writer插件，纳入到整个同步框架中。
在这里插入图片描述

  • Reader：数据采集模块，负责采集数据源的数据，将数据发给FrameWork。
  • Writer：数据写入模块，负责不断的向FrameWork取数据，并将数据写入到目的端。
  • FrameWork：用于连接Reader和Writer，作为两者的数据传输通道，并处理缓存，流控，并发，数据转换等核心问题。

运行流程

用一个DataX作业生命周期的时序图说明DataX的运行流程、核心概念以及每个概念之间的关系。
在这里插入图片描述

  • Job：单个作业同步的作业，称为一个Job，一个Job启动一个进程。
  • Task：根据不同数据源切分策略，一个Job会切分为多个Task，Task是DataX作业的最小单元，每个Task负责一部分数据的同步工作。
  • TaskGroup：Scheduler调度模块对Task进行分组，每个Task组叫做一个Task Group，每个Task Group负责以一定的并发度运行其分得得Task，单个Task Group的并发度为5。
  • Reader->Channel->Writer：每个Task启动后，都会固定启动Reader->Channel->Writer的线程来完成同步工作。

调度决策思路

举个栗子
用户提交了一个DataX作业，并且配置了总的并发度为20，目的是对一个有100张分表的mysql数据源进行同步。DataX的调度决策思路是：

  1. DataX Job根据分库分表切分策略，将同步工作分成100个Task。
  2. 根据配置的总的并发度20，以及每个Task Group的并发度5，DataX计算共需要分配4个TaskGroup。
  3. 4个TaskGroup平分100个Task，每一个TaskGroup负责运行25个Task。

DataX与Sqoop对比
在这里插入图片描述

DataX部署

1）下载DataX安装包并上传到hadoop102的/opt/software
下载地址：http://datax-opensource.oss-cn-hangzhou.aliyuncs.com/datax.tar.gz
2）解压datax.tar.gz到/opt/module 

tar -zxvf datax.tar.gz -C /opt/module\

3）自检，执行如下命令

python /opt/module/datax/bin/datax.py /opt/module/datax/job/job.json

出现如下内容，则表明安装成功

……
2021-10-12 21:51:12.335 [job-0] INFO JobContainer -
任务启动时刻 : 2021-10-12 21:51:02
任务结束时刻 : 2021-10-12 21:51:12
任务总计耗时 : 10s
任务平均流量 : 253.91KB/s
记录写入速度 : 10000rec/s
读出记录总数 : 100000
读写失败总数 : 0

DataX上手

使用概述

  • 任务提交命令：用户需要根据同步数据的数据源和目的地选择相应的Reader和Writer，并将Reader和Writer的信息配置在一个json文件中，然后执行命令提交数据同步任务即可。python bin/datax.py path/to/your/job.json

配置文件格式

查看DataX配置文件模板 

python bin/datax.py -r mysqlreader -w hdfswriter

json最外层是一个job，job包含setting和content两部分，其中setting用于对整个job进行配置，content用户配置数据源和目的地。
在这里插入图片描述
Reader和Writer的具体参数参考官方文档
https://github.com/alibaba/DataX/blob/master/README.md
https://gitee.com/mirrors/DataX/blob/master/README.md
在这里插入图片描述

同步MySQL数据到HDFS

使用一个栗子来完成同步MySQL数据->HDFS的应用
要求：同步gmall数据库中base_province表数据到HDFS的/base_province目录
分析：需选用MySQLReader和HDFSWriter，MySQLReader具有两种模式分别是TableMode和QuerySQLMode，前者使用table，column，where等属性声明需要同步的数据；后者使用一条SQL查询语句声明需要同步的数据。

MySQLReader之TableMode
1）编写配置文件
（1）创建配置文件base_province.json

vim /opt/module/datax/job/base_province.json

（2）配置文件内容如下

    "job": {
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "column": [
                            "id",
                            "name",
                            "region_id",
                            "area_code",
                            "iso_code",
                            "iso_3166_2"
                        ],
                        "where": "id>=3",
                        "connection": [
                            {
                                "jdbcUrl": [
                                    "jdbc:mysql://hadoop102:3306/gmall"
                                ],
                                "table": [
                                    "base_province"
                                ]
                            }
                        ],
                        "password": "123456",
                        "splitPk": "",
                        "username": "root"
                    }
                },
                "writer": {
                    "name": "hdfswriter",
                    "parameter": {
                        "column": [
                            {
                                "name": "id",
                                "type": "bigint"
                            },
                            {
                                "name": "name",
                                "type": "string"
                            },
                            {
                                "name": "region_id",
                                "type": "string"
                            },
                            {
                                "name": "area_code",
                                "type": "string"
                            },
                            {
                                "name": "iso_code",
                                "type": "string"
                            },
                            {
                                "name": "iso_3166_2",
                                "type": "string"
                            }
                        ],
                        "compress": "gzip",
                        "defaultFS": "hdfs://hadoop102:8020",
                        "fieldDelimiter": "\t",
                        "fileName": "base_province",
                        "fileType": "text",
                        "path": "/base_province",
                        "writeMode": "append"
                    }
                }
            }
        ],
        "setting": {
            "speed": {
                "channel": 1
            }
        }
    }
}

2）配置文件说明
（1）Reader参数说明
在这里插入图片描述
（2）Writer参数说明
在这里插入图片描述
注意事项：

HFDS Writer并未提供nullFormat参数：也就是用户并不能自定义null值写到HFDS文件中的存储格式。默认情况下，HFDS Writer会将null值存储为空字符串（‘’），而Hive默认的null值存储格式为\N。所以后期将DataX同步的文件导入Hive表就会出现问题。

解决方案：
一是修改DataX HDFS Writer的源码，增加自定义null值存储格式的逻辑，参考https://blog.csdn.net/u010834071/article/details/105506580。
二是在Hive中建表时指定null值存储格式为空字符串（‘’），例如：

DROPTABLEIFEXISTS base_province;CREATE EXTERNAL TABLE base_province
(`id`         STRING COMMENT'编号',`name`       STRING COMMENT'省份名称',`region_id`  STRING COMMENT'地区ID',`area_code`  STRING COMMENT'地区编码',`iso_code`   STRING COMMENT'旧版ISO-3166-2编码，供可视化使用',`iso_3166_2` STRING COMMENT'新版IOS-3166-2编码，供可视化使用')COMMENT'省份表'ROW FORMAT DELIMITED FIELDSTERMINATEDBY'\t'NULL DEFINED AS''
    LOCATION '/base_province/';

（3）Setting参数说明
在这里插入图片描述
3）提交任务
（1）在HDFS创建/base_province目录
使用DataX向HDFS同步数据时，需确保目标路径已存在
（2）进入DataX根目录
（3）执行命令

python bin/datax.py job/base_province.json

4）查看结果
（1）DataX打印日志

（2）查看HDFS文件

MySQLReader之QuerySQLMode

1）编写配置文件
（1）修改配置文件base_province.json
（2）配置文件内容

    "job": {
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "connection": [
                            {
                                "jdbcUrl": [
                                    "jdbc:mysql://hadoop102:3306/gmall"
                                ],
                                "querySql": [
                                    "select id,name,region_id,area_code,iso_code,iso_3166_2 from base_province where id>=3"
                                ]
                            }
                        ],
                        "password": "123456",
                        "username": "root"
                    }
                },
                "writer": {
                    "name": "hdfswriter",
                    "parameter": {
                        "column": [
                            {
                                "name": "id",
                                "type": "bigint"
                            },
                            {
                                "name": "name",
                                "type": "string"
                            },
                            {
                                "name": "region_id",
                                "type": "string"
                            },
                            {
                                "name": "area_code",
                                "type": "string"
                            },
                            {
                                "name": "iso_code",
                                "type": "string"
                            },
                            {
                                "name": "iso_3166_2",
                                "type": "string"
                            }
                        ],
                        "compress": "gzip",
                        "defaultFS": "hdfs://hadoop102:8020",
                        "fieldDelimiter": "\t",
                        "fileName": "base_province",
                        "fileType": "text",
                        "path": "/base_province",
                        "writeMode": "append"
                    }
                }
            }
        ],
        "setting": {
            "speed": {
                "channel": 1
            }
        }
    }
}

2）配置文件说明
（1）Reader参数说明
在这里插入图片描述
3）提交任务
（1）清空历史数据
（2）进入DataX根目录
（3）执行命令

 python bin/datax.py job/base_province.json

4）查看结果
（1）DataX打印日志
（2）查看HDFS文件

DataX传参
通常情况下，离线数据同步任务需要每日定时重复执行，故HDFS上的目标路径通常会包含一层日期，以对每日同步的数据加以区分，也就是说每日同步数据的目标路径不是固定不变的，因此DataX配置文件中HDFS Writer的path参数的值应该是动态的。为实现这一效果，就需要使用DataX传参的功能。
在JSON配置文件中使用${param}引用参数，在提交任务时使用-p"-Dparam=value"传入参数
举个栗子：
1）编写配置文件
（1）修改配置文件base_province.json
（2）配置文件内容

    "job": {
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "connection": [
                            {
                                "jdbcUrl": [
                                    "jdbc:mysql://hadoop102:3306/gmall"
                                ],
                                "querySql": [
                                    "select id,name,region_id,area_code,iso_code,iso_3166_2 from base_province where id>=3"
                                ]
                            }
                        ],
                        "password": "123456",
                        "username": "root"
                    }
                },
                "writer": {
                    "name": "hdfswriter",
                    "parameter": {
                        "column": [
                            {
                                "name": "id",
                                "type": "bigint"
                            },
                            {
                                "name": "name",
                                "type": "string"
                            },
                            {
                                "name": "region_id",
                                "type": "string"
                            },
                            {
                                "name": "area_code",
                                "type": "string"
                            },
                            {
                                "name": "iso_code",
                                "type": "string"
                            },
                            {
                                "name": "iso_3166_2",
                                "type": "string"
                            }
                        ],
                        "compress": "gzip",
                        "defaultFS": "hdfs://hadoop102:8020",
                        "fieldDelimiter": "\t",
                        "fileName": "base_province",
                        "fileType": "text",
                        "path": "/base_province/${dt}",
                        "writeMode": "append"
                    }
                }
            }
        ],
        "setting": {
            "speed": {
                "channel": 1
            }
        }
    }
}

2）提交任务
（1）创建目标路径
（2）进入DataX根目录
（3）执行命令

python bin/datax.py -p"-Ddt=2020-06-14" job/base_province.json

3）查看结果

同步HDFS数据到MySQL

要求：同步HDFS上的/base_province目录下的数据到MySQL gmall 数据库下的test_province表。
分析：要实现该功能，需选用HDFSReader和MySQLWriter。
1）编写配置文件
（1）创建配置文件test_province.json
（2）配置文件内容如下

{
    "job": {
        "content": [
            {
                "reader": {
                    "name": "hdfsreader",
                    "parameter": {
                        "defaultFS": "hdfs://hadoop102:8020",
                        "path": "/base_province",
                        "column": [
                            "*"
                        ],
                        "fileType": "text",
                        "compress": "gzip",
                        "encoding": "UTF-8",
                        "nullFormat": "\\N",
                        "fieldDelimiter": "\t",
                    }
                },
                "writer": {
                    "name": "mysqlwriter",
                    "parameter": {
                        "username": "root",
                        "password": "123456",
                        "connection": [
                            {
                                "table": [
                                    "test_province"
                                ],
                                "jdbcUrl": "jdbc:mysql://hadoop102:3306/gmall?useUnicode=true&characterEncoding=utf-8"
                            }
                        ],
                        "column": [
                            "id",
                            "name",
                            "region_id",
                            "area_code",
                            "iso_code",
                            "iso_3166_2"
                        ],
                        "writeMode": "replace"
                    }
                }
            }
        ],
        "setting": {
            "speed": {
                "channel": 1
            }
        }
    }
}

2）配置文件说明
（1）Reader参数说明
在这里插入图片描述
（2）Writer参数说明
在这里插入图片描述
3）提交任务
（1）在MySQL中创建gmall.test_province表

DROPTABLEIFEXISTS`test_province`;CREATETABLE`test_province`(`id`bigint(20)NOTNULL,`name`varchar(20)CHARACTERSET utf8 COLLATE utf8_general_ci NULLDEFAULTNULL,`region_id`varchar(20)CHARACTERSET utf8 COLLATE utf8_general_ci NULLDEFAULTNULL,`area_code`varchar(20)CHARACTERSET utf8 COLLATE utf8_general_ci NULLDEFAULTNULL,`iso_code`varchar(20)CHARACTERSET utf8 COLLATE utf8_general_ci NULLDEFAULTNULL,`iso_3166_2`varchar(20)CHARACTERSET utf8 COLLATE utf8_general_ci NULLDEFAULTNULL,PRIMARYKEY(`id`))ENGINE=InnoDBCHARACTERSET= utf8 COLLATE= utf8_general_ci ROW_FORMAT = Dynamic;

（2）进入DataX根目录
（3）执行命令

python bin/datax.py job/test_province.json

4）查看结果
（1）DataX打印日志
（2）查看MySQL目标表数据

DataX优化

速度控制

DataX3.0提供了包括通道(并发)、记录流、字节流三种流控模式，可以随意控制作业速度，让作业在数据库可以承受的范围内达到最佳的同步速度。
关键优化参数：
在这里插入图片描述
注意事项：

1.若配置了总record限速，则必须配置单个channel的record限速
2.若配置了总byte限速，则必须配置单个channe的byte限速
3.若配置了总record限速和总byte限速，channel并发数参数就会失效。因为配置了总record限速和总byte限速之后，实际channel并发数是通过计算得到的：
计算公式为:min(总byte限速/单个channle的byte限速，总record限速/单个channel的record限速)

内存调整

当提升DataX Job内Channel并发数时，内存的占用会显著增加，因为DataX作为数据交换通道，在内存中会缓存较多的数据。
打个比方来说：Channel中会有一个Buffer，作为临时的数据交换的缓冲区，而在部分Reader和Writer的中，也会存在一些Buffer，为了防止OOM等错误，需调大JVM的堆内存。
建议将内存设置为4G或者8G，也可以根据实际情况来调整。
调整JVM xms xmx参数的两种方式：一种是直接更改datax.py脚本；另一种是在启动的时候，加上对应的参数：

python datax/bin/datax.py --jvm="-Xms8G -Xmx8G" /path/to/your/job.json
标签： 大数据
本文转载自: https://blog.csdn.net/weixin_45462732/article/details/124439299
版权归原作者 Keven He 所有， 如有侵权，请联系我们删除。
登录后发布评论

“详解DataX及使用”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？

文章导航