5、DataX（DataX简介、DataX架构原理、DataX部署、使用、同步MySQL数据到HDFS、同步HDFS数据到MySQL）

news2024/11/17 7:36:19

1、DataX简介

1.1 DataX概述

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

源码地址：https://github.com/alibaba/DataX

1.2 DataX支持的数据源

DataX目前已经有了比较全面的插件体系，主流的RDBMS数据库、NOSQL、大数据计算系统都已经接入，目前支持数据如下图。
在这里插入图片描述

2、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运行流程

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

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。

2.5 DataX和Sqoop对比

在这里插入图片描述

3、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

4、出现如下内容，则表明安装成功
在这里插入图片描述

4、DataX使用

4.1 DataX使用概述

4.1.1 DataX任务提交命令

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

 python bin/datax.py path/to/your/job.json
4.1.2 DataX配置文件格式

可以使用如下命名查看DataX配置文件模板。

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

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

4.2 同步MySQL数据到HDFS案例

案例要求：同步gmall数据库中base_province表数据到HDFS的/base_province目录
需求分析：要实现该功能，需选用MySQLReader和HDFSWriter，MySQLReader具有两种模式分别是TableMode和QuerySQLMode，前者使用table，column，where等属性声明需要同步的数据；后者使用一条SQL查询语句声明需要同步的数据。
下面分别使用两种模式进行演示。

4.2.1 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": "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表就会出现问题。
（3）Setting参数说明
在这里插入图片描述
3、提交任务
（1）在HDFS创建/base_province目录
使用DataX向HDFS同步数据时，需确保目标路径已存在

hadoop fs -mkdir /base_province

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

 python bin/datax.py job/base_province.json

4、查看结果
（1）DataX打印日志
在这里插入图片描述
（2）查看HDFS文件

