1.2 DataX 数据同步工具详细教程

DataX 是阿里巴巴开源的一款高效的数据同步工具，旨在实现多种异构数据源之间的高效数据同步。以下是对 DataX 的详细介绍：

架构

DataX 的架构主要包括以下几个核心组件：

DataX Core：负责任务调度、插件加载、日志管理等核心功能。
Reader Plugin：用于从数据源读取数据，不同的数据源对应不同的 Reader 插件。
Writer Plugin：用于将数据写入目标数据源，不同的数据源对应不同的 Writer 插件。
Transformer Plugin：用于在数据传输过程中进行数据转换。

DataX 的架构图如下：

+-------------------------------------------------+
|                     DataX                       |
|  +---------+    +--------------+    +---------+ |
|  |  Reader | -> | DataX Core   | -> |  Writer | |
|  |  Plugin |    | (Engine,     |    |  Plugin | |
|  |         |    |  Scheduler,  |    |         | |
|  |         |    |  Transformer |    |         | |
|  |         |    |  Plugin)     |    |         | |
|  +---------+    +--------------+    +---------+ |
+-------------------------------------------------+

基本工作流程

在这里插入图片描述

DataX 的工作流程可以分为以下几个步骤：

配置任务：用户通过 JSON 文件配置数据同步任务，包括数据源、目标数据源、数据字段映射等。
任务调度：DataX Core 解析配置文件，加载相应的 Reader 和 Writer 插件，并开始任务调度。
数据读取：Reader 插件从数据源读取数据，并将数据传递给 DataX Core。
数据转换：如有需要，Transformer 插件对数据进行转换。
数据写入：Writer 插件将转换后的数据写入目标数据源。
任务结束：数据同步任务完成，DataX 生成任务报告，记录任务执行的详细信息。

使用场景

DataX 可以应用于以下几种常见的数据同步场景：

数据库间数据迁移：如 MySQL 到 Oracle，PostgreSQL 到 MySQL。
大数据平台数据同步：如 HDFS 到 Hive，Hive 到 HBase。
云服务数据迁移：如 RDS 到 OSS，OSS 到 S3。

优越点

DataX 作为一款数据同步工具，具备以下优越点：

高效稳定：DataX 采用多线程并发处理机制，能够高效地完成大规模数据同步任务。
易于扩展：通过插件机制，DataX 可以轻松支持多种数据源的读写操作。
配置灵活：使用 JSON 格式的配置文件，用户可以方便地定义数据同步任务。
支持多种数据源：内置了丰富的 Reader 和 Writer 插件，支持常见的数据库、大数据平台和云服务。
良好的监控和报警机制：DataX 提供详细的任务日志和监控功能，便于用户监控和诊断数据同步任务。
开源免费：DataX 是开源项目，用户可以免费使用，并根据需要进行二次开发。

下面，让我们通过一个具体的案例来了解 DataX 的运行流程：使用 DataX 同步 MySQL 数据到 Hive。

案例：同步 MySQL 数据到 Hive

1. 案例背景

假设我们有一个 MySQL 数据库，其中有一个表 employees，包含员工信息，我们希望将这个表的数据同步到 Hive 中进行数据分析。

2. 环境准备

确保已经安装了 Java 环境，因为 DataX 是基于 Java 开发的。
下载并解压 DataX 工具包到本地目录。
确保 MySQL 和 Hive 服务都是可访问的。

3. 编写 DataX 作业配置文件

创建一个名为 mysql2hive.json 的配置文件，内容如下：

{
    "job": {
        "setting": {
            "speed": {
                "channel": 1
            }
        },
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "username": "your_mysql_username",
                        "password": "your_mysql_password",
                        "connection": [
                            {
                                "jdbcUrl": "jdbc:mysql://your_mysql_host:3306/your_database",
                                "table": [
                                    "employees"
                                ]
                            }
                        ],
                        "column": [
                            "id",
                            "name",
                            "age",
                            "department"
                        ]
                    }
                },
                "writer": {
                    "name": "hivewriter",
                    "parameter": {
                        "username": "your_hive_username",
                        "password": "your_hive_password",
                        "connection": [
                            {
                                "jdbcUrl": "jdbc:hive2://your_hive_host:10000/default",
                                "table": [
                                    "employees"
                                ]
                            }
                        ],
                        "writeMode": "insert",
                        "hadoopConfig": {
                            "fs.defaultFS": "hdfs://your_hadoop_host:9000"
                        },
                        "column": [
                            "id",
                            "name",
                            "age",
                            "department"
                        ]
                    }
                }
            }
        ]
    }
}

