DataX（Doris同步数据到SelectDB）

背景

由于之前的doris数仓在当地的服务器，当数据量越来越大，服务器的性能达不到要求，查询数据常常超时，故需要把当地的doris数仓部署到云上，本文以阿里云为例，迁移工具使用的阿里开源的datax。
datax官方文档
预备

起首使用navicat连接上源数据库与目标数据库

阿里云数仓

在目标数据创建需要迁移的表

源数据库上面执行下面的语句得到建表语句

1. show CREATE TABLE your_table;

复制代码

然后拿到建表语句去目标数据库执行。
datax工具安装

情况

   Linux大概windows
JDK(1.8以上，保举1.8)
Python(2或3都可以)
Apache Maven 3.x (Compile DataX)
  方法一、直接下载DataX工具包：DataX下载地点

下载后解压至当地某个目录，进入bin目录，即可运行同步作业：

1. $ cd  {YOUR_DATAX_HOME}/bin
2. $ python datax.py {YOUR_JOB.json}

复制代码

自检脚本：  python {YOUR_DATAX_HOME}/bin/datax.py {YOUR_DATAX_HOME}/job/job.json
方法二、下载DataX源码，自己编译：DataX源码

(1)、下载DataX源码：

1. $ git clone git@github.com:alibaba/DataX.git

复制代码

(2)、通过maven打包：

1. $ cd  {DataX_source_code_home}
2. $ mvn -U clean package assembly:assembly -Dmaven.test.skip=true

复制代码

打包乐成，日志显示如下：

1. [INFO] BUILD SUCCESS
2. [INFO] -----------------------------------------------------------------
3. [INFO] Total time: 08:12 min
4. [INFO] Finished at: 2015-12-13T16:26:48+08:00
5. [INFO] Final Memory: 133M/960M
6. [INFO] -----------------------------------------------------------------

复制代码

打包乐成后的DataX包位于 {DataX_source_code_home}/target/datax/datax/，布局如下：

1. $ cd  {DataX_source_code_home}
2. $ ls ./target/datax/datax/
3. bin                conf                job                lib                log                log_perf        plugin                script                tmp

复制代码

配置示例：从stream读取数据并打印到控制台

第一步、创建作业的配置文件（json格式）
可以通过命令检察配置模板： python datax.py -r {YOUR_READER} -w {YOUR_WRITER}
$ cd {YOUR_DATAX_HOME}/bin
$ python datax.py -r streamreader -w selectdbwriter

1. PS C:\Users\zhubayi\Downloads\datax\datax\bin> python datax.py -r streamreader -w selectdbwriter
3. DataX (DATAX-OPENSOURCE-3.0), From Alibaba !
4. Copyright (C) 2010-2017, Alibaba Group. All Rights Reserved.
7. Please refer to the streamreader document:
8.      https://github.com/alibaba/DataX/blob/master/streamreader/doc/streamreader.md
10. Please refer to the selectdbwriter document:
11.      https://github.com/alibaba/DataX/blob/master/selectdbwriter/doc/selectdbwriter.md
13. Please save the following configuration as a json file and  use
14.      python {DATAX_HOME}/bin/datax.py {JSON_FILE_NAME}.json
15. to run the job.
17. {
18.     "job": {
19.         "content": [
20.             {
21.                 "reader": {
22.                     "name": "streamreader",
23.                     "parameter": {
24.                         "column": [],
25.                         "sliceRecordCount": ""
26.                     }
27.                 },
28.                 "writer": {
29.                     "name": "selectdbwriter",
30.                     "parameter": {
31.                         "column": [],
32.                         "connection": [
33.                             {
34.                                 "jdbcUrl": "",
35.                                 "selectedDatabase": "",
36.                                 "table": []
37.                             }
38.                         ],
39.                         "loadProps": {},
40.                         "loadUrl": [],
41.                         "password": "",
42.                         "postSql": [],
43.                         "preSql": [],
44.                         "username": ""
45.                     }
46.                 }
47.             }
48.         ],
49.         "setting": {
50.             "speed": {
51.                 "channel": ""
52.             }
53.         }
54.     }
55. }

