datax_web数据增量同步配置

2024-02-25 08:28:58

一、根据日期进行增量数据抽取

1.页面任务配置

打开菜单任务管理页面，选择添加任务

按下图中步骤进行配置

1.任务类型选DataX任务
2.辅助参数选择时间自增
3.增量开始时间选择，即sql中查询时间的开始时间，用户使用此选项方便第一次的全量同步。第一次同步完成后，该时间被更新为上一次的任务触发时间，任务失败不更新。
4.增量时间字段,-DlastTime='%s' -DcurrentTime='%s' 先来解析下这段字符串

1.-D是DataX参数的标识符，必配
2.-D后面的lastTime和currentTime是DataX json中where条件的时间字段标识符，必须和json中的变量名称保持一致

3.='%s'是项目用来去替换时间的占位符，比配并且格式要完全一致

4.注意-DlastTime='%s'和-DcurrentTime='%s'中间有一个空格，空格必须保留并且是一个空格

5.时间格式，可以选择自己数据库中时间的格式，也可以通过json中配置sql时间转换函数来处理

2.JSON配置

{
  "job": {
    "setting": {
      "speed": {
        "channel": 3,
        "byte": -1
      },
      "errorLimit": {
        "record": 0,
        "percentage": 0.02
      }
    },
    "content": [
      {
        "reader": {
          "name": "mysqlreader",
          "parameter": {
            "username": "7aAw6fAFXgqP2weyjjwIAw==",
            "password": "1Sh8F0VGrkzgnRXsNXowUAxSS1xnCyE8TrEgzQ7ZE40=",
            "column": [
              "ID",
              "CREATE_TIME",
              "USER_ID",
              "UPDATE_TIME",
              "LAST_MODIFY_USER_ID"
            ],
            "where": " CREATE_TIME >= FROM_UNIXTIME(${lastTime}) and CREATE_TIME < FROM_UNIXTIME(${currentTime})",
            "splitPk": "ID",
            "connection": [
              {
                "table": [
                  "t_test"
                ],
                "jdbcUrl": [
                  "jdbc:mysql://127.0.0.1:3306/test"
                ]
              }
            ]
          }
        },
        "writer": {
          "name": "clickhousewriter",
          "parameter": {
            "username": "OhlJ4g2KfCRznayQNh0eng==",
            "password": "ONwWYPUDMPXDIREymhWAMQ==",
            "column": [
              "ID",
              "CREATE_TIME",
              "USER_ID",
              "UPDATE_TIME",
              "LAST_MODIFY_USER_ID"
            ],
            "connection": [
              {
                "table": [
                  "tb"
                ],
                "jdbcUrl": "jdbc:clickhouse://localhost:8123/local"
              }
            ]
          }
        }
      }
    ]
  }
}

1.此处的关键点在${lastTime}，${currentTime}，${}是DataX动态参数的固定格式，lastTime，currentTime就是我们页面配置中 -DlastTime='%s' -DcurrentTime='%s'中的lastTime，currentTime，注意字段一定要一致。
2.如果任务配置页面，时间类型选择为时间戳但是数据库时间格式不是时间戳，例如是：2019-11-26 11:40:57 此时可以用FROM_UNIXTIME(${lastTime})进行转换。
select * from test_list where operationDate >= FROM_UNIXTIME(${lastTime}) and operationDate < FROM_UNIXTIME(${currentTime})

二、根据自增Id进行增量数据抽取

1.页面任务配置

打开菜单任务管理页面，选择添加任务

按下图中步骤进行配置

1.任务类型选DataX任务
2.辅助参数选择主键自增
3.增量主键开始ID选择，即sql中查询ID的开始ID，用户使用此选项方便第一次的全量同步。第一次同步完成后，该ID被更新为上一次的任务触发时最大的ID，任务失败不更新。
4.增量时间字段,-DstartId='%s' -DendId='%s' 先来解析下这段字符串

1.-D是DataX参数的标识符，必配

2.-D后面的startId和endId是DataX json中where条件的id字段标识符，必须和json中的变量名称保持一致

3.='%s'是项目用来去替换时间的占位符，比配并且格式要完全一致

4.注意-DstartId='%s'和-DendId='%s' 中间有一个空格，空格必须保留并且是一个空格

5.reader数据源，选择任务同步的读数据源

6.配置reader数据源中需要同步数据的表名及该表的主键

2.JSON配置

{
  "job": {
    "setting": {
      "speed": {
        "channel": 3,
        "byte": -1
      },
      "errorLimit": {
        "record": 0,
        "percentage": 0.02
      }
    },
    "content": [
      {
        "reader": {
          "name": "mysqlreader",
          "parameter": {
            "username": "7aAw6fAFXgqP2weyjjwIAw==",
            "password": "1Sh8F0VGrkzgnRXsNXowUAxSS1xnCyE8TrEgzQ7ZE40=",
            "column": [
              "ID",
              "CREATE_TIME",
              "USER_ID",
              "UPDATE_TIME",
              "LAST_MODIFY_USER_ID"
            ],
            "where": " ID >= ${startId} and ID < ${endId} ",
            "splitPk": "ID",
            "connection": [
              {
                "table": [
                  "t_test"
                ],
                "jdbcUrl": [
                  "jdbc:mysql://127.0.0.1:3306/test"
                ]
              }
            ]
          }
        },
        "writer": {
          "name": "clickhousewriter",
          "parameter": {
            "username": "OhlJ4g2KfCRznayQNh0eng==",
            "password": "ONwWYPUDMPXDIREymhWAMQ==",
            "column": [
              "ID",
              "CREATE_TIME",
              "USER_ID",
              "UPDATE_TIME",
              "LAST_MODIFY_USER_ID"
            ],
            "connection": [
              {
                "table": [
                  "tb"
                ],
                "jdbcUrl": "jdbc:clickhouse://localhost:8123/local"
              }
            ]
          }
        }
      }
    ]
  }
}

此处的关键点在${startId}，${endId}，${}是DataX动态参数的固定格式，startId，endId就是我们页面配置中 -DstartId='%s' -DendId='%s'中的startId，endId，注意字段一定要一致。

码农公寓

相关文章