1. UU电玩城首页
  2. > 编程日记

一文带你看懂DataX离线同步工具(图文并茂，看完就懂)

丷江南南 • • 1年前 • 编程日记 • 阅读(1) • 违法举报

目录

  • 一. DataX简介
    • 1.1 DataX概述
    • 1.2 DataX支持的数据源
  • 二. DataX架构原理
    • 2.1 DataX设计理念
    • 2.2 DataX框架设计
    • 2.3 DataX运行流程
    • 2.4 DataX调度决策思路
    • 2.5 DataX 与 Sqoop 对比
  • 三. DataX使用
    • 3.1 DataX使用概述
      • 3.1.1 DataX任务提交命令
      • 3.1.2 DataX配置文件格式
    • 3.2 同步MySQL数据到HDFS案例
      • 3.2.1 MySQLReader之TableMode
      • 3.2.2 MySQLReader之QuerySQLMode
      • 3.2.3 DataX传参
    • 3.3 同步HDFS数据到MySQL案例
  • 四. DataX优化
    • 4.1 速度控制
    • 4.2 内存调整

一. DataX简介

1.1 DataX概述

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

源码地址：阿里云DataX源码

1.2 DataX支持的数据源

请添加图片描述

二. DataX架构原理

2.1 DataX设计理念

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

请添加图片描述

2.2 DataX框架设计

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

请添加图片描述

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

2.3 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的线程来完成同步工作。

2.4 DataX调度决策思路

举例来说，用户提交了一个 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会把Mysql中的1个表切分成1个Task。

2.5 DataX 与 Sqoop 对比

请添加图片描述

三. DataX使用

3.1 DataX使用概述

3.1.1 DataX任务提交命令

用户只需根据自己同步数据的数据源和目的地选择相应的Reader 和 Writer，并将 Reader 和 Writer 的信息配置在一个 json 文件中，然后执行如下命令提交数据同步任务即可：

请添加图片描述

3.1.2 DataX配置文件格式

使用如下命令查看 DataX 配置文件模板：

请添加图片描述

配置文件模板如下，json 最外层是一个 job，job 包含setting 和 content 两部分，其中setting 用于对整个 job 进行配置，content 用户配置数据源和目的地。

{
    "job": {
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "column": [],
                        "connection": [
                            {
                                "jdbcUrl": [],
                                "table": []
                            }
                        ],
                        "password": "",
                        "username": "",
                        "where": ""
                    }
                },
                "writer": {
                    "name": "hdfswriter",
                    "parameter": {
                        "column": [],
                        "compress": "",
                        "defaultFS": "",
                        "fieldDelimiter": "",
                        "fileName": "",
                        "fileType": "",
                        "path": "",
                        "writeMode": ""
                    }
                }
            }
        ],
        "setting": {
            "speed": {
                "channel": ""
            }
        }
    }
}

请添加图片描述

Reader 和 Writer 的具体参数可参考官方文档，地址如下：DataX配置

3.2 同步MySQL数据到HDFS案例

案例要求：同步 gmall 数据库中 base_province 表数据到 HDFS 的/base_province 目录

需求分析：要实现该功能，需选用 MySQLReader 和 HDFSWriter，MySQLReader 具有两种模式分别是 TableMode 和 QuerySQLMode，前者使用 table,column,where 等属性声明需要同步的数据；后者使用一条 SQL 查询语句声明需要同步的数据。

下面分别使用两种模式进行演示。

3.2.1 MySQLReader之TableMode

1）编写配置文件

（1）创建配置文件 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": "000000",
                        "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 值存储格式的逻辑。
二是在 Hive 中建表时指定 null 值存储格式为空字符串（‘’）,如：

DROP TABLE IF EXISTS 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 FIELDS TERMINATED BY '\t'
 NULL DEFINED AS ''
 LOCATION '/base_province/';

（3）Setting 参数说明

请添加图片描述

3）提交任务

（1）在 HDFS 创建/base_province 目录

使用DataX向 HDFS 同步数据时，需确保目标路径已存在：

请添加图片描述

登录HDFS-namenode节点查看目录生成情况:

请添加图片描述

（2）执行如下命令
请添加图片描述

4）查看结果

（1）DataX 打印日志

请添加图片描述

查看数据库原数据，发现原数据有34条，而DataX打印日志最终读出只有32条：

请添加图片描述

为什么出现上述情况呢？
原因是，我们的base_province.json配置文件中有where限定条件，该条件只筛选表中id>=3的数据记录，所以最终读出的才是32条！

请添加图片描述

（2）查看 HDFS 文件

请添加图片描述

请添加图片描述

注：此处需要用管道，借助zcat命令查看数据，否则会出现乱码！

3.2.2 MySQLReader之QuerySQLMode

1）编写配置文件

（1）新增配置文件 base_province_sql.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": "000000",
                        "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）执行如下命令

请添加图片描述

4）查看结果

（1）DataX 打印日志

请添加图片描述

（2）查看 HDFS 文件

请添加图片描述