复制代码

dorisreader和selectdbwriter配置

1. {
2. "core":{
3.     "transport": {
4.       "channel": {
5.         "speed": {
6.           "byte": 536870912000,
7.           "record": 1000000
8.         }
9.       }
10.     }
11.   },
12.   "job": {
13.     "content": [{
14.       "reader": {
15.         "name": "dorisreader",
16.         "parameter": {
17.           "column": ["id", "name", "nick_name", "age"],
18.           "connection": [{
19.             "jdbcUrl": ["jdbc:mysql://127.0.0.1:9030/test?useSSL=false"],
20.             "table": ["t1"]
21.           }],
22.           "username": "root",
23.           "password": "123456",
24.           "splitPk": "id",
25.           "where": ""
26.         }
27.       },
28.       "writer": {
29.         "name": "selectdbwriter",
30.         "parameter": {
31.           "loadUrl": [
32.             "192.168.3.123:8080"
33.           ],
34.           "loadProps": {
35.             "file.type": "json",
36.             "file.strip_outer_array": true
37.           },
38.           "column": ["id", "name", "nick_name", "age"],
39.           "username": "admin",
40.           "password": "123456",
41.           "postSql": [
43.           ],
44.           "preSql": [
46.           ],
47.           "connection": [{
48.             "jdbcUrl": "jdbc:mysql://192.168.3.123:9030/test?useSSL=false",
49.             "table": [
50.               "t1"
51.             ],
52.             "selectedDatabase": "test"
53.           }],
54.           "maxBatchRows": 1000000,
55.           "maxBatchByteSize": 536870912000
56.         }
57.       }
58.     }],
59.     "setting": {
60.       "speed": {
61.         "channel": "1"
62.       },
63.       "errorLimit": {
64.         "record": 0,
65.         "percentage": 0
66.       }
67.     }
68.   }
69. }

复制代码

这个任务配置文件包罗了一个名为“job”的任务，它有两个子节点：setting和content。此中，setting节点包罗了任务的一些设置，例如并发通道数；content节点则包罗了数据读取和写入的详细配置信息。
core与job参数先容

配置说明core.transport.channel.speed.byte单个channel容纳最多的字节数core.transport.channel.speed.record单个channel容纳最多的record数job.setting.speed.channel该job所需要的channel的个数job.setting.speed.byte该job最大的流量job.setting.speed.record该job最大的record流量 reader参数说明

官方dorisreader配置地点

• jdbcUrl
  
  
  
  • 形貌：形貌的是到对端数据库的JDBC连接信息，使用JSON的数组形貌，并支持一个库填写多个连接地点。之以是使用JSON数组形貌连接信息，是因为阿里集团内部支持多个IP探测，假如配置了多个，DorisReader可以依次探测ip的可连接性，直到选择一个合法的IP。假如全部连接失败，DorisReader报错。 留意，jdbcUrl必须包罗在connection配置单元中。对于阿里集团外部使用情况，JSON数组填写一个JDBC连接即可。
    
  • 必选：是
    
  • 默认值：无
    
  
  
• username
  
  
  
  • 形貌：数据源的用户名
    
  • 必选：是
    
  • 默认值：无
    
  
  
• password
  
  
  
  • 形貌：数据源指定用户名的暗码
    
  • 必选：是
    
  • 默认值：无
    
  
  
• table
  
  
  
  • 形貌：所选取的需要同步的表。使用JSON的数组形貌，因此支持多张表同时抽取。当配置为多张表时，用户自己需保证多张表是同一schema布局，DorisReader不予检查表是否同一逻辑表。留意，table必须包罗在connection配置单元中。
    
  • 必选：是
    
  • 默认值：无
    
  
  
