数据集成工具——DataXDataXWeb

1、Hive通过外部表与HBase表关联

1&＃xff09;、hive建表语句&＃xff1a;

// 第一个字段通常命名为key
CREATE EXTERNAL TABLE hivehbasetable(key INT,name STRING,age INT,gender STRING,clazz STRING,last_mod STRING
) STORED BY &＃39;org.apache.hadoop.hive.hbase.HBaseStorageHandler&＃39;
WITH SERDEPROPERTIES ("hbase.columns.mapping" &＃61; ":key,cf1:name,cf1:age,cf1:gender,cf1:clazz,cf1:last_mod") TBLPROPERTIES("hbase.table.name" &＃61; "student");

2&＃xff09;、hbase表

create &＃39;student&＃39;,&＃39;cf1&＃39;


3&＃xff09;、直接执行查询语句&＃xff1a;

select key,name from hivehbasetable limit 10;


hbase外部表 不能使用sqoop直接导入数据&＃xff0c;必须通过例如&＃xff1a;insert into这样的形式导入

因为sqoop导入数据&＃xff0c;使用的原理是load data&＃xff0c;load data只能在表的存储格式为textfile时&＃xff0c;才能真正将数据加载到表中

2、DataX的安装

DataX不需要依赖其他服务&＃xff0c;直接上传、解压、安装、配置环境变量即可

也可以直接在windows上解压

3、DataX的使用

1&＃xff09;、stream2stream

①、编写配置文件stream2stream.json

# stream2stream.json
{"job": {"content": [{"reader": {"name": "streamreader","parameter": {"sliceRecordCount": 10,"column": [{"type": "long","value": "10"},{"type": "string","value": "hello&＃xff0c;你好&＃xff0c;世界-DataX"}]}},"writer": {"name": "streamwriter","parameter": {"encoding": "UTF-8","print": true}}}],"setting": {"speed": {"channel": 5}}}
}


②、执行同步任务

datax.py stream2stream.json


③、执行结果

2&＃xff09;、mysql2mysql

需要新建student2数据库&＃xff0c;并创建student表

①、编写配置文件mysql2mysql.json

{"job": {"content": [{"reader": {"name": "mysqlreader","parameter": {"username": "root","password": "123456","column": ["id","name","age","gender","clazz","last_mod"],"splitPk": "age","connection": [{"table": ["student"],"jdbcUrl": ["jdbc:mysql://master:3306/student"]}]}},"writer": {"name": "mysqlwriter","parameter": {"writeMode": "insert","username": "root","password": "123456","column": ["id","name","age","gender","clazz","last_mod"],"preSql": ["truncate student2"], "connection": [{"jdbcUrl": "jdbc:mysql://master:3306/student2?useUnicode&＃61;true&characterEncoding&＃61;utf8","table": ["student2"]}]}}}],"setting": {"speed": {"channel": 6}}}
}


②、执行同步任务

datax.py mysql2mysql.json


3&＃xff09;、mysql2hdfs

写hive跟hdfs时一样的

①、编写配置文件mysql2hdfs.json

{"job": {"content": [{"reader": {"name": "mysqlreader","parameter": {"username": "root","password": "123456","column": ["id","name","age","gender","clazz","last_mod"],"splitPk": "age","connection": [{"table": ["student"],"jdbcUrl": ["jdbc:mysql://master:3306/student"]}]}},"writer": {"name": "hdfswriter","parameter": {"defaultFS": "hdfs://master:9000","fileType": "text","path": "/user/hive/warehouse/datax.db/students","fileName": "student","column": [{"name": "id","type": "bigint"},{"name": "name","type": "string"},{"name": "age","type": "INT"},{"name": "gender","type": "string"},{"name": "clazz","type": "string"},{"name": "last_mod","type": "string"}],"writeMode": "append","fieldDelimiter": ","}}}],"setting": {"speed": {"channel": 6}}}
}


4&＃xff09;、hbase2mysql