代码解释：

speed：设置同步速度，channel 表示并发数量。
reader：配置 MySQL 读取器，包括数据库连接信息和要同步的表及列。
writer：配置 Hive 写入器，包括 Hive 连接信息和目标表及列。writeMode 为 insert 表示插入模式。

4. 运行 DataX 作业

在命令行中，进入到 DataX 解压目录的 bin 目录下，执行以下命令来运行 DataX 作业：

python datax.py ../json/mysql2hive.json

5. 监控 DataX 作业

运行 DataX 作业后，你将看到实时的任务执行情况，包括已读取的记录数、速度、错误记录等。DataX 也会生成日志文件，你可以在 log 目录下查看。

6. 验证数据同步结果

同步完成后，你可以在 Hive 中查询 employees 表，验证数据是否已经成功同步。

7. 注意事项

确保配置文件中的数据库连接信息、用户名、密码、表名和列名都是正确的。
Hive 写入器需要 Hadoop 环境配置正确，包括 Hadoop 配置文件和 HDFS 地址。
根据实际环境和需求调整并发数（channel）和其他参数。

通过这个案例，你可以看到 DataX 的强大功能和灵活性，它可以轻松地在不同的数据源之间同步数据。

好的，下面是一个使用 DataX 将 Hive 数据同步到 MySQL 的实际案例。这个案例包括数据同步任务的配置文件和相关步骤。

案例：同步 Hive 数据到 MySQL

环境准备

安装 DataX：从 DataX GitHub 仓库下载并安装 DataX。
配置 Hive 和 MySQL 连接：确保 Hive 和 MySQL 可以通过网络互相访问，并准备好所需的 JDBC 驱动。

配置文件

首先，创建一个 DataX 配置文件 hive_to_mysql.json，定义从 Hive 到 MySQL 的数据同步任务。

{
  "job": {
    "setting": {
      "speed": {
        "channel": 3  // 并发线程数
      }
    },
    "content": [
      {
        "reader": {
          "name": "hdfsreader",
          "parameter": {
            "path": "hdfs://namenode:8020/user/hive/warehouse/your_table", // Hive 表所在的 HDFS 路径
            "defaultFS": "hdfs://namenode:8020",
            "fileType": "orc",  // 文件类型
            "column": [
              {"index": 0, "type": "long"},
              {"index": 1, "type": "string"},
              {"index": 2, "type": "double"}
              // 依次配置所有列
            ],
            "fieldDelimiter": "\u0001",  // 字段分隔符，Hive 默认使用 ^A
            "nullFormat": "\\N"
          }
        },
        "writer": {
          "name": "mysqlwriter",
          "parameter": {
            "username": "your_mysql_username",
            "password": "your_mysql_password",
            "column": [
              "column1",
              "column2",
              "column3"
              // 依次配置所有列
            ],
            "preSql": [
              "DELETE FROM your_mysql_table"  // 在数据写入前执行的 SQL 语句
            ],
            "connection": [
              {
                "table": [
                  "your_mysql_table"
                ],
                "jdbcUrl": "jdbc:mysql://your_mysql_host:3306/your_database"
              }
            ],
            "writeMode": "insert"  // 写入模式
          }
        }
      }
    ]
  }
}

步骤详解

定义 Reader 配置：
- path：Hive 表在 HDFS 上的路径。
- defaultFS：HDFS 的默认文件系统地址。
- fileType：文件类型（如 ORC、Parquet）。
- column：Hive 表的列定义，包括列索引和数据类型。
- fieldDelimiter：字段分隔符，Hive 默认使用 ^A。
- nullFormat：表示空值的格式。
定义 Writer 配置：
- username 和 password：MySQL 数据库的用户名和密码。
- column：对应 MySQL 表的列名。
- preSql：在数据写入之前执行的 SQL 语句，如清空表数据。
- connection：MySQL 数据库连接信息，包括目标表名和 JDBC URL。
- writeMode：写入模式（如插入或更新）。

执行同步任务

启动 DataX：
在 DataX 的安装目录下，运行以下命令来执行数据同步任务：
```
python ${DATAX_HOME}/bin/datax.py /path/to/hive_to_mysql.json
```
其中，${DATAX_HOME} 是 DataX 的安装目录，/path/to/hive_to_mysql.json 是前面创建的配置文件的路径。

