异构数据同步 datax (2)-postgres 写扩展

2024-08-21 00:20

本文主要是介绍异构数据同步 datax (2)-postgres 写扩展,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

1、postgres SQL 支持  插入更新操作(与mysql 语法有一定差异)

可参考下面文章

MySQL + PostgreSQL批量插入更新insertOrUpdate_mysql insert update-CSDN博客

2、datax中,可通过源码调整来实现

参考来源

https://juejin.cn/post/7124899170615296013

3、源码调整注意事项

datax : 版本 

源码下载,自行用idea进行打包编译,修改完如下类,

com.alibaba.datax.plugin.writer.postgresqlwriter.PostgresqlWriter

com.alibaba.datax.plugin.rdbms.writer.util.WriterUtil

编译替换jar文件名:

postgresqlwriter-0.0.1-SNAPSHOT.jar

plugin-rdbms-util-0.0.1-SNAPSHOT.jar

目录树如下:(plugin/writer/postgresqlwriter)

find <目录路径> | sed -e 's/[^-][^\/]*\//--/g' -e 's/--/|-/'

|-lib
|-bin
|-job
|-conf
|-log
|-log_perf
|-tmp
|-script
|-plugin
|---writer
|-----postgresqlwriter
|-------plugin_job_template.json
|-------plugin.json
|-------libs
|---------checker-qual-3.5.0.jar
|---------postgresql-42.3.3.jar
|---------commons-collections-3.0.jar
|---------druid-1.0.15.jar
|---------commons-lang3-3.3.2.jar
|---------logback-core-1.0.13.jar
|---------commons-io-2.4.jar
|---------datax-common-0.0.1-SNAPSHOT.jar
|---------guava-r05.jar
|---------plugin-rdbms-util-0.0.1-SNAPSHOT.jar
|---------hamcrest-core-1.3.jar
|---------logback-classic-1.0.13.jar
|---------commons-math3-3.1.1.jar
|---------slf4j-api-1.7.10.jar
|---------fastjson2-2.0.23.jar
|-------postgresqlwriter-0.0.1-SNAPSHOT.jar

4、使用

4.1 、可以支持带有唯一索引的表的新增或者更新

mysql 表结构