{"job": {"content": [{"reader": {"name": "hbase11xreader","parameter": {"hbaseConfig": {"hbase.zookeeper.quorum": "master:2181"},"table": "student","encoding": "utf-8","mode": "normal","column": [{"name": "rowkey","type": "string"},{"name": "cf1:name","type": "string"},{"name": "cf1:age","type": "string"},{"name": "cf1:gender","type": "string"},{"name": "cf1:clazz","type": "string"}],"range": {"startRowkey": "","endRowkey": "","isBinaryRowkey": false}}},"writer": {"name": "mysqlwriter","parameter": {"writeMode": "insert","username": "root","password": "123456","column": ["id","name","age","gender","clazz"],"preSql": ["truncate student2"], "connection": [{"jdbcUrl": "jdbc:mysql://master:3306/student2?useUnicode&＃61;true&characterEncoding&＃61;utf8","table": ["student2"]}]}}}],"setting": {"speed": {"channel": 6}}}
}


5&＃xff09;、mysql2hbase

mysql中的score表需将cource_id改为course_id&＃xff0c;并将student_id、course_id设为主键&＃xff0c;并将所有字段的类型改为int

hbase需先创建score表&＃xff1a;create ‘score’,‘cf1’

{"job": {"content": [{"reader": {"name": "mysqlreader","parameter": {"username": "root","password": "123456","column": ["student_id","course_id","score"],"splitPk": "course_id","connection": [{"table": ["score"],"jdbcUrl": ["jdbc:mysql://master:3306/student"]}]}},"writer": {"name": "hbase11xwriter","parameter": {"hbaseConfig": {"hbase.zookeeper.quorum": "master:2181"},"table": "score","mode": "normal","rowkeyColumn": [{"index":0,"type":"string"},{"index":-1,"type":"string","value":"_"},{"index":1,"type":"string"}],"column": [{"index":2,"name": "cf1:score","type": "int"}],"encoding": "utf-8"}}}],"setting": {"speed": {"channel": 6}}}
}


4、DataX-web的安装&＃xff08;Java 8、支持Python3&＃xff08;默认Python2.7&＃xff09;、Mysql5.7&＃xff09;

★★★注意★★★&＃xff1a;一定要先配好java、python环境变量

1&＃xff09;、使用idea同步源代码

2&＃xff09;、在mysql5.1中新建dataxweb数据库

3&＃xff09;、执行数据库初始化脚本

4&＃xff09;、修改admin配置文件

• bootstrap.properties

#Database
DB_HOST&＃61;master
DB_PORT&＃61;3306
DB_USERNAME&＃61;root
DB_PASSWORD&＃61;123456
DB_DATABASE&＃61;dataxweb


• application.yml

