简介这份资源是经过源码改造的DataX数据同步引擎专门解决PostgreSQL中geometry地理空间类型无法通过原生DataX正常同步的问题面向需要做空间数据迁移、ETL开发与数据库集成的工程师。包内共113个文件以97个jar依赖为核心辅以10个json配置、3个Python脚本、1个xml、1个md说明和1个properties文件压缩包约56.67MB。使用时直接用下载的DataX替换本地版本即可plugin目录下的reader与writer仅保留postgresql和rdbms模块如需其他数据库支持把对应模块复制进来就能扩展。改造后的引擎可让geometry字段在读写链路中正确解析与落库省去自行研究源码、调试类型转换的时间适合直接投入生产环境验证。目前已有490人学习下载对做空间数据同步的开发者有较高参考价值。1. DataX 改造引擎支持 geometry 类型同步从丢字段到空间数据完整搬运线上做数仓同步的同行大概率遇到过这种场景MySQL 里存着POINT、POLYGON、MULTIPOLYGON这类 geometry 字段用 DataX 从 MySQL 抽到 Hive 或者另一个 MySQL 时任务不报错但目标表里 geometry 列全是null或者干脆在 Reader 阶段就被静默跳过。这不是配置写错了是 DataX 默认的mysqlreader和mysqlwriter对 geometry 类型没有做显式处理——它走的是 JDBC 的getObject而 MySQL Connector/J 对 geometry 返回的是byte[]DataX 内部的类型转换器不认识这个类型于是走了兜底逻辑把值丢掉了。这篇笔记讲的就是怎么在 DataX 的 reader 和 writer 两端把 geometry 类型接住让空间数据能完整同步适合正在做 GIS 数据入仓、或者被 geometry 字段卡住同步链路的工程师。2. 先搞清楚 DataX 为什么接不住 geometry2.1 DataX 的类型转换链路与 geometry 的断点位置DataX 的同步流程可以拆成四段Reader 从源端拉数据 → 转成 DataX 内部Record→ Writer 从Record取值写入目标端。问题出在第一段和第三段。MySQL Reader 在mysqlreader插件里对每一列调用ResultSet.getObject(columnIndex)。对于VARCHAR、INT、DATETIME这些常规类型JDBC 驱动返回的是String、Integer、TimestampDataX 的Column能直接识别。但 geometry 类型在 MySQL Connector/J 里返回的是byte[]里面是 WKBWell-Known Binary格式的二进制。DataX 的Record在setColumn时会走Column的类型判断byte[]不在它支持的类型列表里于是被标记为null或者直接抛类型转换异常被上层吞掉。Writer 端同理。即使你在 Reader 端把 geometry 转成了字符串Writer 端如果目标列是 geometry 类型mysqlwriter拼 SQL 时用的是PreparedStatement.setObject传字符串进去 MySQL 不会自动转成 geometry会报Cannot get geometry object from data you send to the GEOMETRY field。所以改造的核心思路就一句话在 Reader 端把 geometry 转成可传输的字符串WKT 或 WKB 的十六进制在 Writer 端再把字符串转回 geometry。中间传输用字符串绕开 DataX 内部类型系统的限制。2.2 两种传输格式的选型WKT 还是 WKB Hex把 geometry 转成字符串有两条路格式转换函数体积精度可读性适用场景WKTST_AsText(g)较大浮点文本可能有精度损失人可读调试、小数据量、需要人工核对WKB HexHEX(ST_AsBinary(g))较小二进制精确不可读生产同步、大数据量、精度敏感我一般选 WKB Hex。原因很直接WKT 在浮点数转文本时会有精度截断POINT(116.397428 39.90923)这种坐标转成文本再转回来小数点后可能少几位做地图渲染时偏移肉眼可见。WKB Hex 是二进制转十六进制完全无损而且体积比 WKT 小 30% 到 50%网络传输和落盘都更省。选型确定后改造就分两端Reader 端改 SQL 或者改插件代码Writer 端改写入逻辑。3. Reader 端改造让 geometry 以字符串形式进入 DataX3.1 方案一不改代码用 SQL 层转换最快落地如果你的同步任务允许自定义 SQL这是成本最低的做法。在mysqlreader的querySql里把 geometry 列用HEX(ST_AsBinary(geom_col))包一层-- mysqlreader 的 querySql 配置 SELECT id, name, HEX(ST_AsBinary(geom_col)) AS geom_col_wkb, created_at FROM source_table WHERE updated_at ${last_sync_time}这样 Reader 拿到的geom_col_wkb就是一个普通的十六进制字符串DataX 能正常识别为STRING类型一路传到 Writer 端不会丢。逻辑说明ST_AsBinary把 geometry 转成 WKB 二进制HEX把二进制转成十六进制字符串。两层函数嵌套后输出是纯 ASCII 字符DataX 的StringColumn直接接住。参数说明geom_col替换成你实际的 geometry 列名。如果表里有多个 geometry 列每一列都要单独包。注意ST_AsBinary在 MySQL 5.7 和 8.0 里都有但 8.0 的 SRID 处理更严格如果源表带了 SRID 且不是 0ST_AsBinary会带上 SRID 信息目标端解析时要注意。这个方案的边界只适合querySql模式tablecolumn模式没法在 SQL 层做转换。另外如果 geometry 列数据量特别大比如单个MULTIPOLYGON几十 MBHEX之后字符串会膨胀一倍要注意 DataX 的 channel 内存。3.2 方案二改 mysqlreader 插件在代码层做类型映射当同步任务必须用tablecolumn模式或者 geometry 列很多不想每列都写 SQL 函数时就得改插件源码。DataX 的mysqlreader核心类在com.alibaba.datax.plugin.reader.mysqlreader.MysqlReader$Task找到getObject那一段// 改造前直接 getObjectgeometry 返回 byte[] 被丢弃 Object value rs.getObject(i 1); // 改造后判断列类型geometry 走 WKB Hex 转换 int sqlType rs.getMetaData().getColumnType(i 1); Object value; if (sqlType java.sql.Types.BINARY || sqlType java.sql.Types.VARBINARY || sqlType java.sql.Types.LONGVARBINARY) { // MySQL geometry 在 JDBC 里映射为 BINARY/VARBINARY byte[] bytes rs.getBytes(i 1); if (bytes ! null) { value javax.xml.bind.DatatypeConverter.printHexBinary(bytes); } else { value null; } } else { value rs.getObject(i 1); }逻辑说明MySQL Connector/J 把 geometry 类型的 JDBC type 报成BINARY或VARBINARY所以用getColumnType判断命中就走getBytes拿原始 WKB再用DatatypeConverter.printHexBinary转成十六进制字符串。这样不改 SQLtable模式也能用。参数说明DatatypeConverter在 JDK 9 之后被标记为 deprecated如果项目用高版本 JDK换成java.util.HexFormat.of().formatHex(bytes)。另外这个判断条件要放在getObject之前否则getObject已经返回byte[]了再判断类型没意义。改完插件后需要重新打包datax-mysqlreader的 jar替换到 DataX 的plugin/reader/目录下。注意 DataX 的插件加载有缓存替换后要重启 DataX 进程否则还是走旧类。3.3 Reader 端改造后的验证方法改完不要直接上生产先跑一个最小验证。建一张带 geometry 的测试表CREATE TABLE geo_test ( id INT PRIMARY KEY, name VARCHAR(64), geom_col GEOMETRY ); INSERT INTO geo_test VALUES (1, point_a, ST_GeomFromText(POINT(116.397428 39.90923))), (2, poly_b, ST_GeomFromText(POLYGON((0 0, 1 0, 1 1, 0 1, 0 0))));然后写一个只抽不写的 DataX 任务在 Writer 端用streamwriter把内容打到控制台{ job: { content: [{ reader: { name: mysqlreader, parameter: { username: root, password: xxx, column: [id, name, geom_col], connection: [{ table: [geo_test], jdbcUrl: [jdbc:mysql://127.0.0.1:3306/test] }] } }, writer: { name: streamwriter, parameter: { print: true } } }] } }跑完后看控制台输出geom_col那一列应该是十六进制字符串POINT对应的 WKB Hex 开头是0101000000POLYGON开头是0103000000。如果输出是null说明改造没生效检查 jar 是否替换成功、DataX 是否重启。4. Writer 端改造把字符串还原成 geometry 写入目标表4.1 方案一目标端用 SQL 函数还原适合 MySQL 目标如果目标端也是 MySQL最省事的做法是在preSql或者写入时用ST_GeomFromWKB(UNHEX(?))。但mysqlwriter的preSql是在写入前执行的没法对每一行做转换。所以这个方案需要改 Writer 的 SQL 拼装逻辑。在mysqlwriter的Task类里找到拼INSERT语句的地方把 geometry 列的占位符从?改成ST_GeomFromWKB(UNHEX(?))// 改造前所有列统一用 ? 占位 StringBuilder sb new StringBuilder(INSERT INTO table (); // ... 拼列名 sb.append() VALUES (); for (int i 0; i columns.size(); i) { sb.append(i 0 ? ? : ,?); } sb.append()); // 改造后geometry 列用函数包裹 // 假设 geometry 列在 columns 里的索引是 geoIndex for (int i 0; i columns.size(); i) { if (i geoIndex) { sb.append(i 0 ? ST_GeomFromWKB(UNHEX(?)) : ,ST_GeomFromWKB(UNHEX(?))); } else { sb.append(i 0 ? ? : ,?); } }逻辑说明UNHEX(?)把十六进制字符串转回二进制 WKBST_GeomFromWKB把 WKB 转成 geometry 对象。这样 Writer 端setString传进去的十六进制字符串就能正确写入 geometry 列。参数说明geoIndex需要根据实际列顺序确定硬编码不优雅但最直接。更好的做法是从 DataX 的column配置里读列类型但 DataX 的 Writer 配置里没有类型信息所以实际改造时通常是在 Writer 的parameter里加一个geometryColumns数组显式声明哪些列是 geometry。4.2 方案二目标端用 JDBC 的 setBytes 直接写 WKB如果不想在 SQL 里包函数可以在 Writer 端把十六进制字符串转回byte[]然后用setBytes写入。MySQL 的 JDBC 驱动对 geometry 列的setBytes是支持的前提是字节内容是合法的 WKB。// 在 mysqlwriter 的 Task 里替换 setString 的逻辑 if (isGeometryColumn(i)) { String hexStr record.getColumn(i).asString(); if (hexStr ! null) { byte[] wkb java.util.HexFormat.of().parseHex(hexStr); preparedStatement.setBytes(i 1, wkb); } else { preparedStatement.setNull(i 1, java.sql.Types.BINARY); } } else { preparedStatement.setObject(i 1, record.getColumn(i).getRawData()); }逻辑说明HexFormat.parseHex把十六进制字符串转回byte[]setBytes直接写二进制。MySQL 驱动收到setBytes后如果目标列是 geometry会自动做 WKB 解析。参数说明isGeometryColumn(i)需要你根据 Writer 配置里的geometryColumns列表判断。注意setNull时 type 要用Types.BINARY用Types.OTHER有些驱动版本会报错。这个方案比 SQL 函数方案更干净不依赖目标端的 SQL 函数支持而且性能更好——少了一层UNHEX和ST_GeomFromWKB的函数调用。4.3 Writer 端改造后的验证与回读写完不算完要回读验证。在目标表上跑SELECT id, name, ST_AsText(geom_col) AS wkt, ST_SRID(geom_col) AS srid FROM target_table WHERE id IN (1, 2);预期输出id1的wkt是POINT(116.397428 39.90923)id2的wkt是POLYGON((0 0,1 0,1 1,0 1,0 0))。如果wkt是null说明 Writer 端没写进去如果wkt有值但坐标不对检查 SRID 是否在传输过程中丢了。SRID 是个容易翻车的点。MySQL 8.0 的 geometry 列可以带 SRIDST_AsBinary默认会带上 SRID 信息WKB 的第一个字节之后有 4 字节的 SRID但ST_GeomFromWKB默认按 SRID 0 解析。如果源表 SRID 是 4326目标表也是 4326需要在 Writer 端用ST_GeomFromWKB(UNHEX(?), 4326)显式指定或者用ST_SRID(g, 4326)后处理。5. 避坑与排查geometry 同步的五个血泪教训5.1 现象任务不报错但目标表 geometry 列全是 null原因Reader 端getObject返回byte[]DataX 内部类型转换把它当成了不支持的类型静默置 null。这是最常见的情况因为 DataX 对不认识的类型不抛异常只打 debug 日志。解决按第 3 章的方案改 Reader用 SQL 层HEX(ST_AsBinary())或者改插件代码。改完后用streamwriter验证 Reader 输出确认不是 null 再往下走。5.2 现象Writer 报Cannot get geometry object from data you send to the GEOMETRY field原因Writer 端把十六进制字符串当普通字符串setString写入了 geometry 列MySQL 拒绝隐式转换。解决按第 4 章改 Writer用ST_GeomFromWKB(UNHEX(?))或者setBytes。注意setString和setBytes不能混用geometry 列必须走二进制路径。5.3 现象同步后坐标偏移小数点后几位对不上原因用了 WKT 格式传输浮点数转文本时精度截断。POINT(116.397428 39.90923)转 WKT 再转回来可能变成POINT(116.39743 39.90923)。解决换 WKB Hex 格式。WKB 是二进制精确表示不经过文本转换不会有精度损失。如果已经用了 WKT 且数据量不大可以在目标端用ST_GeomFromText重新解析但精度损失不可逆。5.4 现象大数据量同步时 DataX 内存溢出原因geometry 转十六进制后字符串体积膨胀单个MULTIPOLYGON可能几十 MBDataX 的 channel 默认 512MB 内存扛不住。解决调小channel数或者在 Reader 的querySql里加LIMIT分批。更彻底的做法是在 Reader 端不转十六进制直接传byte[]但需要改 DataX 内部Record的类型支持改动量大。实际项目中我一般用分批同步每批 1000 行配合where id lastMaxId做增量。5.5 现象MySQL 5.7 和 8.0 之间同步 geometry 报 SRID 错误原因5.7 的 geometry 列默认 SRID 是 08.0 可以显式指定 SRID。ST_AsBinary在 8.0 里如果列带了 SRID输出的 WKB 会包含 SRID 信息5.7 的ST_GeomFromWKB不认。解决在 Reader 端用ST_AsBinary(ST_SRID(geom_col, 0))强制把 SRID 归零或者在 Writer 端用ST_GeomFromWKB(UNHEX(?), 0)显式指定 SRID。如果业务需要保留 SRID目标端也必须是 8.0并且 Writer 端要传对应的 SRID 值。6. 进阶用 DataX 做增量 geometry 同步的一个实用技巧增量同步 geometry 的难点在于updated_at字段能标记行变更但 geometry 列本身没有变更标记。如果一行数据的 geometry 被更新了updated_at会变这没问题。但如果 geometry 列被更新而updated_at没变比如某些批量 GIS 操作直接改了几何图形但没触发时间戳更新增量同步就会漏数据。我一般会在源表加一个geom_version字段用触发器或者应用层在 geometry 变更时递增。DataX 的querySql里用WHERE geom_version ${last_version}做增量条件。这样即使updated_at没变geometry 变更也能被捕获。另一个技巧是在 Writer 端写入前先用ST_Equals判断目标行是否已存在且几何相同相同就跳过不同就更新。这需要在 Writer 端改逻辑用INSERT ... ON DUPLICATE KEY UPDATE配合ST_Equals条件INSERT INTO target_table (id, name, geom_col) VALUES (?, ?, ST_GeomFromWKB(UNHEX(?))) ON DUPLICATE KEY UPDATE name IF(ST_Equals(geom_col, VALUES(geom_col)), name, VALUES(name)), geom_col IF(ST_Equals(geom_col, VALUES(geom_col)), geom_col, VALUES(geom_col));逻辑说明ON DUPLICATE KEY UPDATE在主键冲突时触发ST_Equals判断新旧几何是否相同相同就不更新不同才更新。这样避免了无意义的 geometry 重写减少 IO。参数说明VALUES(geom_col)在 MySQL 8.0.20 之后被标记为 deprecated换成别名写法AS new ON DUPLICATE KEY UPDATE geom_col IF(ST_Equals(geom_col, new.geom_col), geom_col, new.geom_col)。注意ST_Equals对 SRID 敏感两边 SRID 必须一致。这个方案我用了两年多在日增百万级 geometry 行的场景下比全量覆盖写入省了大概 60% 的 IO。唯一要注意的是ST_Equals对复杂多边形的计算开销不小如果 geometry 特别复杂建议先比ST_Area或者ST_NPoints做快速判断相同再跳过不同再走ST_Equals精确判断。踩过最大的坑是 SRID 不一致导致ST_Equals永远返回 false然后每次同步都全量更新IO 直接打满。后来在 Writer 端加了一个 SRID 校验源和目标 SRID 不一致就直接报错退出不再往下走。这个校验逻辑很简单但省了很多事后排查的时间。希望帮到你。本文还有配套的精品资源点击获取
