datax id增量导入数据

系统管理员 2023-09-29 14:29 7阅读 0赞

## datax 通过id增量导入数据 ##

#### 1.编写要导出mysql原表最大id到文件的datax脚本 ####

{
        
        "job": {
        
            "content": [
                {
        
                    "reader": {
        
                        "name": "mysqlreader", 
                        "parameter": {
        
                            "connection": [ 
                                {
        
                                    "jdbcUrl": ["jdbc:mysql://域名端口/数据库名?characterEncoding=utf8"],
                                    "querySql": [
                               "select max(id) from 表名;"
                            ]
                                }
                            ],
                            "password": "密码",
                            "username": "账号"
                        }
                    },
                    "writer": {
        
                        "name": "txtfilewriter",
                        "parameter": 
                   {
        
                      "fileName": "文件名",
                      "fileFormat": "csv",
                      "path": "文件地址",
                      "writeMode": "truncate（先删除原文件）"
                        }
                    }
                }
            ],
            "setting": {
        
                "speed": {
        
                    "channel": "3"
                }
            }
        }
    }

#### 2.编写要导入postgresql表的最大id到文件的datax脚本 ####

{
        
        "job": {
        
            "content": [
                {
        
                    "reader": {
        
                        "name": "postgresqlreader", 
                        "parameter": {
        
                            "connection": [ 
                                {
        
                                    "jdbcUrl": ["jdbc:postgresql://域名端口/数据库名"],
                                    "querySql": [
                               "select coalesce(max(id),0) from 表名;"
                            ]
                                }
                            ],
                            "password": "密码",
                            "username": "账号"
                        }
                    },
                    "writer": {
        
                        "name": "txtfilewriter",
                        "parameter": 
                   {
        
                      "fileName": "文件名",
                      "fileFormat": "csv",
                      "path": "文件地址",
                      "writeMode": "truncate（先删除原文件）"
                        }
                    }
                }
            ],
            "setting": {
        
                "speed": {
        
                    "channel": "3"
                }
            }
        }
    }

#### 3.编写将mysql原表数据增量导入postgresql的datax脚本 ####

{
        
        "job": {
        
            "content": [
                {
        
                    "reader": {
        
                        "name": "mysqlreader", 
                        "parameter": {
        
                            "connection": [ 
                                {
        
                                    "jdbcUrl": ["jdbc:mysql://域名端口/数据库名?characterEncoding=utf8"],
                                    "querySql": [
                               "select * from cw_bxdf.uf_project where 1=1;"
                            ]
                                }
                            ],
                            "password": "密码",
                            "username": "账号"
                        }
                    },
                    "writer": {
        
                        "name": "postgresqlwriter",
                        "parameter": {
        
                            "column":["*"],
                            "connection": [
                                {
        
                                    "jdbcUrl": "jdbc:postgresql://域名端口/数据库名",
                                    "table":["表名"]
                                }
                            ],
                            "password": "密码",
                            "postSql": [],
                            "preSql": [],
                            "username": "账号"
                        }
                    }
                }
            ],
            "setting": {
        
                "speed": {
        
                    "channel": "3"
                }
            }
        }
    }

#### 4.用shell脚本将这些datax脚本全部串连起来运行，最终实现从mysql增量导入postgresql中 ####

#! /bin/bash
    #获取原数据库最大id值，并写入一个 csv 文件
    python /root/datax/bin/datax.py /文件目录/uf_project_max_id.json
    # $?是shell变量,表示"最后一次执行命令"的退出状态.0为成功,非0为失败, -ne 为不等于
    if [ $? -ne 0 ]; then
       echo "data.sh error, can not get yuanduan_time from target db!"
    exit 1
    fi
    
    #获取目标数据库最大id值，并写入一个 csv 文件
    python /root/datax/bin/datax.py /文件目录/ods_project_pr_max_id.json
    # $?是shell变量,表示"最后一次执行命令"的退出状态.0为成功,非0为失败, -ne 为不等于
    if [ $? -ne 0 ]; then
      echo "data.sh error, can not get mubiao_time from target db!"
      exit 1
    fi
    
    
    # 找到 DataX 写入的文本文件，并将内容读取到一个变量中
    uf_project_max_id=`cat /文件目录/uf_project__max_id_result__*`
    ods_project_pr_max_id=`cat /文件目录/ods_project_pr_max_id_result__*`
    
    # 判断原数据库id是是否大于目标数据库id，大于就运行下面脚本
    if [ $uf_project_max_id -ge $ods_project_pr_max_id ]; then
    # 将sql where条件拼接
     WHERE="id > '$ods_project_pr_max_id'"
     # 将sql where条件替换成拼接的where条件，并写入文件覆盖到复制出的tmp，datax脚本
     sed "s/1=1/$WHERE/g" /文件目录/ods_project_pr.json > /文件目录/ods_project_pr_tmp.json
     # 最后运行覆盖后的tmp脚本，完成从mysql增量导入postgresql
     python /root/datax/bin/datax.py /文件目录/ods_project_pr_tmp.json
    fi