server:#port: 8080port: 8080
spring:#数据源datasource:# username: root#password: root#url: jdbc:mysql://localhost:3306/datax_web?serverTimezone&＃61;Asia/Shanghai&useLegacyDatetimeCode&＃61;false&useSSL&＃61;false&nullNamePatternMatchesAll&＃61;true&useUnicode&＃61;true&characterEncoding&＃61;UTF-8password: ${DB_PASSWORD:password}username: ${DB_USERNAME:username}url: jdbc:mysql://${DB_HOST:127.0.0.1}:${DB_PORT:3306}/${DB_DATABASE:dataxweb}?serverTimezone&＃61;Asia/Shanghai&useLegacyDatetimeCode&＃61;false&useSSL&＃61;false&nullNamePatternMatchesAll&＃61;true&useUnicode&＃61;true&characterEncoding&＃61;UTF-8driver-class-name: com.mysql.jdbc.Driverhikari:## 最小空闲连接数量minimum-idle: 5## 空闲连接存活最大时间&＃xff0c;默认600000&＃xff08;10分钟&＃xff09;idle-timeout: 180000## 连接池最大连接数&＃xff0c;默认是10maximum-pool-size: 10## 数据库连接超时时间,默认30秒&＃xff0c;即30000connection-timeout: 30000connection-test-query: SELECT 1##此属性控制池中连接的最长生命周期&＃xff0c;值0表示无限生命周期&＃xff0c;默认1800000即30分钟max-lifetime: 1800000# datax-web emailmail:host: smtp.qq.comport: 25#username: xxx&＃64;qq.com#password: xxxusername: rootpassword: rootproperties:mail:smtp:auth: truestarttls:enable: truerequired: truesocketFactory:class: javax.net.ssl.SSLSocketFactorymanagement:health:mail:enabled: falseserver:servlet:context-path: /actuatormybatis-plus:# mapper.xml文件扫描mapper-locations: classpath*:/mybatis-mapper/*Mapper.xml# 实体扫描&＃xff0c;多个package用逗号或者分号分隔#typeAliasesPackage: com.yibo.essyncclient.*.entityglobal-config:# 数据库相关配置db-config:# 主键类型 AUTO:"数据库ID自增", INPUT:"用户输入ID", ID_WORKER:"全局唯一ID (数字类型唯一ID)", UUID:"全局唯一ID UUID";id-type: AUTO# 字段策略 IGNORED:"忽略判断",NOT_NULL:"非 NULL 判断"),NOT_EMPTY:"非空判断"field-strategy: NOT_NULL# 驼峰下划线转换column-underline: true# 逻辑删除logic-delete-value: 0logic-not-delete-value: 1# 数据库类型db-type: mysqlbanner: false# mybatis原生配置configuration:map-underscore-to-camel-case: truecache-enabled: falsecall-setters-on-nulls: truejdbc-type-for-null: &＃39;null&＃39;type-handlers-package: com.wugui.datax.admin.core.handler# 配置mybatis-plus打印sql日志
logging:#level:# com.wugui.datax.admin.mapper: info
#path: ./data/applogs/adminlevel:com.wugui.datax.admin.mapper: errorpath: ./applogs/admin#datax-job, access token
datax:job:accessToken:#i18n (default empty as chinese, "en" as english)i18n:## triggerpool max sizetriggerpool:fast:max: 200slow:max: 100### log retention dayslogretentiondays: 30datasource:aes:key: AD42F6697B035B75


5&＃xff09;、启动DataX-Web-Admin

• 访问: http://127.0.0.1:8080/index.html&＃xff0c;用户名&＃xff1a;admin&＃xff0c;密码&＃xff1a;123456

6&＃xff09;、修改datax-web-executor配置文件

• application.yml

# web port
server:port: 8081#port: 8081# log config
logging:config: classpath:logback.xmlpath: ./applogs/executor/jobhandler#path: ./data/applogs/executor/jobhandlerdatax:job:admin:### datax admin address list, such as "http://address" or "http://address01,http://address02"#addresses: http://127.0.0.1:8080addresses: http://127.0.0.1:8080executor:appname: datax-executorip:#port: 9999port: 9999### job log path#logpath: ./data/applogs/executor/jobhandlerlogpath: ./applogs/executor/jobhandler### job log retention dayslogretentiondays: 30### job, access tokenaccessToken:executor:#jsonpath: D:\\temp\\executor\\json\\jsonpath: ./json#pypath: F:\tools\datax\bin\datax.pypypath: C:\Users\zzk10\Documents\MacOS\DataIntegrate\datax\datax\bin\datax.py


注意pypath&＃xff1a;这个路径需要解压datax.tar.gz到自己的路径&＃xff0c;不要使用中文路径

解压后使用datax-web中的python3脚本替换掉 datax/bin/

7&＃xff09;、启动datax-web-executor

只要进程没有自己停止&＃xff0c;一直在后台运行即可

5、DataX-Web的使用

1&＃xff09;、新建数据源

• mysql

• hive

  需要先启动hiveserver2服务

  下面这条命令需要在linux shell中执行

  hive --service hiveserver2

2&＃xff09;新建项目test1

3&＃xff09;、新建任务模板

4&＃xff09;、构建任务

• 创建reader

• 创建writer

• 字段映射

• 构建任务

• 执行任务

• 查看日志

5&＃xff09;、添加hbase 数据源会报错&＃xff0c;需要修改两个pom.xml文件&＃xff08;hbase1.4.6&＃xff09;

改完pom文件&＃xff0c;记得重新reimport&＃xff0c;才会生效

• 重新测试hbase连通性