datax关于postsql数据增量迁移的问题

news2025/1/12 4:07:42

看官方文档是不支持的

数据源及同步方案_大数据开发治理平台 DataWorks(DataWorks)-阿里云帮助中心 (aliyun.com)

看了下源码有个postsqlwriter 看了下也就拼接sql

将

PostgresqlWriter中的不允许更新先注释了 让他过去先

然后看到 WriterUtil中的对应方法

getWriteTemplate

确实只有对应mysql的方法

在下面加个postgres的

else if(forceUseUpdate ||
                ((dataBaseType == DataBaseType.PostgreSQL ) && writeMode.trim().toLowerCase().startsWith("update"))) {
            writeDataSqlTemplate = new StringBuilder()
                    .append("INSERT INTO %s (").append(StringUtils.join(columnHolders, ","))
                    .append(") VALUES(").append(StringUtils.join(valueHolders, ","))
                    .append(")")
                    .append(onDuplicateKeyUpdateStringPostgres(columnHolders,writeMode.replace("update#","")))
                    .toString();
}

其中调用方法

public static String onDuplicateKeyUpdateStringPostgres(List<String> columnHolders,String pk){
        if (columnHolders == null || columnHolders.size() < 1) {
            return "";
        }
        StringBuilder sb = new StringBuilder();
        sb.append(" ON CONFLICT ("+pk+") DO UPDATE SET ");
        boolean first = true;
        for(String column:columnHolders){
            if(column.equalsIgnoreCase(pk)){
                continue;
            }
            if(!first){
                sb.append(",");
            }else{
                first = false;
            }
            sb.append(column);
            sb.append("=EXCLUDED.");
            sb.append(column);
        }

        return sb.toString();
    }

然后配置文件中带上id