基于datax抽取mysql数据到HDFS

一、安装datax

通过https://datax-opensource.oss-cn-hangzhou.aliyuncs.com/202210/datax.tar.gz这个地址下载datax.tar.gz，解压到某个目录，如我的目录/opt/conf/datax-20230301，解压完后会在当前目录下生成datax目录，进入datax目录后的目录如下图所示：

之后在datax安装目录下，运行以下命令，赋予执行权限。

1. sudo chmod -R 755 ./*

复制代码

二、测试datax是否正确安装

1. /opt/conf/datax-20230301/datax/bin/datax.py /opt/conf/datax-20230301/datax/job/job.json

复制代码

运行以上命令，看是否能正确启动，启动后运行完结果如下图：

如果那个正确运行，说明/opt/conf/datax-20230301/datax/bin/datax.py这个文件的编码不是utf-8，需要重新编码。用我这个替换一下即可正常使用。
datax.py
三、编写配置文件

在datax安装目录下的job文件夹，使用以下命令新建配置文件

1. vim job_air_data_source_mysql_hdfs.json

复制代码

之后将下面的json文件内容拷贝粘贴到刚才打开的文件，保存即可。

1. {
2.   "job": {
3.     "setting": {
4.       "speed": {
5.         "channel": 3
6.       },
7.       "errorLimit": {
8.         "record": 0,
9.         "percentage": 0.02
10.       }
11.     },
12.     "content": [
13.       {
14.         "reader": {
15.           "name": "mysqlreader",
16.           "parameter": {
17.             "username": "root",
18.             "password": "root",
19.             "column": ["*"],
20.             "splitPk": "id",
21.             "connection": [
22.               {
23.                 "table": [
24.                   "air_data_source"
25.                 ],
26.                 "jdbcUrl": [
27.                   "jdbc:mysql://master:3306/air_data"
28.                 ]
29.               }
30.             ]
31.           }
32.         },
33.         "writer": {
34.           "name": "hdfswriter",
35.           "parameter": {
36.             "defaultFS": "hdfs://master:9820",
37.             "fileType": "TEXT",
38.             "path": "/user/hive/warehouse/air_data.db/air_data_source",
39.             "fileName": "air_data_source_202302",
40.             "column": [
41.               {"name": "        id        ","type": "STRING"},       
42.               {"name": "        airlinelogo        ","type": "STRING"},       
43.               {"name": "        airlineshortcompany        ","type": "STRING"},       
44.               {"name": "        arractcross        ","type": "STRING"},       
45.               {"name": "        arracttime        ","type": "STRING"},       
46.               {"name": "        arrairport        ","type": "STRING"},       
47.               {"name": "        arrcode        ","type": "STRING"},       
48.               {"name": "        arrontimerate        ","type": "STRING"},       
49.               {"name": "        arrplancross        ","type": "STRING"},       
50.               {"name": "        arrplantime        ","type": "STRING"},       
51.               {"name": "        arrterminal        ","type": "STRING"},       
52.               {"name": "        checkintable        ","type": "STRING"},       
53.               {"name": "        checkintablewidth        ","type": "STRING"},       
54.               {"name": "        depactcross        ","type": "STRING"},       
55.               {"name": "        depacttime        ","type": "STRING"},       
56.               {"name": "        depairport        ","type": "STRING"},       
57.               {"name": "        depcode        ","type": "STRING"},       
58.               {"name": "        depplancross        ","type": "STRING"},       
59.               {"name": "        depplantime        ","type": "STRING"},       
60.               {"name": "        depterminal        ","type": "STRING"},       
61.               {"name": "        flightno        ","type": "STRING"},       
62.               {"name": "        flightstate        ","type": "STRING"},       
63.               {"name": "        localdate        ","type": "STRING"},       
64.               {"name": "        mainflightno        ","type": "STRING"},       
65.               {"name": "        shareflag        ","type": "STRING"},       
66.               {"name": "        statecolor        ","type": "STRING"}       
67.             ],
68.               "writeMode": "truncate",
69.               "fieldDelimiter": "\u0001",
70.               "compress":"GZIP"
71.               }
72.               }
73.               }
74.             ]
75.               }
76.               }

复制代码

四、Hive建数据库、数据表

1. create database air_data;
2. use air_data;
3. CREATE TABLE `air_data_source`(
4.   `id` int COMMENT '主键',
5.   `airlinelogo` string COMMENT '航空公司logo',
6.   `airlineshortcompany` string COMMENT '航空公司简称',
7.   `arractcross` string,
8.   `arracttime` string COMMENT '实际起飞时间',
9.   `arrairport` string,
10.   `arrcode` string,
11.   `arrontimerate` string COMMENT '到达准点率',
12.   `arrplancross` string,
13.   `arrplantime` string COMMENT '计划到达时间',
14.   `arrterminal` string,
15.   `checkintable` string,
16.   `checkintablewidth` string,
17.   `depactcross` string,
18.   `depacttime` string COMMENT '实际到达时间',
19.   `depairport` string COMMENT '到达机场名称',
20.   `depcode` string COMMENT '到达机场代码',
21.   `depplancross` string,
22.   `depplantime` string COMMENT '计划起飞时间',
23.   `depterminal` string,
24.   `flightno` string COMMENT '航班号',
25.   `flightstate` string COMMENT '航班状态',
26.   `localdate` string,
27.   `mainflightno` string,
28.   `shareflag` string,
29.   `statecolor` string)
30. COMMENT '航空数据原始表'
31. ROW FORMAT SERDE
32.   'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
33. STORED AS INPUTFORMAT
34.   'org.apache.hadoop.mapred.TextInputFormat'
35. OUTPUTFORMAT
36.   'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
37. ;

复制代码

运行完以上任务后，接着可以进行数据抽取了。
四、运行任务

在当前目录下执行以下命令：

1. /opt/conf/datax-20230301/datax/bin/datax.py /opt/conf/datax-20230301/datax/job/job_air_data_source_mysql_hdfs.json

复制代码

即可正确启动数据同步任务，运行完结果如下：

查看HDFS上是否已经有了数据文件，运行一下命令，得到输出。

1. hadoop fs -ls hdfs://master:9820/user/hive/warehouse/air_data.db/air_data_source

复制代码

至此，利用datax将mysql数据同步到hdfs任务已配置完成。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！