• column
  
  
  
  • 形貌：所配置的表中需要同步的列名集合，使用JSON的数组形貌字段信息。用户使用代表默认使用所有列配置，例如['']。
    支持列裁剪，即列可以挑选部分列举行导出。
    支持列换序，即列可以不按照表schema信息举行导出。
    支持常量配置，用户需要按照Doris SQL语法格式: [“id”, “table”, “1”, “‘bazhen.csy’”, “null”, “to_char(a + 1)”, “2.3” , “true”] id为普通列名，table为包罗保留字的列名，1为整形数字常量，'bazhen.csy’为字符串常量，null为空指针，to_char(a + 1)为表达式，2.3为浮点数，true为布尔值。
    
  • 必选：是
    
  • 默认值：无
    
  
  
• splitPk
  
  
  
  • 形貌：DorisReader举行数据抽取时，假如指定splitPk，表示用户希望使用splitPk代表的字段举行数据分片，DataX因此会启动并发任务举行数据同步，这样可以大大提供数据同步的效能。
    保举splitPk用户使用表主键，因为表主键通常情况下比较均匀，因此切分出来的分片也不容易出现数据热门。
    目前splitPk仅支持整形数据切分，不支持浮点、字符串、日期等其他类型。假如用户指定其他非支持类型，DorisReader将报错！
    假如splitPk不填写，包括不提供splitPk大概splitPk值为空，DataX视作使用单通道同步该表数据。
    
  • 必选：否
    
  • 默认值：空
    
  
  
• where
  
  
  
  • 形貌：筛选条件，DorisReader根据指定的column、table、where条件拼接SQL，并根据这个SQL举行数据抽取。在实际业务场景中，通常会选择当天的数据举行同步，可以将where条件指定为gmt_create > $bizdate 。留意：不可以将where条件指定为limit 10，limit不是SQL的合法where子句。
    where条件可以有用地举行业务增量同步。假如不填写where语句，包括不提供where的key大概value，DataX均视作同步全量数据。
    
  • 必选：否
    
  • 默认值：无
    
  
  
• querySql
  
  
  
  • 形貌：在有些业务场景下，where这一配置项不敷以形貌所筛选的条件，用户可以通过该配置型来自定义筛选SQL。当用户配置了这一项之后，DataX体系就会忽略table，column这些配置型，直接使用这个配置项的内容对数据举行筛选，例如需要举行多表join后同步数据，使用select a,b from table_a join table_b on table_a.id = table_b.id
    当用户配置querySql时，DorisReader直接忽略table、column、where条件的配置，querySql优先级大于table、column、where选项。
    
  • 必选：否
    
  • 默认值：无
    
  
  

类型转换
目前DorisReader支持大部分Doris类型，但也存在部分个别类型没有支持的情况，请留意检查你的类型。
下面列出DorisReaderr针对Doris类型转换列表:
DataX 内部类型doris 数据类型Longint, tinyint, smallint, int, bigint,LargintDoublefloat, double, decimalStringvarchar, char, text, string, map, json, array, structDatedate, datetimeBooleanBoolean 请留意:
tinyint(1) DataX视作为整形。
writer配置参数说明

官方selectdbwriter配置地点

• jdbcUrl
  
  
  
  • 形貌：selectdb 的 JDBC 连接串，用户执行 preSql 或 postSQL。
    
  • 必选：是
    
  • 默认值：无
    
  
  
• loadUrl
  
  
  
  • 形貌：作为 selecdb 的连接目标。格式为 “ip:port”。此中 IP 是 selectdb的private-link，port 是selectdb 集群的 http_port
    
  • 必选：是
    
  • 默认值：无
    
  
  
• username
  
  
  
  • 形貌：访问selectdb数据库的用户名
    
  • 必选：是
    
  • 默认值：无
    
  
  
• password
  
  
  
  • 形貌：访问selectdb数据库的暗码
    
  • 必选：否
    
  • 默认值：空
    
  
  
• connection.selectedDatabase
  
  
  
  • 形貌：需要写入的selectdb数据库名称。
    
  • 必选：是
    
  • 默认值：无
    
  
  
• connection.table
  
  
  
  • 形貌：需要写入的selectdb表名称。
    
    
    
    • 必选：是
      
    • 默认值：无
      
    
    
  
  
