datax 持续数据同步_DataX Mongodb同步数据到Hive与mysql

忘是亡心i 2023-01-03 04:49 544阅读 0赞

***前言：***

--------------------

最近比较清闲，又是学习的一天，今天来和大家分享一下阿里的离线数据同步神器——DataX3.0。

目前来说，虽然DataX具有一个全面的插件体系，几乎涵盖了所有的主流数据源，但因为其在使用过程中，编写配置文件其实是一件比较麻烦的事，所以我司只在一些特别的数据源同步任务的时候对其进行了使用，例如今天要和大家分享的，如何使用DataX同步Mogodb的数据到Hive与Mysql。废话不多说，献上干货。![6fcb8b23673604007ad6e5b6c7a07f09.png][]

#  #

## ***DataX3.0概览*** ##

--------------------

#  #

*  **Github主页地址：***https://github.com/alibaba/DataX/blob/master/introduction.md*

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

*  #### **设计理念** ####

![e7e84d4551bcdb4bce5e30e41bc1fe19.png][]

#### **为了解决异构数据源同步问题，DataX将复杂的网状的同步链路变成了星型数据链路，DataX作为中间传输载体负责连接各种数据源。当需要接入一个新的数据源的时候，只需要将此数据源对接到DataX，便能跟已有的数据源做到无缝数据同步。** ####

*  ## **DataX3.0框架设计** ##

## **DataX本身作为离线数据同步框架，采用Framework + plugin架构构建。****将数据源读取和写入抽象成为Reader/Writer插件，纳入到整个同步框架中。** ##

![5728fad4f37c4bb4ff5590934f3e2ea2.png][]

*  Reader：Reader为数据采集模块，负责采集数据源的数据，将数据发送给Framework。
 *  Writer：Writer为数据写入模块，负责不断向Framework取数据，并将数据写入到目的端。
 *  Framework：Framework用于连接reader和writer，作为两者的数据传输通道，并处理缓冲，流控，并发，数据转换等核心技术问题。

# ***如何从mongodb 同步数据到Hive与Mysql*** #

--------------------

明白了DataX的框架设计，对于如何从mongodb同步数据到Hive就非常简单了，无非就是把mongodb作为reader，Hive或者Mysql作为writer，掌握好具体的json配置即可。

*  **mogodb作为reader：**

<table> 
 <tbody> 
  <tr> 
   <td>DataX内部类型</td> 
   <td>MongoDB数据类型</td> 
  </tr> 
  <tr> 
   <td>Long</td> 
   <td>int.Long</td> 
  </tr> 
  <tr> 
   <td>Double</td> 
   <td>double</td> 
  </tr> 
  <tr> 
   <td>String</td> 
   <td><p>string,array</p></td> 
  </tr> 
  <tr> 
   <td>Date</td> 
   <td>date</td> 
  </tr> 
  <tr> 
   <td>Boolean</td> 
   <td>boolean</td> 
  </tr> 
  <tr> 
   <td>Bytes</td> 
   <td>bytes</td> 
  </tr> 
 </tbody> 
</table>

"reader": {
                 "name": "mongodbreader",        "parameter": {
                     "address": ["127.0.0.1:27017"],//address：MongoDB的数据地址信息，因为MonogDB可能是个集群，则ip端口信息需要以Json数组的形式给出。【必填】            "userName": "",//userName：MongoDB的用户名。【选填】            "userPassword": "",//userPassword：MongoDB的密码。【选填】            "dbName": "tag_per_data",//同步的库名            "collectionName": "tag_data12",//collectionName：MonogoDB的集合名。【必填】            "column": [//column：MongoDB的文档列名。【必填】                      {
                                   "name": "categoryid",//name：Column的名字。【必填】                          "type": "Array",//type：Column的类型。【选填】                          "spliter": ""//splitter：因为MongoDB支持数组类型，但是Datax框架本身不支持数组类型，所以mongoDB读出来的数组类型要通过这个分隔符合并成字符串。【选填】                      },                  。。。。。                      {
                                   "name": "scorec",                          "type": "int"                      }            ]        }},

*  **Hive作为writer：**

![7b9acdb5c7612820561387a5ecbac40e.png][]