CREATE TABLE `sys_test_copy2` (`user_id` bigint NOT NULL DEFAULT '0' COMMENT '用户ID',`email` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT '' COMMENT '用户邮箱',`iso_country_code` varchar(3) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT 'ISO国家代码',`country` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '国家',`brand_no` varchar(30) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '品牌',`source` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '来源',`create_time` datetime DEFAULT NULL COMMENT '创建时间',PRIMARY KEY (`user_id`),UNIQUE KEY `sys_test_copy2_u1` (`email`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;

目标 PG表结构

CREATE TABLE "public"."sys_test_copy2" ("user_id" int8 NOT NULL,"email" varchar(50) COLLATE "pg_catalog"."default","iso_country_code" varchar(3) COLLATE "pg_catalog"."default","country" varchar(50) COLLATE "pg_catalog"."default","brand_no" varchar(30) COLLATE "pg_catalog"."default","source" varchar(50) COLLATE "pg_catalog"."default","create_time" timestamp(6),CONSTRAINT "sys_test_copy2_pkey" PRIMARY KEY ("user_id")
)
;ALTER TABLE "public"."sys_test_copy2" OWNER TO "postgres";CREATE UNIQUE INDEX "sys_test_copy2_u1" ON "public"."sys_test_copy2" USING btree ("email" COLLATE "pg_catalog"."default" "pg_catalog"."text_ops" ASC NULLS LAST
);

datax job: 

{"job":{"content":[{"reader":{"name":"mysqlreader","parameter":{"username": "root","password": "xxxxxx","connection": [{"jdbcUrl": ["jdbc:mysql://192.168.5.180:3306/xxxx?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&serverTimezone=GMT%2B8"],"querySql": [" SELECT * from  sys_test_copy2"]}}},"writer":{"name":"postgresqlwriter","parameter":{"writeMode": "update!@#(user_id)!@#(email)","column":["id","name"],"connection":[{"jdbcUrl":"jdbc:postgresql://127.0.0.1:5432/postgres","table":["sys_test_copy2"]}],"password":"xxxx","username":"postgres"}}}],"setting":{"speed":{"channel":6}}}
}

执行job,生成的模版语句:

INSERT INTO %s (user_id,email,iso_country_code,country,brand_no,source,create_time) VALUES(?::int8,?::varchar,?::varchar,?::varchar,?::varchar,?::varchar,?::timestamp) ON CONFLICT (user_id) DO UPDATE SET email=EXCLUDED.email

4.2、根据主键进行新增或者更新

INSERT INTO sys_test_copy1(user_id, email) VALUES (5592, 'xxxx5@hotmail.com')  ON CONFLICT (user_id) do nothing;

表结构就不放了,去掉唯一索引

datax job:

{"job": {"setting": {"speed": {"channel": 5},"errorLimit": {"record": 0,"percentage": 0.02}},"content": [{"reader": {"name": "mysqlreader","parameter": {"username": "root","password": "数据库密码","connection": [{"jdbcUrl": ["jdbc:mysql://192.168.5.180:3306/xxxx?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&serverTimezone=GMT%2B8"],"querySql": [" SELECT * from  sys_test_copy1"]}]}},"writer": {"name": "postgresqlwriter","parameter": {"username": "postgres","password": "数据库密码","writeMode": "insert!@#(user_id)","column": ["*"],                      "connection": [{"table": ["sys_test_copy1"],"jdbcUrl": "jdbc:postgresql://192.168.5.190:5432/xxxx",}]}}}]}
}

其实都是写的 insert into on CONFLICT 语句

com.alibaba.datax.plugin.rdbms.writer.util.WriterUtil

下面的代码后续调整下规则,

private static String onDuplicateKeyUpdateString(String writeMode, List<String> columnHolders) {String[] writeModeArr = writeMode.split("!@#", -1);int writeModeArrLen = writeModeArr.length;writeMode = writeModeArr[0];StringBuilder sb = new StringBuilder();if ("insert".equals(writeMode) && writeModeArrLen == 2) {sb.append(" ON CONFLICT ").append(writeModeArr[1]).append(" do nothing");}if ("update".equals(writeMode) && writeModeArrLen == 3) {sb.append(" ON CONFLICT ").append(writeModeArr[1]);String[] updateFieldArr = writeModeArr[2].replace("(","").replace(")","").split(",", -1);List<String> updateSqlList = new ArrayList<>();for (String updateField : updateFieldArr) {if (!columnHolders.contains(updateField)) {continue;}updateSqlList.add(updateField + "=EXCLUDED." + updateField);}if (updateSqlList.isEmpty()) {sb.append(" DO NOTHING");} else {sb.append(" DO UPDATE SET ").append(StringUtils.join(updateSqlList, ","));}}return sb.toString();}

小结:

pg插件,目前不支持插入更新操作,需要手工调整源码来适配。适配注意点,是根据你是否配置唯一索引来决定。(insert or update)

下期将简单介绍下,如果通过xxl-job 来执行 脚本

python datax.py ./job/mysql_postgres_job.json


 

这篇关于异构数据同步 datax (2)-postgres 写扩展的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1091592

相关文章

Java利用JSONPath操作JSON数据的技术指南

《Java利用JSONPath操作JSON数据的技术指南》JSONPath是一种强大的工具,用于查询和操作JSON数据,类似于SQL的语法,它为处理复杂的JSON数据结构提供了简单且高效... 目录1、简述2、什么是 jsONPath?3、Java 示例3.1 基本查询3.2 过滤查询3.3 递归搜索3.4

MySQL大表数据的分区与分库分表的实现

《MySQL大表数据的分区与分库分表的实现》数据库的分区和分库分表是两种常用的技术方案,本文主要介绍了MySQL大表数据的分区与分库分表的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有... 目录1. mysql大表数据的分区1.1 什么是分区?1.2 分区的类型1.3 分区的优点1.4 分

Mysql删除几亿条数据表中的部分数据的方法实现

《Mysql删除几亿条数据表中的部分数据的方法实现》在MySQL中删除一个大表中的数据时,需要特别注意操作的性能和对系统的影响,本文主要介绍了Mysql删除几亿条数据表中的部分数据的方法实现,具有一定... 目录1、需求2、方案1. 使用 DELETE 语句分批删除2. 使用 INPLACE ALTER T

Python 中的异步与同步深度解析(实践记录)

《Python中的异步与同步深度解析(实践记录)》在Python编程世界里,异步和同步的概念是理解程序执行流程和性能优化的关键,这篇文章将带你深入了解它们的差异,以及阻塞和非阻塞的特性,同时通过实际... 目录python中的异步与同步:深度解析与实践异步与同步的定义异步同步阻塞与非阻塞的概念阻塞非阻塞同步

Python Dash框架在数据可视化仪表板中的应用与实践记录

《PythonDash框架在数据可视化仪表板中的应用与实践记录》Python的PlotlyDash库提供了一种简便且强大的方式来构建和展示互动式数据仪表板,本篇文章将深入探讨如何使用Dash设计一... 目录python Dash框架在数据可视化仪表板中的应用与实践1. 什么是Plotly Dash?1.1

Redis 中的热点键和数据倾斜示例详解

《Redis中的热点键和数据倾斜示例详解》热点键是指在Redis中被频繁访问的特定键,这些键由于其高访问频率,可能导致Redis服务器的性能问题,尤其是在高并发场景下,本文给大家介绍Redis中的热... 目录Redis 中的热点键和数据倾斜热点键(Hot Key)定义特点应对策略示例数据倾斜(Data S

Python实现将MySQL中所有表的数据都导出为CSV文件并压缩

《Python实现将MySQL中所有表的数据都导出为CSV文件并压缩》这篇文章主要为大家详细介绍了如何使用Python将MySQL数据库中所有表的数据都导出为CSV文件到一个目录,并压缩为zip文件到... python将mysql数据库中所有表的数据都导出为CSV文件到一个目录,并压缩为zip文件到另一个

Java常用注解扩展对比举例详解

《Java常用注解扩展对比举例详解》:本文主要介绍Java常用注解扩展对比的相关资料,提供了丰富的代码示例,并总结了最佳实践建议,帮助开发者更好地理解和应用这些注解,需要的朋友可以参考下... 目录一、@Controller 与 @RestController 对比二、使用 @Data 与 不使用 @Dat

SpringBoot整合jasypt实现重要数据加密

《SpringBoot整合jasypt实现重要数据加密》Jasypt是一个专注于简化Java加密操作的开源工具,:本文主要介绍详细介绍了如何使用jasypt实现重要数据加密,感兴趣的小伙伴可... 目录jasypt简介 jasypt的优点SpringBoot使用jasypt创建mapper接口配置文件加密

使用Python高效获取网络数据的操作指南

《使用Python高效获取网络数据的操作指南》网络爬虫是一种自动化程序,用于访问和提取网站上的数据,Python是进行网络爬虫开发的理想语言,拥有丰富的库和工具,使得编写和维护爬虫变得简单高效,本文将... 目录网络爬虫的基本概念常用库介绍安装库Requests和BeautifulSoup爬虫开发发送请求解