hadoop fs -cat /base_province/* | zcat
4.2.2 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": "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）清空历史数据

 hadoop fs -rm -r -f /base_province/*

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

python bin/datax.py job/base_province.json

4、查看结果
（1）DataX打印日志
在这里插入图片描述
（2）查看HDFS文件

hadoop fs -cat /base_province/* | zcat
4.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": {
                        "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/${dt}",
                        "writeMode": "append"
                    }
                }
            }
        ],
        "setting": {
            "speed": {
                "channel": 1
            }
        }
    }
}

2、提交任务
（1）创建目标路径

 hadoop fs -mkdir /base_province/2020-06-14

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

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

3、查看结果

hadoop fs -ls /base_province

4.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参数说明
在这里插入图片描述
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）进入DataX根目录
（3）执行如下命令

 python bin/datax.py job/test_province.json

4、查看结果
（1）DataX打印日志
（2）查看MySQL目标表数据
在这里插入图片描述

5、DataX优化

5.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限速)

5.2 内存调整

当提升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

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/700466.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

SpringBoot3.0整合RocketMQ时出现未能加载bean文件

SpringBoot3.0整合RocketMQ时出现未能加载bean文件

SpringBoot3.0整合RocketMQ时出现未能加载bean文件 问题 APPLICATION FAILED TO START Description: Field rocketMQTemplate in com.spt.message.service.MqProducerService required a bean of type ‘org.apache.rocketmq.spring.core.RocketMQTemplate’ that could not …
阅读更多...
Scala里的WordCount 案例

Scala里的WordCount 案例

7.7.5 普通 WordCount 案例 package chapter07object TestWordCount__简单版 {def main(args: Array[String]): Unit {//单词计数：将集合中出现的相同单词计数，进行计数，取计数排名的前三的结果val stringList List("Hello Scala Hbas…
阅读更多...
2023年前端面试汇总-React

2023年前端面试汇总-React

1. 组件基础 1.1. React事件机制 <div onClick{this.handleClick.bind(this)}>点我</div> React并不是将click事件绑定到了div的真实DOM上&#xff0c;而是在document处监听了所有的事件&#xff0c;当事件发生并且冒泡到document处的时候&#xff0c;React将事…
阅读更多...
C语言进阶--动态内存管理

C语言进阶--动态内存管理

目录 一.为什么使用动态内存分配&#xff1f; 二.动态内存函数 2.1.malloc和free malloc函数 free函数 2.2.calloc和realloc calloc函数 realloc函数 三.常见的动态内存错误 对NULL指针的解引用 对动态开辟空间的越界访问 对非动态开辟内存使用free释放 使用free释…
阅读更多...
新型进网许可标志规格样式及申请使用流程说明

新型进网许可标志规格样式及申请使用流程说明

新型进网许可标志规格样式及申请使用流程说明 一、新型进网许可标志规格样式 (一)标志样式 新型进网许可标志分为彩色样式和黑白样式&#xff0c;如图1所示&#xff0c;生产企业可以自行选择使用。 (二)标志要素 新型进网许可标志由许可标识、设备型号、数字编码等要素组成…
阅读更多...
13年测试老鸟，web性能测试-测试用例总结 (全覆盖)，卷起来...

13年测试老鸟，web性能测试-测试用例总结 (全覆盖)，卷起来...

目录&#xff1a;导读 前言一、Python编程入门到精通二、接口自动化项目实战三、Web自动化项目实战四、App自动化项目实战五、一线大厂简历六、测试开发DevOps体系七、常用自动化测试工具八、JMeter性能测试九、总结&#xff08;尾部小惊喜&#xff09; 前言 性能测试用例主要…
阅读更多...
弟子规 古译今 详解

弟子规 古译今 详解

博主 默语带您 Go to New World. ✍ 个人主页—— 默语 的博客&#x1f466;&#x1f3fb; 《java 面试题大全》 &#x1f369;惟余辈才疏学浅&#xff0c;临摹之作或有不妥之处&#xff0c;还请读者海涵指正。☕&#x1f36d; 《MYSQL从入门到精通》数据库是开发者必会基础之…
阅读更多...
docker专题系列之十五：卸载docker

docker专题系列之十五：卸载docker

一、准备工作 1.杀死docker有关的容器&#xff1a; docker kill $(docker ps -a -q)2.删除所有docker容器&#xff1a; docker rm $(docker ps -a -q)3.删除所有docker镜像&#xff1a; docker rmi $(docker images -q)4.停止 docker 服务&#xff1a; systemctl stop dock…
阅读更多...
销售管理和客户管理为什么要用CRM管理软件？

销售管理和客户管理为什么要用CRM管理软件？

对于公司营销管理、企业管理和客户关系管理而言&#xff0c;CRM管理软件的使用是必不可少的。 通过对企业营销过程中所遭遇的客户资源管理问题进行分析&#xff0c;我们可以发现其困扰的核心问题都围绕着信息的获取和管理。 一些常见的问题包括信息缺失、信息混乱、信息管理困…
阅读更多...
Win10临时文件夹在哪个位置打开？Win10临时文件夹打开教程

Win10临时文件夹在哪个位置打开？Win10临时文件夹打开教程

Win10系统中的临时文件夹是一个用于存储临时数据的目录&#xff0c;现在用户需要访问或清理这个临时文件夹&#xff0c;以便管理计算机&#xff0c;但是对于一些用户来说&#xff0c;找到并打开Win10临时文件夹的实际位置可能会有些困惑。本教程中小编将向用户们介绍Win10临时文…
阅读更多...
网络数据包封装/解封装

网络数据包封装/解封装

数据的发送和接收其实就是封装及解封装的过程。 协议栈逐层向下传递数据&#xff0c;并添加报文头部和报文尾部的过程称为封装。反过来就是解封装。 封装过程 封装过程&#xff1a;上层原始数据→数据段(segment)→数据包(packet)→数据帧(frame)→比特流(bit) 应用层&#…
阅读更多...
可以捡漏的211，复试线即国家线，最低269分上岸！

可以捡漏的211，复试线即国家线，最低269分上岸！

一、学校及专业介绍 新疆大学&#xff08;Xinjiang University&#xff09;&#xff0c;简称“新大”&#xff0c;位于新疆维吾尔自治区首府乌鲁木齐市&#xff0c;国家“双一流”建设高校、国家“211工程”建设高校、国家西部大开发重点建设高校。 1.1 招生情况 一共2个专业进…
阅读更多...
Ubuntu18.04，opencv-4.3.0和opencv_contrib-4.3.0安装（填坑）

Ubuntu18.04，opencv-4.3.0和opencv_contrib-4.3.0安装（填坑）

如果觉得本篇文章对您的学习起到帮助作用&#xff0c;请 点赞 关注 评论 &#xff0c;留下您的足迹&#x1f4aa;&#x1f4aa;&#x1f4aa; 本文主要Ubuntu18.04安装opencv-4.3.0和opencv_contrib-4.3.0&#xff0c;坑巨多&#xff0c;因此记录以备日后查看&#xff0c;同时…
阅读更多...
LLM-Client一个轻量级的LLM集成工具

LLM-Client一个轻量级的LLM集成工具

大型语言模型(llm)已经彻底改变了我们与文本交互的方式&#xff0c;OpenAI、Google、AI21、HuggingfaceHub、Anthropic和众多开源模型提供了不同的功能和优势。但是每个模型都有其独特的体系结构、api和兼容性需求&#xff0c;集成这些模型是一项耗时且具有挑战性的任务。 所以…
阅读更多...
关于Long的前后端传参，精度丢失问题

关于Long的前后端传参，精度丢失问题

若后端接受格式为 List<Long> 则前端传参时不能为 int型 需要改为 &#xff0c;加上双引号
阅读更多...
接口签名验证

接口签名验证

1.appId和secretKey定义 appIdAPPID secretKey cfq4189auoo13y17ur9n2rl7v2tkz3iq 2.sign获取算法 md5&#xff08;secretKey参数json字符串timestampsecretKey&#xff09;后的大写字母。 /*** 参数按key进行排序* param obj* return*/public static String getParamStr(Objec…
阅读更多...
数据库高级- 数据库索引概述及设计原则

数据库高级- 数据库索引概述及设计原则

数据库索引 MySQL官方对索引的定义为&#xff1a;索引&#xff08;index&#xff09;是帮助MySQL高效获取数据的数据结构&#xff08;有序&#xff09;。在数据之外&#xff0c;数据库系统还维护者满足特定查找算法的数据结构&#xff0c;这些数据结构以某种方式引用&#xff0…
阅读更多...
信号链噪声分析10

信号链噪声分析10

目录 概要 整体架构流程 技术名词解释 技术细节 1.移除缓冲级对信号链的影响 2.SNR 与增益的关系 小结 概要 提示&#xff1a;这里可以添加技术概要 信号链构成单元参数示例&#xff0c;以 LTC2387-18&#xff08;2 级 AFE&#xff09;为例&#xff0c;这是一款 15MSPS 精密 S…
阅读更多...
C/C++ 线程池工作原理 代码实现

C/C++ 线程池工作原理 代码实现

1. 线程池作用 如果多次使用线程&#xff0c;那么就需要多次的创建并撤销线程。但是创建/撤销的过程会消耗资源。线程池是一种数据结构&#xff0c;其中维护着多个线程&#xff0c;这避免了在处理短时间任务时&#xff0c;创建与销毁线程的代价。即在程序开始运行前预先创建一…
阅读更多...
设计模式GOF

设计模式GOF

设计模式&#xff08;Design pattern&#xff09;代表了最佳的实践&#xff0c;通常被有经验的面向对象的软件开发人员所采用。设计模式是软件开发人员在软件开发过程中面临的一般问题的解决方案。这些解决方案是众多软件开发人员经过相当长的一段时间的试验和错误总结出来的。…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023