可以看到MySQLReader的两种模式最终都达到了同样的效果，那么这两种mode有什么区别?
答：TableMode只能查询一张表，而QuerySQLMode可以通过join等操作关联查询多张表，并且可以完成更复杂的查询工作(通过修改json文件中reader模块中的querySql语句).

3.2.3 DataX传参

通常情况下，离线数据同步任务需要每日定时重复执行，故 HDFS 上的目标路径通常会包含一层日期，以对每日同步的数据加以区分，也就是说每日同步数据的目标路径不是固定不变的，因此 DataX 配置文件中 HDFS Writer 的path参数的值应该是动态的。为实现这一效果，就需要使用 DataX 传参的功能。

DataX 传参的用法如下，在 JSON 配置文件中使用${param}引用参数，在提交任务时使用-p"-Dparam=value"传入参数值，具体示例如下。

1）编写配置文件

（1）修改配置文件 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": "000000",
                        "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/${dt}",
                        "writeMode": "append"
                    }
                }
            }
        ],
        "setting": {
            "speed": {
                "channel": 1
            }
        }
    }
}

2）提交任务

（1）创建目标路径

请添加图片描述

（2）执行如下命令

请添加图片描述

注：此处通过-p"-Ddt=2023-06-19"命令来实现传入到指定目录下。

3）查看结果

请添加图片描述

3.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": "000000",
                        "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 参数说明

请添加图片描述

注：此处WriteMode有三种模式，分别如下:
insert into(insert) : 当要写入的Mysql表无主键时，为正常写入；当有主键时，一旦出现重复数据，会出现数据重复异常。
replace into(replace) : 要求写入的Mysql表必须有主键，且当主键存在重复时，会delete对应整行数据，然后再insert。
ON DUPLICATE KEY UPDATE(update) : 要求写入的Mysql表必须有主键，该模式是在原有数据记录的基础上作修改，而不做删除操作。

3）提交任务

（1）在 MySQL 中创建 gmall.test_province 表

DROP TABLE IF EXISTS `test_province`;
CREATE TABLE `test_province` (
 `id` bigint(20) NOT NULL,
 `name` varchar(20) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT 
NULL,
 `region_id` varchar(20) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT 
NULL,
 `area_code` varchar(20) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT 
NULL,
 `iso_code` varchar(20) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT 
NULL,
 `iso_3166_2` varchar(20) CHARACTER SET utf8 COLLATE utf8_general_ci NULL 
DEFAULT NULL,
 PRIMARY KEY (`id`)
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = 
Dynamic;

（2）执行如下命令

请添加图片描述

4）查看结果

（1）DataX 打印日志

请添加图片描述

（2）查看 MySQL 目标表数据

请添加图片描述

为什么DataX日志中读出记录总数为64，而实际写入Mysql中的数据只有32条呢？
由于DataX从HDFS中读取的时候，会从根目录依次读取其中文件与子目录，而2023-06-19子目录下又保存有一份与根目录下gz文件同样的文件，所以DataX实际读取的数据为2*32=64条：
请添加图片描述而又因为Writer写入的策略选用为replace，该策略会对重复的数据作覆盖(仅保留其中一份数据)，所以写入Mysql中的数据就只有32条:
请添加图片描述

四. DataX优化

4.1 速度控制

DataX3.0 提供了包括通道(并发)、记录流、字节流三种流控模式，可以随意控制你的作业速度，让你的作业在数据库可以承受的范围内达到最佳的同步速度。

关键优化参数如下：

请添加图片描述

注意事项：
1.若配置了总 record 限速，则必须配置单个 channel 的 record 限速(避免类似“数据倾斜”的现象出现)
2.若配置了总 byte 限速，则必须配置单个 channe 的 byte 限速(避免类似“数据倾斜”的现象出现)
3.若配置了总 record 限速和总 byte 限速，channel 并发数参数就会失效。因为配置了总record 限速和总 byte 限速之后，实际 channel 并发数是通过计算得到的：
计算公式为:
min(总 byte 限速/单个 channel 的 byte 限速，总 record 限速/单个 channel 的 record 限速)

4.2 内存调整

当提升 DataX Job 内 Channel 并发数时，内存的占用会显著增加，因为 DataX 作为数据交换通道在内存中会缓存较多的数据。例如 Channel 中会有一个 Buffer，作为临时的数据交换的缓冲区，而在部分 Reader 和 Writer 的中，也会存在一些 Buffer，为了防止 OOM 等错误，需调大JVM的堆内存。

建议将内存设置为 4G 或者 8G，这个也可以根据实际情况来调整。

调整 JVM xms xmx 参数的两种方式：一种是直接更改 datax.py 脚本；另一种是在启动的时候，加上对应的参数，如下：文章来源地址https://uudwc.com/A/wo388

python datax/bin/datax.py --jvm="-Xms8G -Xmx8G" /path/to/your/job.json

原文地址:https://blog.csdn.net/f986153489/article/details/131289275

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处： 如若内容造成侵权/违法违规/事实不符，请联系站长进行投诉反馈，一经查实，立即删除！

h
上一篇 2023年09月24日 04:45
下一篇 2023年09月24日 04:48