DataX入门教程2
接DataX入门教程1
MongoDB
什么是MongoDB
MongoDB 是由C++语言编写的，是一个基于分布式文件存储的开源数据库系统。MongoDB 旨在为WEB应用提供可扩展的高性能数据存储解决方案。MongoDB 将数据存储为一个文档，数据结构由键值(key=>value)对组成。MongoDB 文档类似于 JSON 对象。字段值可以包含其他文档，数组及文档数组。
MongoDB优缺点
基础概念解析
|
SQL术语/概念 |
MongoDB术语/概念 |
解释/说明 |
|
database |
database |
数据库 |
|
table |
collection |
数据库表/集合 |
|
row |
document |
数据记录行/文档 |
|
column |
field |
数据字段/域 |
|
index |
index |
索引 |
|
table joins |
不支持 |
表连接,MongoDB不支持 |
|
primary key |
primary key |
主键,MongoDB自动将_id字段设置为主键 |
通过下图实例，我们也可以更直观的了解Mongo中的一些概念：
安装
下载地址
https://www.mongodb.com/download-center#community
安装
1）上传压缩包到虚拟机中,解压
[daydayup@hadoop102 software]$ tar -zxvf mongodb-linux-x86_64-rhel70-5.0.2.tgz -C /opt/module/
2）重命名
[daydayup@hadoop102 module]$ mv mongodb-linux-x86_64- rhel70-5.0.2/ mongodb
3）创建数据库目录
MongoDB的数据存储在data目录的db目录下，但是这个目录在安装过程不会自动创建，所以需要手动创建data目录，并在data目录中创建db目录。
[daydayup@hadoop102 module]$ sudo mkdir -p /data/db
[daydayup@hadoop102 mongodb]$ sudo chmod 777 -R /data/db/
5）启动MongoDB服务
[daydayup@hadoop102 mongodb]$ bin/mongod
6）进入shell页面
[daydayup@hadoop102 mongodb]$ bin/mongo
基础概念详解
数据库
一个mongodb中可以建立多个数据库。MongoDB的默认数据库为"db"，该数据库存储在data目录中。MongoDB的单个实例可以容纳多个独立的数据库，每一个都有自己的集合和权限，不同的数据库也放置在不同的文件中。
1）显示所有数据库
> show dbs
admin 0.000GB
config 0.000GB
local 0.000GB
- admin：从权限的角度来看，这是"root"数据库。要是将一个用户添加到这个数据库，这个用户自动继承所有数据库的权限。一些特定的服务器端命令也只能从这个数据库运行，比如列出所有的数据库或者关闭服务器。
- local：这个数据永远不会被复制，可以用来存储限于本地单台服务器的任意集合
- config：当Mongo用于分片设置时，config数据库在内部使用，用于保存分片的相关信息。
2）显示当前使用的数据库
> db
test
3）切换数据库
> use local
switched to db local
> db
local
集合
集合就是 MongoDB 文档组，类似于MySQL中的table。
集合存在于数据库中，集合没有固定的结构，这意味着你在对集合可以插入不同格式和类型的数据，但通常情况下我们插入集合的数据都会有一定的关联性。
MongoDB 中使用 createCollection() 方法来创建集合。下面我们来看看如何创建集合：
语法格式：
db.createCollection(name, options)
参数说明：
name: 要创建的集合名称
options: 可选参数, 指定有关内存大小及索引的选项，有以下参数：
|
字段 |
类型 |
描述 |
|---|---|---|
|
capped |
布尔 |
（可选）如果为 true，则创建固定集合。固定集合是指有着固定大小的集合，当达到最大值时，它会自动覆盖最早的文档。 |
|
autoIndexId |
布尔 |
（可选）如为 true，自动在 _id 字段创建索引。默认为 false。 |
|
size |
数值 |
（可选）为固定集合指定一个最大值（以字节计）。 |
|
max |
数值 |
（可选）指定固定集合中包含文档的最大数量。 |
案例1：在test库中创建一个daydayup的集合
> use test
switched to db test
> db.createCollection("daydayup")
{ "ok" : 1 }
> show collections
Daydayup
//插入数据
> db.daydayup.insert({"name":"daydayup","url":"www.daydayup.com"})
WriteResult({ "nInserted" : 1 })
//查看数据
> db.daydayup.find()
{ "_id" : ObjectId("5d0314ceecb77ee2fb2d7566"), "name" : "daydayup", "url" : "www.daydayup.com" }
说明：
ObjectId 类似唯一主键，可以很快的去生成和排序，包含 12 bytes，由24个16进制数字组成的字符串（每个字节可以存储两个16进制数字）,含义是：
-
- 前 4 个字节表示创建 unix 时间戳
- 接下来的 3 个字节是机器标识码
- 紧接的两个字节由进程 id 组成 PID
- 最后三个字节是随机数
案例2：创建一个固定集合mycol
> db.createCollection("mycol",{ capped : true,autoIndexId : true,size : 6142800, max : 1000})
> show tables;
daydayup
mycol
案例3：自动创建集合
在 MongoDB 中，你不需要创建集合。当你插入一些文档时，MongoDB 会自动创建集合。
> db.mycol2.insert({"name":"daydayup"})
WriteResult({ "nInserted" : 1 })
> show collections
daydayup
mycol
mycol2
案例4：删除集合
> db.mycol2.drop()
True
> show tables;
daydayup
mycol
文档(Document)
文档是一组键值(key-value)对组成。MongoDB 的文档不需要设置相同的字段，并且相同的字段不需要相同的数据类型，这与关系型数据库有很大的区别，也是 MongoDB 非常突出的特点。
一个简单的例子：
{"name":"daydayup"}
注意：
1、文档中的键/值对是有序的。
2、MongoDB区分类型和大小写。
3、MongoDB的文档不能有重复的键。
4、文档的键是字符串。除了少数例外情况，键可以使用任意UTF-8字符。
DataX导入导出案例
读取MongoDB的数据导入到HDFS
1）编写配置文件
[daydayup@hadoop102 datax]$ vim job/mongdb2hdfs.json
{
"job": {
"content": [
{
"reader": {
"name": "mongodbreader",
"parameter": {
"address": ["127.0.0.1:27017"],
"collectionName": "daydayup",
"column": [
{
"name":"name",
"type":"string"
},
{
"name":"url",
"type":"string"
}
],
"dbName": "test",
}
},
"writer": {
"name": "hdfswriter",
"parameter": {
"column": [
{
"name":"name",
"type":"string"
},
{
"name":"url",
"type":"string"
}
],
"defaultFS": "hdfs://hadoop102:9000",
"fieldDelimiter": "\t",
"fileName": "mongo.txt",
"fileType": "text",
"path": "/",
"writeMode": "append"
}
}
}
],
"setting": {
"speed": {
"channel": "1"
}
}
}
}
2）mongodbreader参数解析
- address： MongoDB的数据地址信息，因为MonogDB可能是个集群，则ip端口信息需要以Json数组的形式给出。【必填】
- userName：MongoDB的用户名。【选填】
- userPassword： MongoDB的密码。【选填】
- collectionName： MonogoDB的集合名。【必填】
- column：MongoDB的文档列名。【必填】
- name：Column的名字。【必填】
- type：Column的类型。【选填】
- splitter：因为MongoDB支持数组类型，但是Datax框架本身不支持数组类型，所以mongoDB读出来的数组类型要通过这个分隔符合并成字符串。【选填】
3）执行
[daydayup@hadoop102 datax]$ bin/datax.py job/mongdb2hdfs.json
4）查看结果
读取MongoDB的数据导入MySQL
1）在MySQL中创建表
mysql> create table daydayup(name varchar(20),url varchar(20));
2）编写DataX配置文件
[daydayup@hadoop102 datax]$ vim job/mongodb2mysql.json
{
"job": {
"content": [
{
"reader": {
"name": "mongodbreader",
"parameter": {
"address": ["127.0.0.1:27017"],
"collectionName": "daydayup",
"column": [
{
"name":"name",
"type":"string"
},
{
"name":"url",
"type":"string"
}
],
"dbName": "test",
}
},
"writer": {
"name": "mysqlwriter",
"parameter": {
"column": ["*"],
"connection": [
{
"jdbcUrl": "jdbc:mysql://hadoop102:3306/test",
"table": ["daydayup"]
}
],
"password": "000000",
"username": "root",
"writeMode": "insert"
}
}
}
],
"setting": {
"speed": {
"channel": "1"
}
}
}
}
3）执行
[daydayup@hadoop102 datax]$ bin/datax.py job/mongodb2mysql.json
4）查看结果
mysql> select * from daydayup;
+---------+-----------------+
| name | url |
+---------+-----------------+
| daydayup | www.daydayup.com |
+---------+-----------------+
SQLServer
什么是SQLServer
美国Microsoft公司推出的一种关系型数据库系统。SQL Server是一个可扩展的、高性能的、为分布式客户机/服务器计算所设计的数据库管理系统，实现了与WindowsNT的有机结合，提供了基于事务的企业级信息管理系统方案。SQL Server的基本语法和MySQL基本相同。
（1）高性能设计，可充分利用WindowsNT的优势。
（2）系统管理先进，支持Windows图形化管理工具，支持本地和远程的系统管理和配置。
（3）强壮的事务处理功能，采用各种方法保证数据的完整性。
（4）支持对称多处理器结构、存储过程、ODBC，并具有自主的SQL语言。 SQLServer以其内置的数据复制功能、强大的管理工具、与Internet的紧密集成和开放的系统结构为广大的用户、开发人员和系统集成商提供了一个出众的数据库平台。
安装
安装要求
系统要求：
1、centos或redhat7.0以上系统
2、内存2G以上
说明：
linux下安装sqlserver数据库有2种办法：
- 使用rpm安装包安装
rpm安装包地址：https://packages.microsoft.com/rhel/7/mssql-server-2017/
安装时缺少什么依赖，就使用yum进行安装补齐
- 使用yum镜像安装
安装步骤
1）下载 Microsoft SQL Server 2017 Red Hat 存储库配置文件
sudo curl -o /etc/yum.repos.d/mssql-server.repo https://packages.microsoft.com/config/rhel/7/mssql-server-2017.repo
2）执行安装
yum install -y mssql-server
3）完毕之后运行做相关配置
sudo /opt/mssql/bin/mssql-conf setup
安装配置
1）执行配置命令
sudo /opt/mssql/bin/mssql-conf setup
2）选择安装的版本
3）接受许可条款
3）选择语言
4）配置系统管理员密码
5）完成
安装命令行工具
1）下载存储库配置文件
sudo curl -o /etc/yum.repos.d/msprod.repo https://packages.microsoft.com/config/rhel/7/prod.repo
2）执行安装
sudo yum remove mssql-tools unixODBC-utf16-devel
sudo yum install mssql-tools unixODBC-devel
3）配置环境变量
sudo vim /etc/profile.d/my_env.sh
#添加环境变量
export PATH="$PATH:/opt/mssql-tools/bin
source /etc/profile.d/my_env.sh
4）进入命令行
sqlcmd -S localhost -U SA -P 密码 # 用命令行连接
简单使用
启停命令
#启动
systemctl start mssql-server
#重启
systemctl restart mssql-server
#停止
systemctl stop mssql-server
#查看状态
systemctl status mssql-server
#具体配置路径
/opt/mssql/bin/mssql-conf
创建数据库
1) 建库
> create database datax
> go
(2) 看当前数据库列表
> select * from SysDatabases
> go
(3) 看当前数据表
> use 库名
> select * from sysobjects where xtype='u'
> go
(4) 看表的内容
> select * from 表名;
> go
DataX导入导出案例
创建表并插入数据
create table student(id int,name varchar(25))
go
insert into student values(1,'zhangsan')
go
读取SQLServer的数据导入到HDFS
1）编写配置文件
[daydayup@hadoop102 datax]$ vim job/sqlserver2hdfs.json
{
"job": {
"content": [
{
"reader": {
"name": "sqlserverreader",
"parameter": {
"column": [
"id",
"name"
],
"connection": [
{
"jdbcUrl": [
"jdbc:sqlserver://hadoop2:1433;DatabaseName=datax"
],
"table": [
"student"
]
}
],
"username": "root",
"password": "000000"
}
},
"writer": {
"name": "hdfswriter",
"parameter": {
"column": [
{
"name": "id",
"type": "int"
},
{
"name": "name",
"type": "string"
}
],
"defaultFS": "hdfs://hadoop102:9000",
"fieldDelimiter": "\t",
"fileName": "sqlserver.txt",
"fileType": "text",
"path": "/",
"writeMode": "append"
}
}
}
],
"setting": {
"speed": {
"channel": "1"
}
}
}
}
读取SQLServer的数据导入MySQL
[daydayup@hadoop102 datax]$ vim job/sqlserver2mysql.json
{
"job": {
"content": [
{
"reader": {
"name": "sqlserverreader",
"parameter": {
"column": [
"id",
"name"
],
"connection": [
{
"jdbcUrl": [
"jdbc:sqlserver://hadoop2:1433;DatabaseName=datax"
],
"table": [
"student"
]
}
],
"username": "root",
"password": "000000"
}
},
"writer": {
"name": "mysqlwriter",
"parameter": {
"column": ["*"],
"connection": [
{
"jdbcUrl": "jdbc:mysql://hadoop102:3306/datax",
"table": ["student"]
}
],
"password": "000000",
"username": "root",
"writeMode": "insert"
}
}
}
],
"setting": {
"speed": {
"channel": "1"
}
}
}
}
DB2
什么是db2
DB2是IBM公司于1983年研制的一种关系型数据库系统(Relational Database Management System)，主要应用于大型应用系统，具有较好的可伸缩性 。DB2是IBM推出的第二个关系型数据库，所以称为db2。DB2 提供了高层次的数据利用性、完整性、安全性 、并行性、可恢复性，以及小规模到大规模应用程序的执行能力，具有与平台无关的基本功能和SQL命令运行环境。可以同时在不同操作系统使用，包括Linux、UNIX 和 Windows。
db2数据库对象关系
1、instance, 同一台机器上可以安装多个DB2 instance。
2、database, 同一个instance下面可以创建有多个database。
3、schema, 同一个database下面可以配置多个schema。
4、table，同一个schema下可以创建多个table。
安装前的准备
安装依赖
yum install -y bc binutils compat-libcap1 compat-libstdc++33 elfutils-libelf elfutils-libelf-devel fontconfig-devel glibc glibc-devel ksh libaio libaio-devel libX11 libXau libXi libXtst libXrender libXrender-devel libgcc libstdc++ libstdc++-devel libxcb make smartmontools sysstat kmod* gcc-c++ compat-libstdc++-33 libstdc++.so.6 kernel-devel pam-devel.i686 pam.i686 pam32*
修改配置文件sysctl.conf
[root@hadoop102 module]# vim /etc/sysctl.conf
删除里面的内容，添加如下内容：
net.ipv4.ip_local_port_range = 9000 65500
fs.file-max = 6815744
kernel.shmall = 10523004
kernel.shmmax = 6465333657
kernel.shmmni = 4096
kernel.sem = 250 32000 100 128
net.core.rmem_default=262144
net.core.wmem_default=262144
net.core.rmem_max=4194304
net.core.wmem_max=1048576
fs.aio-max-nr = 1048576
修改配置文件limits.conf
[root@hadoop102 module]# vim /etc/security/limits.conf
在文件末尾添加：
* soft nproc 65536
* hard nproc 65536
* soft nofile 65536
* hard nofile 65536
重启机器生效。
上传安装包并解压
[root@hadoop102 software]# tar -zxvf v11.5.4_linuxx64_server_dec.tar.gz -C /opt/module/
[root@hadoop102 module]# chmod 777 server_dec
安装
在root用户下操作
执行预检查命令
./db2prereqcheck -l -s //检查环境
执行安装
./db2_install
1）接受许可条款
可能会出现两次询问是否接受条款，都选“是”即可。
2）确认安装路径，默认
3）选择安装SERVER
4）不安装pureScale
等待安装完成即可
5）查看许可
/opt/ibm/db2/V11.5/adm/db2licm -l
添加组和用户
groupadd -g 2000 db2iadm1
groupadd -g 2001 db2fadm1
useradd -m -g db2iadm1 -d /home/db2inst1 db2inst1
useradd -m -g db2fadm1 -d /home/db2fenc1 db2fenc1
passwd db2inst1
passwd db2fenc1
-
- db2inst1: 实例所有者
- db2fenc1: 受防护用户
创建实例
cd /opt/ibm/db2/V11.5/instance
./db2icrt -p 50000 -u db2fenc1 db2inst1
创建样本数据库、开启服务
su - db2inst1
db2sampl
db2start
连接
db2
conncet to sample #连接到某个数据库
select * from staff
创建表、插入数据
CREATE TABLE STUDENT(ID int ,NAME varchar(20));
INSERT INTO STUDENT VALUES(11, 'lisi');
commit;
DataX导入导出案例
注册db2驱动
datax暂时没有独立插件支持db2，需要使用通用的使用rdbmsreader或rdbmswriter。
1）注册reader的db2驱动
[daydayup@hadoop102 datax]$ vim /opt/module/datax/plugin/reader/rdbmsreader/plugin.json
#在drivers里添加db2的驱动类
"drivers":["dm.jdbc.driver.DmDriver", "com.sybase.jdbc3.jdbc.SybDriver", "com.edb.Driver","com.ibm.db2.jcc.DB2Driver"]
2）注册writer的db2驱动
[daydayup@hadoop102 datax]$ vim /opt/module/datax/plugin/writer/rdbmswriter/plugin.json
#在drivers里添加db2的驱动类
"drivers":["dm.jdbc.driver.DmDriver", "com.sybase.jdbc3.jdbc.SybDriver", "com.edb.Driver","com.ibm.db2.jcc.DB2Driver"]
读取DB2的数据导入到HDFS
1）编写配置文件
[daydayup@hadoop102 datax]$ vim job/db2-2-hdfs.json
{
"job": {
"content": [
{
"reader": {
"name": "rdbmsreader",
"parameter": {
"column": [
"ID",
"NAME"
],
"connection": [
{
"jdbcUrl": [
"jdbc:db2://hadoop2:50000/sample"
],
"table": [
"STUDENT"
]
}
],
"username": "db2inst1",
"password": "daydayup"
}
},
"writer": {
"name": "hdfswriter",
"parameter": {
"column": [
{
"name": "id",
"type": "int"
},
{
"name": "name",
"type": "string"
}
],
"defaultFS": "hdfs://hadoop102:9000",
"fieldDelimiter": "\t",
"fileName": "db2.txt",
"fileType": "text",
"path": "/",
"writeMode": "append"
}
}
}
],
"setting": {
"speed": {
"channel": "1"
}
}
}
}
读取DB2的数据导入MySQL
[daydayup@hadoop102 datax]$ vim job/db2-2-mysql.json
{
"job": {
"content": [
{
"reader": {
"name": "rdbmsreader",
"parameter": {
"column": [
"ID",
"NAME"
],
"connection": [
{
"jdbcUrl": [
"jdbc:db2://hadoop2:50000/sample"
],
"table": [
"STUDENT"
]
}
],
"username": "db2inst1",
"password": "daydayup"
}
},
"writer": {
"name": "mysqlwriter",
"parameter": {
"column": ["*"],
"connection": [
{
"jdbcUrl": "jdbc:mysql://hadoop102:3306/datax",
"table": ["student"]
}
],
"password": "000000",
"username": "root",
"writeMode": "insert"
}
}
}
],
"setting": {
"speed": {
"channel": "1"
}
}
}
}
执行流程源码分析
总体流程
- 黄色： Job 部分的执行阶段，
- 蓝色： Task 部分的执行阶段，
- 绿色：框架执行阶段。
程序入口
datax.py
……
ENGINE_COMMAND = "java -server ${jvm} %s -classpath %s ${params} com.alibaba.datax.core.Engine -mode ${mode} -jobid ${jobid} -job ${job}" % (
DEFAULT_PROPERTY_CONF, CLASS_PATH)
……
Engine.java
public void start(Configuration allConf) {
……
//JobContainer会在schedule后再行进行设置和调整值
int channelNumber =0;
AbstractContainer container;
long instanceId;
int taskGroupId = -1;
if (isJob) {
allConf.set(CoreConstant.DATAX_CORE_CONTAINER_JOB_MODE, RUNTIME_MODE);
container = new JobContainer(allConf);
instanceId = allConf.getLong(
CoreConstant.DATAX_CORE_CONTAINER_JOB_ID, 0);
} else {
container = new TaskGroupContainer(allConf);
instanceId = allConf.getLong(
CoreConstant.DATAX_CORE_CONTAINER_JOB_ID);
taskGroupId = allConf.getInt(
CoreConstant.DATAX_CORE_CONTAINER_TASKGROUP_ID);
channelNumber = allConf.getInt(
CoreConstant.DATAX_CORE_CONTAINER_TASKGROUP_CHANNEL);
}
……
container.start();
}
JobContainer.java
/**
* jobContainer主要负责的工作全部在start()里面，包括init、prepare、split、scheduler、
* post以及destroy和statistics
*/
@Override
public void start() {
LOG.info("DataX jobContainer starts job.");
boolean hasException = false;
boolean isDryRun = false;
try {
this.startTimeStamp = System.currentTimeMillis();
isDryRun = configuration.getBool(CoreConstant.DATAX_JOB_SETTING_DRYRUN, false);
if(isDryRun) {
LOG.info("jobContainer starts to do preCheck ...");
this.preCheck();
} else {
userConf = configuration.clone();
LOG.debug("jobContainer starts to do preHandle ...");
//Job前置操作
this.preHandle();
LOG.debug("jobContainer starts to do init ...");
//初始化reader和writer
this.init();
LOG.info("jobContainer starts to do prepare ...");
//全局准备工作，比如odpswriter清空目标表
this.prepare();
LOG.info("jobContainer starts to do split ...");
//拆分Task
this.totalStage = this.split();
LOG.info("jobContainer starts to do schedule ...");
this.schedule();
LOG.debug("jobContainer starts to do post ...");
this.post();
LOG.debug("jobContainer starts to do postHandle ...");
this.postHandle();
LOG.info("DataX jobId [{}] completed successfully.", this.jobId);
this.invokeHooks();
}
} ……
}
Task切分逻辑
JobContainer.java
private int split() {
this.adjustChannelNumber();
if (this.needChannelNumber <= 0) {
this.needChannelNumber = 1;
}
List<Configuration> readerTaskConfigs = this
.doReaderSplit(this.needChannelNumber);
int taskNumber = readerTaskConfigs.size();
List<Configuration> writerTaskConfigs = this
.doWriterSplit(taskNumber);
List<Configuration> transformerList = this.configuration.getListConfiguration(CoreConstant.DATAX_JOB_CONTENT_TRANSFORMER);
LOG.debug("transformer configuration: "+ JSON.toJSONString(transformerList));
/**
* 输入是reader和writer的parameter list，输出是content下面元素的list
*/
List<Configuration> contentConfig = mergeReaderAndWriterTaskConfigs(
readerTaskConfigs, writerTaskConfigs, transformerList);
LOG.debug("contentConfig configuration: "+ JSON.toJSONString(contentConfig));
this.configuration.set(CoreConstant.DATAX_JOB_CONTENT, contentConfig);
return contentConfig.size();
}
并发数的确定
private void adjustChannelNumber() {
int needChannelNumberByByte = Integer.MAX_VALUE;
int needChannelNumberByRecord = Integer.MAX_VALUE;
boolean isByteLimit = (this.configuration.getInt(
CoreConstant.DATAX_JOB_SETTING_SPEED_BYTE, 0) > 0);
if (isByteLimit) {
long globalLimitedByteSpeed = this.configuration.getInt(
CoreConstant.DATAX_JOB_SETTING_SPEED_BYTE, 10 * 1024 * 1024);
// 在byte流控情况下，单个Channel流量最大值必须设置，否则报错！
Long channelLimitedByteSpeed = this.configuration
.getLong(CoreConstant.DATAX_CORE_TRANSPORT_CHANNEL_SPEED_BYTE);
if (channelLimitedByteSpeed == null || channelLimitedByteSpeed <= 0) {
throw DataXException.asDataXException(
FrameworkErrorCode.CONFIG_ERROR,
"在有总bps限速条件下，单个channel的bps值不能为空，也不能为非正数");
}
needChannelNumberByByte =
(int) (globalLimitedByteSpeed / channelLimitedByteSpeed);
needChannelNumberByByte =
needChannelNumberByByte > 0 ? needChannelNumberByByte : 1;
LOG.info("Job set Max-Byte-Speed to " + globalLimitedByteSpeed + " bytes.");
}
boolean isRecordLimit = (this.configuration.getInt(
CoreConstant.DATAX_JOB_SETTING_SPEED_RECORD, 0)) > 0;
if (isRecordLimit) {
long globalLimitedRecordSpeed = this.configuration.getInt(
CoreConstant.DATAX_JOB_SETTING_SPEED_RECORD, 100000);
Long channelLimitedRecordSpeed = this.configuration.getLong(
CoreConstant.DATAX_CORE_TRANSPORT_CHANNEL_SPEED_RECORD);
if (channelLimitedRecordSpeed == null || channelLimitedRecordSpeed <= 0) {
throw DataXException.asDataXException(FrameworkErrorCode.CONFIG_ERROR,
"在有总tps限速条件下，单个channel的tps值不能为空，也不能为非正数");
}
needChannelNumberByRecord =
(int) (globalLimitedRecordSpeed / channelLimitedRecordSpeed);
needChannelNumberByRecord =
needChannelNumberByRecord > 0 ? needChannelNumberByRecord : 1;
LOG.info("Job set Max-Record-Speed to " + globalLimitedRecordSpeed + " records.");
}
// 取较小值
this.needChannelNumber = needChannelNumberByByte < needChannelNumberByRecord ?
needChannelNumberByByte : needChannelNumberByRecord;
// 如果从byte或record上设置了needChannelNumber则退出
if (this.needChannelNumber < Integer.MAX_VALUE) {
return;
}
boolean isChannelLimit = (this.configuration.getInt(
CoreConstant.DATAX_JOB_SETTING_SPEED_CHANNEL, 0) > 0);
if (isChannelLimit) {
this.needChannelNumber = this.configuration.getInt(
CoreConstant.DATAX_JOB_SETTING_SPEED_CHANNEL);
LOG.info("Job set Channel-Number to " + this.needChannelNumber
+ " channels.");
return;
}
throw DataXException.asDataXException(
FrameworkErrorCode.CONFIG_ERROR,
"Job运行速度必须设置");
}
调度
JobContainer.java
private void schedule() {
/**
* 这里的全局speed和每个channel的速度设置为B/s
*/
int channelsPerTaskGroup = this.configuration.getInt(
CoreConstant.DATAX_CORE_CONTAINER_TASKGROUP_CHANNEL, 5);
int taskNumber = this.configuration.getList(
CoreConstant.DATAX_JOB_CONTENT).size();
//确定的channel数和切分的task数取最小值，避免浪费
this.needChannelNumber = Math.min(this.needChannelNumber, taskNumber);
PerfTrace.getInstance().setChannelNumber(needChannelNumber);
/**
* 通过获取配置信息得到每个taskGroup需要运行哪些tasks任务
*/
List<Configuration> taskGroupConfigs = JobAssignUtil.assignFairly(this.configuration,
this.needChannelNumber, channelsPerTaskGroup);
LOG.info("Scheduler starts [{}] taskGroups.", taskGroupConfigs.size());
ExecuteMode executeMode = null;
AbstractScheduler scheduler;
try {
//可以看到3.0进行了阉割，只有STANDALONE模式
executeMode = ExecuteMode.STANDALONE;
scheduler = initStandaloneScheduler(this.configuration);
//设置 executeMode
for (Configuration taskGroupConfig : taskGroupConfigs) {
taskGroupConfig.set(CoreConstant.DATAX_CORE_CONTAINER_JOB_MODE, executeMode.getValue());
}
if (executeMode == ExecuteMode.LOCAL || executeMode == ExecuteMode.DISTRIBUTE) {
if (this.jobId <= 0) {
throw DataXException.asDataXException(FrameworkErrorCode.RUNTIME_ERROR,
"在[ local | distribute ]模式下必须设置jobId，并且其值 > 0 .");
}
}
LOG.info("Running by {} Mode.", executeMode);
this.startTransferTimeStamp = System.currentTimeMillis();
scheduler.schedule(taskGroupConfigs);
this.endTransferTimeStamp = System.currentTimeMillis();
} catch (Exception e) {
LOG.error("运行scheduler 模式[{}]出错.", executeMode);
this.endTransferTimeStamp = System.currentTimeMillis();
throw DataXException.asDataXException(
FrameworkErrorCode.RUNTIME_ERROR, e);
}
/**
* 检查任务执行情况
*/
this.checkLimit();
}
确定组数和分组
assignFairly方法：
1）确定taskGroupNumber，
2）做分组分配，
3）做分组优化
public static List<Configuration> assignFairly(Configuration configuration, int channelNumber, int channelsPerTaskGroup) {
Validate.isTrue(configuration != null, "框架获得的 Job 不能为 null.");
List<Configuration> contentConfig = configuration.getListConfiguration(CoreConstant.DATAX_JOB_CONTENT);
Validate.isTrue(contentConfig.size() > 0, "框架获得的切分后的 Job 无内容.");
Validate.isTrue(channelNumber > 0 && channelsPerTaskGroup > 0,
"每个channel的平均task数[averTaskPerChannel]，channel数目[channelNumber]，每个taskGroup的平均channel数[channelsPerTaskGroup]都应该为正数");
//TODO 确定taskgroup的数量
int taskGroupNumber = (int) Math.ceil(1.0 * channelNumber / channelsPerTaskGroup);
Configuration aTaskConfig = contentConfig.get(0);
String readerResourceMark = aTaskConfig.getString(CoreConstant.JOB_READER_PARAMETER + "." +
CommonConstant.LOAD_BALANCE_RESOURCE_MARK);
String writerResourceMark = aTaskConfig.getString(CoreConstant.JOB_WRITER_PARAMETER + "." +
CommonConstant.LOAD_BALANCE_RESOURCE_MARK);
boolean hasLoadBalanceResourceMark = StringUtils.isNotBlank(readerResourceMark) ||
StringUtils.isNotBlank(writerResourceMark);
if (!hasLoadBalanceResourceMark) {
// fake 一个固定的 key 作为资源标识（在 reader 或者 writer 上均可，此处选择在 reader 上进行 fake）
for (Configuration conf : contentConfig) {
conf.set(CoreConstant.JOB_READER_PARAMETER + "." +
CommonConstant.LOAD_BALANCE_RESOURCE_MARK, "aFakeResourceMarkForLoadBalance");
}
// 是为了避免某些插件没有设置 资源标识 而进行了一次随机打乱操作
Collections.shuffle(contentConfig, new Random(System.currentTimeMillis()));
}
LinkedHashMap<String, List<Integer>> resourceMarkAndTaskIdMap = parseAndGetResourceMarkAndTaskIdMap(contentConfig);
List<Configuration> taskGroupConfig = doAssign(resourceMarkAndTaskIdMap, configuration, taskGroupNumber);
// 调整 每个 taskGroup 对应的 Channel 个数（属于优化范畴）
adjustChannelNumPerTaskGroup(taskGroupConfig, channelNumber);
return taskGroupConfig;
}
调度实现
AbstractScheduler.java
public void schedule(List<Configuration> configurations) {
Validate.notNull(configurations,
"scheduler配置不能为空");
int jobReportIntervalInMillSec = configurations.get(0).getInt(
CoreConstant.DATAX_CORE_CONTAINER_JOB_REPORTINTERVAL, 30000);
int jobSleepIntervalInMillSec = configurations.get(0).getInt(
CoreConstant.DATAX_CORE_CONTAINER_JOB_SLEEPINTERVAL, 10000);
this.jobId = configurations.get(0).getLong(
CoreConstant.DATAX_CORE_CONTAINER_JOB_ID);
errorLimit = new ErrorRecordChecker(configurations.get(0));
/**
* 给 taskGroupContainer 的 Communication 注册
*/
this.containerCommunicator.registerCommunication(configurations);
int totalTasks = calculateTaskCount(configurations);
startAllTaskGroup(configurations);
Communication lastJobContainerCommunication = new Communication();
long lastReportTimeStamp = System.currentTimeMillis();
try {
while (true) {
/**
* step 1: collect job stat
* step 2: getReport info, then report it
* step 3: errorLimit do check
* step 4: dealSucceedStat();
* step 5: dealKillingStat();
* step 6: dealFailedStat();
* step 7: refresh last job stat, and then sleep for next while
*
* above steps, some ones should report info to DS
*
*/
……
}
}
……
}
ProcessInnerScheduler.java
public void startAllTaskGroup(List<Configuration> configurations) {
this.taskGroupContainerExecutorService = Executors
.newFixedThreadPool(configurations.size());
for (Configuration taskGroupConfiguration : configurations) {
TaskGroupContainerRunner taskGroupContainerRunner = newTaskGroupContainerRunner(taskGroupConfiguration);
this.taskGroupContainerExecutorService.execute(taskGroupContainerRunner);
}
this.taskGroupContainerExecutorService.shutdown();
}
数据传输
接8.3.2丢到线程池执行
TaskGroupContainer.start()
-> taskExecutor.doStart()
可以看到调用插件的start方法
public void doStart() {
this.writerThread.start();
// reader没有起来，writer不可能结束
if (!this.writerThread.isAlive() || this.taskCommunication.getState() == State.FAILED) {
throw DataXException.asDataXException(
FrameworkErrorCode.RUNTIME_ERROR,
this.taskCommunication.getThrowable());
}
this.readerThread.start();
……
}
可以看看generateRunner()
ReaderRunner.java
public void run() {
……
try {
channelWaitWrite.start();
……
initPerfRecord.start();
taskReader.init();
initPerfRecord.end();
……
preparePerfRecord.start();
taskReader.prepare();
preparePerfRecord.end();
……
dataPerfRecord.start();
taskReader.startRead(recordSender);
recordSender.terminate();
……
postPerfRecord.start();
taskReader.post();
postPerfRecord.end();
// automatic flush
// super.markSuccess(); 这里不能标记为成功，成功的标志由 writerRunner 来标志（否则可能导致 reader 先结束，而 writer 还没有结束的严重 bug）
} catch (Throwable e) {
LOG.error("Reader runner Received Exceptions:", e);
super.markFail(e);
} finally {
LOG.debug("task reader starts to do destroy ...");
PerfRecord desPerfRecord = new PerfRecord(getTaskGroupId(), getTaskId(), PerfRecord.PHASE.READ_TASK_DESTROY);
desPerfRecord.start();
super.destroy();
desPerfRecord.end();
channelWaitWrite.end(super.getRunnerCommunication().getLongCounter(CommunicationTool.WAIT_WRITER_TIME));
long transformerUsedTime = super.getRunnerCommunication().getLongCounter(CommunicationTool.TRANSFORMER_USED_TIME);
if (transformerUsedTime > 0) {
PerfRecord transformerRecord = new PerfRecord(getTaskGroupId(), getTaskId(), PerfRecord.PHASE.TRANSFORMER_TIME);
transformerRecord.start();
transformerRecord.end(transformerUsedTime);
}
}
}
限速的实现
比如看MysqlReader的startReader方法
-》CommonRdbmsReaderTask.startRead()
-》transportOneRecord()
-》sendToWriter()
-》BufferedRecordExchanger. flush()
-》Channel.pushAll()
-》Channel. statPush()
private void statPush(long recordSize, long byteSize) {
currentCommunication.increaseCounter(CommunicationTool.READ_SUCCEED_RECORDS,
recordSize);
currentCommunication.increaseCounter(CommunicationTool.READ_SUCCEED_BYTES,
byteSize);
//在读的时候进行统计waitCounter即可，因为写（pull）的时候可能正在阻塞，但读的时候已经能读到这个阻塞的counter数
currentCommunication.setLongCounter(CommunicationTool.WAIT_READER_TIME, waitReaderTime);
currentCommunication.setLongCounter(CommunicationTool.WAIT_WRITER_TIME, waitWriterTime);
boolean isChannelByteSpeedLimit = (this.byteSpeed > 0);
boolean isChannelRecordSpeedLimit = (this.recordSpeed > 0);
if (!isChannelByteSpeedLimit && !isChannelRecordSpeedLimit) {
return;
}
long lastTimestamp = lastCommunication.getTimestamp();
long nowTimestamp = System.currentTimeMillis();
long interval = nowTimestamp - lastTimestamp;
if (interval - this.flowControlInterval >= 0) {
long byteLimitSleepTime = 0;
long recordLimitSleepTime = 0;
if (isChannelByteSpeedLimit) {
long currentByteSpeed = (CommunicationTool.getTotalReadBytes(currentCommunication) -
CommunicationTool.getTotalReadBytes(lastCommunication)) * 1000 / interval;
if (currentByteSpeed > this.byteSpeed) {
// 计算根据byteLimit得到的休眠时间
byteLimitSleepTime = currentByteSpeed * interval / this.byteSpeed
- interval;
}
}
if (isChannelRecordSpeedLimit) {
long currentRecordSpeed = (CommunicationTool.getTotalReadRecords(currentCommunication) -
CommunicationTool.getTotalReadRecords(lastCommunication)) * 1000 / interval;
if (currentRecordSpeed > this.recordSpeed) {
// 计算根据recordLimit得到的休眠时间
recordLimitSleepTime = currentRecordSpeed * interval / this.recordSpeed
- interval;
}
}
// 休眠时间取较大值
long sleepTime = byteLimitSleepTime < recordLimitSleepTime ?
recordLimitSleepTime : byteLimitSleepTime;
if (sleepTime > 0) {
try {
Thread.sleep(sleepTime);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
……
}
}
DataX使用优化
关键参数
- job.setting.speed.channel : channel并发数
- job.setting.speed.record : 2全局配置channel的record限速
- job.setting.speed.byte：全局配置channel的byte限速
- core.transport.channel.speed.record：单个channel的record限速
- core.transport.channel.speed.byte：单个channel的byte限速
优化1：提升每个channel的速度
在DataX内部对每个Channel会有严格的速度控制，分两种，一种是控制每秒同步的记录数，另外一种是每秒同步的字节数，默认的速度限制是1MB/s，可以根据具体硬件情况设置这个byte速度或者record速度，一般设置byte速度，比如：我们可以把单个Channel的速度上限配置为5MB
优化2：提升DataX Job内Channel并发数
并发数 = taskGroup的数量 * 每个TaskGroup并发执行的Task数 (默认为5)。
提升job内Channel并发有三种配置方式：
配置全局Byte限速以及单Channel Byte限速
Channel个数 = 全局Byte限速 / 单Channel Byte限速
{
"core": {
"transport": {
"channel": {
"speed": {
"byte": 1048576
}
}
}
},
"job": {
"setting": {
"speed": {
"byte" : 5242880
}
},
...
}
}
core.transport.channel.speed.byte=1048576，job.setting.speed.byte=5242880，所以Channel个数 = 全局Byte限速 / 单Channel Byte限速=5242880/1048576=5个
配置全局Record限速以及单Channel Record限速
Channel个数 = 全局Record限速 / 单Channel Record限速
{
"core": {
"transport": {
"channel": {
"speed": {
"record": 100
}
}
}
},
"job": {
"setting": {
"speed": {
"record" : 500
}
},
...
}
}
core.transport.channel.speed.record=100，job.setting.speed.record=500,所以配置全局Record限速以及单Channel Record限速，Channel个数 = 全局Record限速 / 单Channel Record限速=500/100=5
直接配置Channel个数
只有在上面两种未设置才生效，上面两个同时设置是取值小的作为最终的channel数。
{
"job": {
"setting": {
"speed": {
"channel" : 5
}
},
...
}
}
直接配置job.setting.speed.channel=5，所以job内Channel并发=5个
优化3：提高JVM堆内存
当提升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" XXX.json
标签：name,MongoDB,入门教程,job,DataX,hadoop102,channel,daydayup From： https://www.cnblogs.com/kkk247843405/p/16920514.html