<table> 
 <tbody> 
  <tr> 
   <td><strong>DataX内部类型</strong></td> 
   <td><strong>Hive数据类型</strong></td> 
  </tr> 
  <tr> 
   <td>Long</td> 
   <td>TINYINT,SMALLINT,INT,BIGINT</td> 
  </tr> 
  <tr> 
   <td>Double</td> 
   <td>FLOAT,DOUBLE</td> 
  </tr> 
  <tr> 
   <td>String</td> 
   <td>STRING,VARCHAR,CHAR</td> 
  </tr> 
  <tr> 
   <td>Boolean</td> 
   <td>BOOLEAN</td> 
  </tr> 
  <tr> 
   <td>Date</td> 
   <td>DATE,TIMESTAMP</td> 
  </tr> 
 </tbody> 
</table>

"writer": {
               "name": "hdfswriter"      "parameter": {
                 "path": "/user/hive/warehouse/tmp_mongo_baofoo_log.db/tmp_log_fi_auth_payment",//存储到Hadoop hdfs文件系统的路径信息，HdfsWriter会根据并发配置在Path目录下写入多个文件。为与hive表关联，请填写hive表在hdfs上的存储路径。                        //例：Hive上设置的数据仓库的存储路径为：/user/hive/warehouse/ ，已建立数据库：test，表：hello；则对应的存储路径为：/user/hive/warehouse/test.db/hello        "fileName": "tmp_log_fi_auth_payment",//HdfsWriter写入时的文件名，实际执行时会在该文件名后添加随机的后缀作为每个线程写入实际文件名。        "haveKerberos": "false",//是否有Kerberos认证，默认false 例如如果用户配置true，则配置项kerberosKeytabFilePath，kerberosPrincipal为必填。        "compress": "SNAPPY",//hdfs文件压缩类型，默认不填写意味着没有压缩。其中：text类型文件支持压缩类型有gzip、bzip2;orc类型文件支持的压缩类型有NONE、SNAPPY(需要用户安装SnappyCodec)。        "column": [//写入数据的字段，不支持对部分列写入。为与hive中表关联，需要指定表中所有字段名和字段类型，其中：name指定字段名，type指定字段类型。               {
                           "name": "trade_no",                  "type": "string"                },       。。。。。。。              {
                           "name": "mobile",                  "type": "string"                }        ],        "defaultFS": "hdfs://172.20.85.39:8020", //Hadoop hdfs文件系统namenode节点地址。        "writeMode": "append", //hdfswriter写入前数据清理处理模式：                  //append，写入前不做任何处理，DataX hdfswriter直接使用filename写入，并保证文件名不冲突。                  //nonConflict，如果目录下有fileName前缀的文件，直接报错。        "fieldDelimiter": " ",//hdfswriter写入时的字段分隔符,需要用户保证与创建的Hive表的字段分隔符一致，否则无法在Hive表中查到数据        "fileType": "orc" //文件的类型，目前只支持用户配置为"text"或"orc"。text表示textfile文件格式   orc表示orcfile文件格式      },     }

*  **Mysql作为writer：**

<table> 
 <tbody> 
  <tr> 
   <td><strong>DataX 内部类型</strong></td> 
   <td><strong>Mysql 数据类型</strong></td> 
  </tr> 
  <tr> 
   <td>Long</td> 
   <td>int, tinyint, smallint, mediumint, int, bigint, year</td> 
  </tr> 
  <tr> 
   <td>Double</td> 
   <td>float, double, decimal</td> 
  </tr> 
  <tr> 
   <td>String</td> 
   <td>varchar, char, tinytext, text, mediumtext, longtext</td> 
  </tr> 
  <tr> 
   <td>Date</td> 
   <td>date, datetime, timestamp, time</td> 
  </tr> 
  <tr> 
   <td>Boolean</td> 
   <td>bit, bool</td> 
  </tr> 
  <tr> 
   <td>Bytes</td> 
   <td>tinyblob, mediumblob, blob, longblob, varbinary</td> 
  </tr> 
 </tbody> 
</table>

