• 数据集成工具 ---- datax 3.0

    1、datax:

            是一个异构数据源离线同步工具，致力于实现关系型数据库（mysql、oracle等）hdfs、hive、hbase等各种异构数据源之间的数据同步

    2、参考网址文献：

    https://github.com/alibaba/DataX/blob/master/introduction.mdhttps://github.com/alibaba/DataX/blob/master/userGuid.mdhttps://github.com/alibaba/DataX/blob/master/introduction.md

    3、Datax的框架设计：

    Datax作为离线数据同步工具，主要的是采用了Framework+plugin架构构成，将数据源的读数据和写数据封装成对应的Reader和Writer插件，纳入到整体的同步框架中。

            1、Reader：作为数据的采集模块，负责采集数据源的数据，将数据发送给Framework

            2、Writer：作为数据写入模块，负责不断的向Framework取出数据，将数据写入到对应的目的端

            3、Framework：用于连接Reader和Writer，作为两者的数据传输通道，并处理缓冲、数据转换等核心技术问题。

    4、Datax的核心架构：

    Datax3.0 开源版本是支持单机多线程来完成同步作业运行，因为底层是使用java做开发。整体的架构：

    模块的核心介绍：

            1、Datax完成单个数据同步做作业，被称之为job，Datax接收到一个job时就会启动一个进程来完成数据同步工作，所以Datax job 模块是单个作业的中枢管理中心，主要是承担数据清理，子任务切分、TaskGroup 管理。

            2、当Datax启动后，Datax job会根据不同的源数据将job切分成不同的Task，所以Task是Datax的最小作业单位，每一个Task都会负责一部分的数据同步。

            3、切分成多个Task后，Datax job 就会调用scheduler模块，根据配置的并发数量，将拆分的Task重新组合，组装成TaskGroup，每一个TaskGroup都负责一定的Task任务的执行，默认TaskGroup并发数量数5个。

            4、每一个Task都是由TaskGroup所监控执行启动，每一个Task启动后都会按照Reader---Channel---Writer的执行顺序执行。

            5、当任务启动后，Datax job就会监控所有的TaskGroup的执行情况，当所有的TaskGroup任务完成后，job就会退出，当出现异常，就会异常退出并且进程退出值非0.

    5、Datax的核心优势：

            1、可靠的数据质量监控

            2、丰富的数据转换功能

            3、精准的控制速度

            4、容错机制：

                    1、线程内部重试

                            DataX的核心插件都经过团队的全盘review，不同的网络交互方式都有不同的重试策略。

                    2、线程级别重试

                            目前DataX已经可以实现TaskFailover，针对于中间失败的Task，DataX框架可以做到整个Task级别的重新调度。

            5、极简的体验

    6、Datax与Sqoop的区别：
    功能Dataxsqoop
    运行模式单进程多线程MR
    分布式是不支持分布式支持
    流控有需要定制
    统计信息支持不支持，分布式的数据收集不方便
    数据校验只有core部分有校验功能不支持，分布式的数据收集不方便
    监控需要定制需要定制
    7、Datax部署：

    1、下载jar包：

            下载路径：https://github.com/alibaba/DataX

    2、解压文件，配置环境变量：  

    1. #解压jar包
    2. tar -zxvf datax.tar.gz
    3.  
    4. #配置环境变量：
    5. vim /etc/profile
    6.  
    7.  
    8. export   DATAX_HOME=/user/loacl/soft/datax
    9. export   PATH=.:$PATH:$DATAX_HOME/bin
    10.  
    11. #配置好环境变量，让配置文件生效
    12. source /etc/profile

    3、使执行文件拥有执行权：

    1. 添加执行权：
    2.  
    3. chmod +x  data.py
    8、Datax的使用：
    1. 在datax中会自动的生成模板的命令：
    2.  
    3. datax.py -r streamreader -w streamwriter
            1、streamreader  to  streamwriter，数据打印在控制台上面

    参数说明：

    1. "sliceRecordCount": 100  #指定打印的个数
    2.  
    3. "channel": 1  #指定并发度
    1. #创建json文件：
    2.  
    3. vim streamreadertostreamwriter.json
    4.  
    5.  
    6. {
    7.     "job": {
    8.         "content": [
    9.             {
    10.                 "reader": {
    11.                     "name": "streamreader", 
    12.                     "parameter": {
    13.                         "column": [
    14.                             {
    15.                                 "type":"string",
    16.                                 "value":"wyz"
    17.                             },
    18.                             {
    19.                                 "type":"int",
    20.                                 "value":"18"
    21.                             }
    22.                         ], 
    23.                         "sliceRecordCount": 100  #指定打印的个数
    24.                     }
    25.                 }, 
    26.                 "writer": {
    27.                     "name": "streamwriter", 
    28.                     "parameter": {
    29.                         "encoding": "", 
    30.                         "print": true
    31.                     }
    32.                 }
    33.             }
    34.         ], 
    35.         "setting": {
    36.             "speed": {
    37.                 "channel": 1  #指定并发度
    38.             }
    39.         }
    40.     }
    41. }
    42.  
    43.  
    44.  
    45. #脚本执行命令：
    46. datax.py streamreadertostreamwriter.json
            2、mysql to mysql
    1. 1、可以通过命令获取模板：
    2. datax.py  -r mysqlreader -w mysqlwriter 
    3.  
    4. 2、可以通过github上的模板进行编写：分别是mysqlreader和mysqlwriter，参数会比较详细
    5.  
    6.  
    7. 3、在插入数据的需要注意是在将数据写入的时候如果出现在数据，那么此时可能是创建的表出了问题
    8.     例如：表中的某个字段是主键，主键唯一
    1. vim mysqlreaderTomysqlwriter.json
    2.  
    3. {
    4.     "job": {
    5.         "content": [
    6.             {
    7.                 "reader": {
    8.                     "name": "mysqlreader", 
    9.                     "parameter": {
    10.                         "column": [
    11.                                 "id",
    12. 							 "name",
    13. 							 "age",
    14. 							 "clazz",
    15. 							 "gender"
    16.                         
    17.                         ], 
    18.                         "connection": [
    19.                             {
    20.                                 "jdbcUrl": ["jdbc:mysql://192.168.226.1:3306/bigdata25"], 
    21.                                 "table": ["stu"]
    22.                             }
    23.                         ], 
    24.                         "password": "123456", 
    25.                         "username": "root", 
    26.                         "where": ""      #不是必须要写的，作用是可以在读数据时进行一次过滤
    27.                     }
    28.                 }, 
    29.                 "writer": {
    30.                     "name": "mysqlwriter", 
    31.                     "parameter": {
    32.                         "column": [
    33.                                 "id",
    34. 							 "name",
    35. 							 "age",
    36. 							 "clazz",
    37. 							 "gender"
    38.                         ], 
    39.                         "connection": [
    40.                             {
    41.                                 "jdbcUrl": "jdbc:mysql://192.168.226.1:3306/bigdata25", 
    42.                                 "table": ["data_test"]
    43.                             }
    44.                         ], 
    45.                         "password": "123456", 
    46.                         "preSql": [],   #不是必须写的，作用是再写入数据前可以执行该sql
    47.                         "session": [], 
    48.                         "username": "root", 
    49.                         "writeMode": "insert"   #必选，指定数据写入的模式，分成三种：insert（一般默认）、replace、update
    50.                     }
    51.                 }
    52.             }
    53.         ], 
    54.         "setting": {
    55.             "speed": {
    56.                 "channel": 1
    57.             }
    58.         }
    59.     }
    60. }
    61.  
    62.  
    63.  
    64.  
    65. #执行脚本：
    66. datax.py   mysqlreaderTomysqlwriter.json

     将数据写入到mysql时，写入的表是需要提前创建的。

            3、mysql to hdfs

    参数解释：

    1. 使用datax的时候hdfswriter只支持两种文件形式，分别是text和orc
    2.  
    3. "fileType": "text",  #支持两种方式：text和orc，text表示的是textfile，orc表示的orcfile
    4.  
    5. "compress": "", #指定文件的压缩形式，不指定代表不用压缩，
    6.                 text支持的压缩方式：gzip，bzip2，
    7.                 orc支持的压缩方式有NONE和SNAPPY
    8.  
    9.  "writeMode": "append"   #表示的是数据在写入的操作，分成三种：
    10.                             append，写入前不做任何处理  
    11.                             nonconflit 如果文件存在，直接报错  
    12.                             truncate：如果文件存在，那就先删除在写入
    13.  
    14. "path": "/bigdata25/datax/datax_mysqltohdfs/"  #文件不存在是需要提前创建的
    15.  
    1. vim mysqlTohdfs.json
    2.  
    3. {
    4.     "job": {
    5.         "content": [
    6.             {
    7.                 "reader": {
    8.                     "name": "mysqlreader", 
    9.                     "parameter": {
    10.                         "column": ["id","name","age","clazz","gender"], 
    11.                         "connection": [
    12.                             {
    13.                                 "jdbcUrl": ["jdbc:mysql://master:3306/bigdata25"], 
    14.                                 "table": ["stu"]
    15.                             }
    16.                         ], 
    17.                         "password": "123456", 
    18.                         "username": "root", 
    19.                         "where": ""
    20.                     }
    21.                 }, 
    22.                 "writer": {
    23.                     "name": "hdfswriter", 
    24.                     "parameter": {
    25.                         "column": [
    26.                             {
    27.                                 "name":"id",
    28.                                 "type":"string"
    29.                             },
    30.                             {
    31.                                 "name":"stu_name",
    32.                                 "type":"string" 
    33.                             },
    34.                             {
    35.                                 "name":"age",
    36.                                 "type":"string"
    37.                             },
    38.                             {
    39.                                 
    40.                                 "name":"clazz",
    41.                                 "type":"string"
    42.                             },
    43.                             {
    44.                                 "name":"gender",
    45.                                 "type":"string" 
    46.                             }
    47.                         ], 
    48.                         "compress": "", 
    49.                         "defaultFS": "hdfs://master:9000", 
    50.                         "fieldDelimiter": ",", 
    51.                         "fileName": "stu_mysqltohdfs", 
    52.                         "fileType": "text",  
    53.                         "path": "/bigdata25/datax/datax_mysqltohdfs/", 
    54.                         "writeMode": "append"                         												
    55.                     }
    56.                 }
    57.             }
    58.         ], 
    59.         "setting": {
    60.             "speed": {
    61.                 "channel": 1
    62.             }
    63.         }
    64.     }
    65. }
    66.  
    67. #执行脚本：
    68. datax.py mysqlTohdfs.json
            4、mysql to hive

     原理思想：

    1. 实际上还是将数据存入到hdfs上面，hive通过记录元数据信息来获取数据
    2. 原理：
    3. 	创建好hive表在保存在hdfs上，是有文件路径，然后通过写入到指定的hdfs文件路径就能将数据写入到hive表中
    4. 当开启hive的时候，在hive中创建的表会默认的存储hdfs的/user/hive/warehouse/目录下

    前期准备：

    1. 前期准备：
    2. 启动hive（后台启动）：nohup hive --service metastore &
    3. 连接hive：hive
    4.  
    5. 创建hive表（在没有说明的情况下一般在都是创建一个外部表）
    6. 创建一个datax数据库：
    7. 	create database datax；
    8. 切换数据库：use datax
    9. 创建外部表：
    10. 	create external table if not exists datax_mysqltohive(
    11.     	 id string,
    12.          name string,
    13.          age int,
    14.          clazz string,
    15. 		gender string
    16.     		)
    17. row format delimited  fields terminated  by ',' stored as textfile

    编写脚本：

    1. vim mysqlTohive.json
    2.  
    3. {
    4.     "job": {
    5.         "content": [
    6.             {
    7.                 "reader": {
    8.                     "name": "mysqlreader", 
    9.                     "parameter": {
    10.                         "column": ["id","name","age","clazz","gender"], 
    11.                         "connection": [
    12.                             {
    13.                                 "jdbcUrl": ["jdbc:mysql://master:3306/bigdata25"], 
    14.                                 "table": ["stu"]
    15.                             }
    16.                         ], 
    17.                         "password": "123456", 
    18.                         "username": "root", 
    19.                         "where": ""
    20.                     }
    21.                 }, 
    22.                 "writer": {
    23.                     "name": "hdfswriter", 
    24.                     "parameter": {
    25.                         "column": [
    26.                             {
    27.                                 "name":"id",
    28.                                 "type":"string"
    29.                             },
    30.                             {
    31.                                 "name":"stu_name",
    32.                                 "type":"string" 
    33.                             },
    34.                             {
    35.                                 "name":"age",
    36.                                 "type":"string"
    37.                             },
    38.                             {
    39.                                 
    40.                                 "name":"clazz",
    41.                                 "type":"string"
    42.                             },
    43.                             {
    44.                                 "name":"gender",
    45.                                 "type":"string" 
    46.                             }
    47.                         ], 
    48.                         "compress": "", 
    49.                         "defaultFS": "hdfs://master:9000", 
    50.                         "fieldDelimiter": ",", 
    51.                         "fileName": "datax_mysqltohive1", 
    52.                         "fileType": "text", 
    53.                         "path": "/user/hive/warehouse/datax.db/datax_mysqltohive", 
    54.                         "writeMode": "append"
    55.                     }
    56.                 }
    57.             }
    58.         ], 
    59.         "setting": {
    60.             "speed": {
    61.                 "channel": 1
    62.             }
    63.         }
    64.     }
    65. }
    66.  
    67.  
    68. #执行脚本命令：
    69. datax.py  mysqlTohive.json
            5、mysql to hbase(Hbase11XWriter)

    参数解释：

    1. "mode": "normal" 写hbase的模式，目前只支持normal模式
    2.  
    3. "hbaseConfig"  {"hbase.zookeeper.quorum": "***"} 描述：连接HBase集群需要的配置信息，JSON格式
    4.  
    5. "table": "writer" 表的名称，大小写比较敏感
    6.  
    7. "encoding" 编码方式
    8.  
    9. "rowkeyColumn" 描述：要写入的hbase的rowkey列。index：指定该列对应reader端column的索引，从0开始，若为常量index为－1；type：指定写入数据类型，用于转换HBase byte[]；value：配置常量，常作为多个字段的拼接符。hbasewriter会将rowkeyColumn中所有列按照配置顺序进行拼接作为写入hbase的rowkey，不能全为常量
    10.  
    11.  
    12. "versionColumn" 表示指定写入hbase的时间戳，支持当前时间、指定时间列、指定时间

    前期准备：

    1. 前期准备：
    2.  
    3. 启动zookeeper：
    4. 	zkServer.sh start(每一个节点上都是需要启动的)
    5. 查看zk的状态：
    6. 	zkServer.sh status 
    7.  
    8. 启动hbase：
    9. 	start-hbase.sh
    10.  
    11. 连接hbase：
    12. 	sqlline.py master,node1,node2
    13.  
    14. 进入hbase的客户端：hbase shell 
    15.  
    16. hbase中查看表的命令：！table
    17. 退出命令 ！quit
    18. 查看表：list
    19. 在hbase中创建创建表,指定表名和列簇：create 'student','cf1'

    编写脚本：

    1. vim mysqlTohbase.json
    2.  
    3.  
    4.  
    5. {
    6.     "job": {
    7.         "content": [
    8.             {
    9.                 "reader": {
    10.                     "name": "mysqlreader", 
    11.                     "parameter": {
    12.                         "column": ["id","name","age","clazz","gender"], 
    13.                         "connection": [
    14.                             {
    15.                                 "jdbcUrl": ["jdbc:mysql://master:3306/bigdata25"], 
    16.                                 "table": ["stu"]
    17.                             }
    18.                         ], 
    19.                         "password": "123456", 
    20.                         "username": "root", 
    21.                         "where": ""
    22.                     }
    23.                 }, 
    24.                  "writer": {
    25.                       "name": "hbase11xwriter",
    26.                       "parameter": {
    27.                         "hbaseConfig": {
    28.                           "hbase.zookeeper.quorum": "master:2181,node1:2181,node2:2181"
    29.                         },
    30.                         "table": "NEW_STU",
    31.                         "mode": "normal",
    32.                         "rowkeyColumn": [
    33.                             {
    34.                               "index":0,
    35.                               "type":"string"
    36.                             },
    37.                             {
    38.                               "index":-1,
    39.                               "type":"string",
    40.                               "value":"_"
    41.                             }
    42.                         ],
    43.                         "column": [
    44.                           {
    45.                             "index":1,
    46.                             "name": "cf1:name",
    47.                             "type": "string"
    48.                           },
    49.                           {
    50.                             "index":2,
    51.                             "name": "cf1:age",
    52.                             "type": "int"
    53.                           },
    54.                           {
    55.                             "index":3,
    56.                             "name": "cf1:clazz",
    57.                             "type": "string"
    58.                           },
    59.                           {
    60.                             "index":4,
    61.                             "name": "cf1:gender",
    62.                             "type": "string"
    63.                           }
    64.                         ],
    65.                         "versionColumn":{
    66.                           "index": -1,
    67.                           "value":"123456789"
    68.                         },
    69.                         "encoding": "utf-8"
    70.                       
    71.                     }
    72.                 }
    73.             }
    74.         ],
    75.         "setting": {
    76.             "speed": {
    77.                 "channel": 1
    78.             }
    79.         }  
    80.     }
    81. }
    82.  
    83. #执行脚本
    84. datax.py mysqlTohbase.json
            6、mysql增量同步数据到hive中。

    最主要的工作就是在原先的mysql数据导入的hive中的基础上进行where过滤

    1. vim mysqlTohive.json
    2.  
    3. {
    4.     "job": {
    5.         "content": [
    6.             {
    7.                 "reader": {
    8.                     "name": "mysqlreader", 
    9.                     "parameter": {
    10.                         "column": ["id","name","age","clazz","gender"], 
    11.                         "connection": [
    12.                             {
    13.                                 "jdbcUrl": ["jdbc:mysql://master:3306/bigdata25"], 
    14.                                 "table": ["stu"]
    15.                             }
    16.                         ], 
    17.                         "password": "123456", 
    18.                         "username": "root", 
    19.                         "where": "id >20"
    20.                     }
    21.                 }, 
    22.                 "writer": {
    23.                     "name": "hdfswriter", 
    24.                     "parameter": {
    25.                         "column": [
    26.                             {
    27.                                 "name":"id",
    28.                                 "type":"string"
    29.                             },
    30.                             {
    31.                                 "name":"stu_name",
    32.                                 "type":"string" 
    33.                             },
    34.                             {
    35.                                 "name":"age",
    36.                                 "type":"string"
    37.                             },
    38.                             {
    39.                                 
    40.                                 "name":"clazz",
    41.                                 "type":"string"
    42.                             },
    43.                             {
    44.                                 "name":"gender",
    45.                                 "type":"string" 
    46.                             }
    47.                         ], 
    48.                         "compress": "", 
    49.                         "defaultFS": "hdfs://master:9000", 
    50.                         "fieldDelimiter": ",", 
    51.                         "fileName": "datax_mysqltohive1", 
    52.                         "fileType": "text", 
    53.                         "path": "/user/hive/warehouse/datax.db/datax_mysqltohive", 
    54.                         "writeMode": "append"
    55.                     }
    56.                 }
    57.             }
    58.         ], 
    59.         "setting": {
    60.             "speed": {
    61.                 "channel": 1
    62.             }
    63.         }
    64.     }
    65. }
    66.  
    67.  
    68. #执行脚本命令：
    69. datax.py  mysqlTohive.json
    9、在使用datax过程中出现的错误：

    1、配置文件出现错误，脚本不完整：

  • 相关阅读:
    基于Nodejs的医生预约平台的设计和实现
    JSP内置对象及作用域（三）
    给你安利一款带有AI功能的数据库管理工具
    【Harmony OS】【JAVA UI】鸿蒙应用如何集成OKHttp网络三方库
    java计算机毕业设计基于ssm的少儿编程管理系统(源代码+数据库+Lw文档）
    CommonModule.dll动态链接库（DLL）文件丢失的处理方法
    The Grapes NFT 概览与数据分析
    LeetCode1247
    TI DSP的中断
    企业上云时代，适配进销存管理系统从四方面选
  • 原文地址：https://blog.csdn.net/m0_62078954/article/details/136646970