优化和调试

日志查看：
DataX 在执行过程中会生成详细的日志，便于查看同步任务的执行情况和调试错误。
并发优化：
根据数据量和服务器性能，调整 channel 数量以优化同步速度。
错误处理：
如果任务执行失败，根据日志信息检查配置文件，确保 Hive 和 MySQL 的连接信息正确无误。

通过上述步骤，我们可以使用 DataX 高效地将 Hive 数据同步到 MySQL。DataX 的灵活配置和高并发处理能力使其能够应对大规模数据同步任务，同时提供了详细的日志和监控功能，便于管理和调试。

dataX job 性能优化

对 DataX job 进行性能优化可以从以下几个方面入手：

并发配置优化 ：
- 合理配置读写并发数，根据数据源性能和网络带宽逐步调整并发数，以确定最佳并发数量。
- 配置全局 Byte 限速和单 Channel Byte 限速，通过设置 job.setting.speed.byte 和 core.transport.channel.speed.byte 来控制 DataX job 内 Channel 并发。
批量提交大小优化 ：
- 调整批量提交大小 batchSize，减少 DataX 与数据库的网络交互次数，提升数据同步效率。
调整 JVM 堆内存 ：
- 为了防止 OOM 错误，增加 JVM 的堆内存，建议设置为 4G 或 8G。
数据库连接池使用 ：
- 使用数据库连接池提高数据读取和写入的效率。
SQL 语句优化 ：
- 优化 SQL 语句，创建索引和分区表，减少查询时间。
合理使用 splitPk ：
- 使用 splitPk 进行任务切分，提高任务并行度，尤其适用于大规模数据同步。
调整 Reader 和 Writer 参数 ：
- 根据 Reader 和 Writer 的类型调整参数，例如 fetchSize 对于 OracleReader 可以提升性能。
网络优化 ：
- 考虑网络带宽对 DataX 传输速度的影响，优化网络设置或使用内网地址提高数据传输效率。
日志级别调整 ：
- 调整日志级别，例如将 trace 改为 enable，减少日志输出，提高性能。
资源分配：
- 确保 DataX 作业运行在具有足够 CPU 和内存资源的机器上。
监控和分析：
- 使用 DataX 提供的监控工具分析作业执行情况，根据实际情况调整配置。

通过上述优化措施，可以有效提高 DataX job 的性能和数据同步效率。在实际操作中，可能需要根据具体的数据源和网络环境进行综合考虑和调整。

DataX 的优化参数主要在 DataX 作业的 JSON 配置文件中设置。以下是一些关键的优化参数及其在 JSON 配置文件中的位置：

并发数（Channel 个数）：

在 "job" -> "setting" -> "speed" 下设置 "channel" 参数。

{
  "job": {
    "setting": {
      "speed": {
       { "channel": 5} }
    }
  }
 }

批量提交大小（Batch Size）：
- 在对应的 Writer 插件的 "parameter" 下设置 "batchSize" 参数。
```
{
  "writer": {
    "parameter": {
```

{ “batchSize”: 2000}
}
}
}


3. **JVM 堆内存**：
- JVM 堆内存通常在启动 DataX 作业的命令行中设置，例如使用 `-Xms8G -Xmx8G` 参数。
```shell
python datax.py --jvm="-Xms8G -Xmx8G" your_datax_job.json

数据库连接池：
- 某些数据库插件可能支持连接池，具体参数根据插件文档设置，在 Reader 或 Writer 的 "parameter" 下配置。
SQL 语句优化：
- 在 Reader 插件的 "parameter" 下的 "querySql" 或 "table" 属性中优化 SQL 语句。

SplitPk：

在 Reader 插件的 "parameter" 下设置 "splitPk" 参数，用于数据分片。

 {
  "reader": {
    "parameter": {
                 { "splitPk": "id"}
     }
  }
 }

Reader 和 Writer 特定参数：
- 根据使用的 Reader 或 Writer 类型，在 "parameter" 下设置特定参数，如 "fetchSize" 等。
日志级别：
- 日志级别通常在 DataX 配置文件 conf/core.json 中设置，例如 "logLevel": "debug"。
资源分配：
- 资源分配主要取决于运行 DataX 作业的服务器配置，确保服务器有足够的 CPU 和内存资源。
监控和分析：
- 监控和分析通常通过 DataX 的日志输出和监控工具进行，不需要在 JSON 配置文件中设置。