乐趣区

关于大数据:DataX-Web数据增量同步配置说明

jiezi

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

1. 页面工作配置

关上菜单工作治理页面，抉择增加工作

按下图中 5 个步骤进行配置

  • 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 配置

datax.json

{
  "job": {
    "setting": {
      "speed": {"channel": 16}
    },
    "content": [
      {
        "reader": {
          "name": "mysqlreader",
          "parameter": {
            "splitPk": "id",
            "username": "root",
            "password": "root",
            "column": ["*"],
            "connection": [
              {
                "jdbcUrl": ["jdbc:mysql://localhost:3306/test?characterEncoding=utf8"],
"querySql": ["select * from test_list where operationDate >= FROM_UNIXTIME(${lastTime}) and operationDate < FROM_UNIXTIME(${currentTime})"
                                ]
              }
            ]
          }
        },
        "writer": {
          "name": "mysqlwriter",
          "parameter": {
            "username": "root",
            "password": "123456",
            "column": ["*"],
            "batchSize": "4096",
            "connection": [
              {
                "jdbcUrl": "jdbc:mysql://localhost:3307/test?characterEncoding=utf8",
                "table": ["test_list"]
              }
            ]
          }
        }
      }
    ]
  }
}

querySql 解析 

select * from test_list where operationDate >= ${lastTime} and operationDate < ${currentTime}
  • 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. 页面工作配置

关上菜单工作治理页面，抉择增加工作

按下图中 4 个步骤进行配置

  • 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 配置

datax.json

{
   "job": {
     "setting": {
       "speed": {
         "channel": 3,
         "byte": 1048576
       },
       "errorLimit": {
         "record": 0,
         "percentage": 0.02
       }
     },
     "content": [
       {
         "reader": {
           "name": "mysqlreader",
           "parameter": {
             "username": "yRjwDFuoPKlqya9h9H2Amg==",
             "password": "yRjwDFuoPKlqya9h9H2Amg==",
             "splitPk": "","connection": [
               {
                 "querySql": ["select * from job_log where id>= ${startId} and id< ${endId}"
                 ],
                 "jdbcUrl": ["jdbc:mysql://localhost:3306/datax_web"]
               }
             ]
           }
         },
         "writer": {
           "name": "mysqlwriter",
           "parameter": {
             "username": "mCFD+p1IMsa0rHicbQohcA==",
             "password": "PhYxJmA/nuBJD1OxKTRzZH8sxuRddOv83hdqDOVR+i0=",
             "column": [
               "`id`",
               "`job_group`",
               "`job_id`",
               "`job_desc`",
               "`executor_address`",
               "`executor_handler`",
               "`executor_param`",
               "`executor_sharding_param`",
               "`executor_fail_retry_count`",
               "`trigger_time`",
               "`trigger_code`",
               "`trigger_msg`",
               "`handle_time`",
               "`handle_code`",
               "`handle_msg`",
               "`alarm_status`",
               "`process_id`",
               "`max_id`"
             ],
             "connection": [
               {
                 "table": ["job_log"],
                 "jdbcUrl": "jdbc:mysql://47.98.125.243:3306/datax_web"
               }
             ]
           }
         }
       }
     ]
   }
 }

querySql 解析 

select * from job_log where id>= ${startId} and id< ${endId}
  • 1. 此处的关键点在 ${startId}，${endId}，${} 是 DataX 动静参数的固定格局，startId，endId 就是咱们页面配置中

-DstartId=’%s’ -DendId=’%s’ 中的 startId，endId，留神字段肯定要统一。

三、JVM 启动参数配置

此抉择为非必选，能够配置 DataX 启动时 JVM 的参数，具体配置不做详解。

  • JVM 启动参数拼接后果为：-j “-Xms2G -Xmx2G”

四、常见问题

请查看 issue 列表或者提 issue 阐明问题，咱们会尽快回复。

https://github.com/WeiYe-Jing…

退出移动版