"writer": {
                "name": "mysqlwriter"        "parameter": {
                   "password": "******",//目的数据库的密码          "session": [            "set session sql_mode='ANSI'"// DataX在获取Mysql连接时，执行session指定的SQL语句，修改当前connection session属性          ],          "column": [            "operator_ip",             。。。。//目的表需要写入数据的字段,字段之间用英文逗号分隔。例如: "column": ["id","name","age"]。如果要依次写入全部列，使用表示, 例如: "column": [""]。              "member_id"          ],          "connection": [            {
                       "jdbcUrl": "jdbc:mysql://192.168.6.160:3306/BAOFOO_STAT",//目的数据库的 JDBC 连接信息。作业运行时，DataX 会在你提供的 jdbcUrl 后面追加如下属性：yearIsDateType=false&zeroDateTimeBehavior=convertToNull&rewriteBatchedStatements=true                    //     注意：1、在一个数据库上只能配置一个 jdbcUrl 值。这与 MysqlReader 支持多个备库探测不同，因为此处不支持同一个数据库存在多个主库的情况(双主导入数据情况)                    //          2、jdbcUrl按照Mysql官方规范，并可以填写连接附加控制信息，比如想指定连接编码为 gbk ，则在 jdbcUrl 后面追加属性 useUnicode=true&characterEncoding=gbk。具体请参看 Mysql官方文档或者咨询对应 DBA。              "table": [                "mf_operator_log_today" //目的表的表名称。支持写入一个或者多个表。当配置为多张表时，必须确保所有表结构保持一致。              ]            }          ],          "writeMode": "insert",//控制写入数据到目标表采用 insert into 或者 replace into 或者 ON DUPLICATE KEY UPDATE 语句                 //所有选项：insert/replace/update          "username": "cs_hadoop",//目的数据库的用户名          "preSql": [            "delete from BAOFOO_STAT.mf_operator_log_today where create_time<='2000-01-01'"            //写入数据到目的表前，会先执行这里的标准语句。如果 Sql 中有你需要操作到的表名称，请使用 @table 表示，这样在实际执行 Sql 语句时，会对变量按照实际表名称进行替换。比如你的任务是要写入到目的端的100个同构分表(表名称为:datax_00,datax01, ... datax_98,datax_99)，并且你希望导入数据前，先对表中数据进行删除操作，那么你可以这样配置："preSql":["delete from 表名"]，效果是：在执行到每个表写入数据前，会先执行对应的 delete from 对应表名称            //postSql：            //描述：写入数据到目的表后，会执行这里的标准语句。(原理同 preSql )          ]        }}

最后，将以上的模版内容以拼积木的形式，进行job content的拼接，完成相应的功能需求，例如：

{
          "core":{//可通过增加如下的core参数，去除掉DataX默认对速度的限制；        "transport":{
                     "channel":{
                         "speed":{
                             "channel": 15,                     "record":-1,                     "byte":-1,                     "batchSize":2048                }            }        }    },  "job": {
               "setting": {//流量控制        "speed": {
                   "channel": "20",//同步时候的并发数              //- Channel个数的增加，带来的是更多的CPU消耗以及内存消耗。             //- 如果Channel并发配置过高导致JVM内存不够用，会出现的情况是发生频繁的Full GC，导出速度会骤降，适得其反。这个可以通过观察日志发现          "record":-1,           "byte":-1, //同步时候的字节大小          "batchSize":2048         },     "errorLimit": {   //脏数据控制                       //Job支持用户对于脏数据的自定义监控和告警，包括对脏数据最大记录数阈值(record值)或者脏数据占比阈值(percentage值)，当Job传输过程出现的脏数据大于用户指定的数量/百分比，DataX Job报错退出。                "record": 0,                "percentage": 0.02            }        },    "content": [      {
                 "reader": {//对应数据采集模块         。。。        },        "writer": {//对应数据写入模块         。。。        }      }    ]  }}

***总结：***

--------------------

大数据是条不归路啊，加油吧，骚年！![9689cc34e2ee78b2e2506bf4051333d4.png][]![9689cc34e2ee78b2e2506bf4051333d4.png][]![9689cc34e2ee78b2e2506bf4051333d4.png][]

![af93c41492f3655dc84445d1d27e0b4d.gif][]

[6fcb8b23673604007ad6e5b6c7a07f09.png]: /images/20221119/98293c550f234d5eb5408951cb3fc11e.png
[e7e84d4551bcdb4bce5e30e41bc1fe19.png]: /images/20221119/8f4070b8b48f408ea8cce56d04d4c09e.png
[5728fad4f37c4bb4ff5590934f3e2ea2.png]: /images/20221119/1af2e015aab2489f92f132a80634bbbc.png
[7b9acdb5c7612820561387a5ecbac40e.png]: /images/20221119/c4798b2203284c61b94b921686a13daf.png
[9689cc34e2ee78b2e2506bf4051333d4.png]: /images/20221119/01e5dfb19c1d446aac926b996d287a62.png
[af93c41492f3655dc84445d1d27e0b4d.gif]: /images/20221119/695be730106b444e9b763677ea7ef69a.png