• column
  
  
  
  • 形貌：目标表需要写入数据的字段，这些字段将作为天生的 Json 数据的字段名。字段之间用英文逗号分隔。例如: “column”: [“id”,“name”,“age”]。
    
  • 必选：是
    
  • 默认值：否
    
  
  
• preSql
  
  
  
  • 形貌：写入数据到目标表前，会先执行这里的尺度语句。
    
  • 必选：否
    
  • 默认值：无
    
  
  
• postSql
  
  
  
  • 形貌：写入数据到目标表后，会执行这里的尺度语句。
    
  • 必选：否
    
  • 默认值：无
    
  
  
• maxBatchRows
  
  
  
  • 形貌：每批次导入数据的最大行数。和 batchSize 共同控制每批次的导入数量。每批次数据到达两个阈值之一，即开始导入这一批次的数据。
    
  • 必选：否
    
  • 默认值：500000
    
  
  
• batchSize
  
  
  
  • 形貌：每批次导入数据的最大数据量。和 maxBatchRows 共同控制每批次的导入数量。每批次数据到达两个阈值之一，即开始导入这一批次的数据。
    
  • 必选：否
    
  • 默认值：90M
    
  
  
• maxRetries
  
  
  
  • 形貌：每批次导入数据失败后的重试次数。
    
  • 必选：否
    
  • 默认值：3
    
  
  
• labelPrefix
  
  
  
  • 形貌：每批次上传文件的 label 前缀。最终的 label 将有 labelPrefix + UUID 构成全局唯一的 label，确保数据不会重复导入
    
  • 必选：否
    
  • 默认值：datax_selectdb_writer_
    
  
  
• loadProps
  
  
  
  • 形貌：COPY INOT 的请求参数
    这里包括导入的数据格式：file.type等，导入数据格式默认我们使用csv，支持JSON，具体可以参照下面类型转换部分
    
  • 必选：否
    
  • 默认值：无
    
  
  
• clusterName
  
  
  
  • 形貌：selectdb could 集群名称
    
  • 必选：否
    
  • 默认值：无
    
  
  
• flushQueueLength
  
  
  
  • 形貌：队列长度
    
  • 必选：否
    
  • 默认值：1
    
  
  
• flushInterval
  
  
  
  • 形貌：数据写入批次的时间隔断，假如maxBatchRows 和 batchSize 参数设置的有很大，那么很大概达不到你这设置的数据量巨细，会执行导入。
    
  • 必选：否
    
  • 默认值：30000ms
    
  
  

类型转换
默认传入的数据均会被转为字符串，并以\t作为列分隔符，\n作为行分隔符，构成csv文件举行Selectdb导入操纵。
默认是csv格式导入，如需更改列分隔符， 则正确配置 loadProps 即可：

1. "loadProps": {
2.     "file.column_separator": "\\x01",
3.     "file.line_delimiter": "\\x02"
4. }

复制代码

如需更改导入格式为json， 则正确配置 loadProps 即可：

1. "loadProps": {
2.     "file.type": "json",
3.     "file.strip_outer_array": true
4. }

复制代码

启动DataX

1. $ cd {YOUR_DATAX_DIR_BIN}
2. $ python datax.py ./doris2selecdb.json

复制代码

同步竣事，显示日志如下：

1. 2024-08-26 17:10:18.721 [job-0] INFO  JobContainer - PerfTrace not enable!
2. 2024-08-26 17:10:18.721 [job-0] INFO  StandAloneJobContainerCommunicator - Total 12976386 records, 1444190303 bytes | Speed 2.33MB/s, 21993 records/s | Error 0 records, 0 bytes |  All Task WaitWriterTime 28.820s |  All Task WaitReaderTime 308.386s | Percentage 100.00%
3. 2024-08-26 17:10:18.722 [job-0] INFO  JobContainer -
4. 任务启动时刻                    : 2024-08-26 17:00:24
5. 任务结束时刻                    : 2024-08-26 17:10:18
6. 任务总计耗时                    :                593s
7. 任务平均流量                    :            2.33MB/s
8. 记录写入速度                    :          21993rec/s
9. 读出记录总数                    :            12976386
10. 读写失败总数                    :                   0

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。