企业官网建设流程全解析

数据仓库和计算引擎之间的数据搬运，是每个做数仓的人都绕不开的日常。我所在的团队每天有大量ETL任务需要在计算引擎和MPP数据库之间来回倒腾数据，早些年用原生JDBC一条条写、一批批读，速度慢不说，还经常因为连接管理不当把数据库连接池打满。后来换成官方Connector，情况好了不少，但遇到大表、宽表、分区表的时候，写入性能依然是个瓶颈。这次想聊的，就是围绕Spark与Greenplum之间读写这条链路，尤其是Greenplum-Spark Connector在高速写入场景下的那些门道——它到底是什么、为什么快、怎么配置、踩过哪些坑。不管你是刚接手数仓ETL的新手，还是正在为写入慢发愁的老手，这篇应该都能给你一些可以直接抄作业的东西。

1. 先搞清楚这条链路到底在解决什么问题

1.1 为什么原生JDBC方式会成为瓶颈

很多人第一次做Spark写Greenplum，直觉就是拿JDBC驱动，df.write.mode("append").jdbc(url, table, props)，一行代码搞定。小数据量下确实没问题，几万行数据眨眼就进去了。但只要数据量上到千万级、亿级，问题就暴露得非常明显。

根本原因在于JDBC写入的模型是"单点串行"的。Spark的每个分区通过一个JDBC连接，用INSERT INTO ... VALUES (...)或者批量addBatch的方式往Greenplum里灌数据。Greenplum虽然是MPP架构，底层有多个Segment节点可以并行处理，但JDBC这条通道把并行度卡死在了Spark侧的连接数上。你开10个分区就是10条连接，每条连接对应Greenplum的一个Master入口，Master再分发到Segment——这个过程中Master很容易成为单点瓶颈。

更麻烦的是事务和锁。大批量INSERT在Greenplum里会持有锁，并发写入同一张表时锁竞争激烈，写入速度断崖式下跌。我实测过一个场景：2000万行、约40个字段的宽表，用原生JDBC批量写入，耗时接近40分钟，而且期间数据库CPU飙到90%以上，其他查询基本没法跑。

1.2 Connector的核心思路：绕开Master，直连Segment

Greenplum-Spark Connector的设计思路和JDBC完全不同。它的核心是利用Greenplum的外部表（External Table）机制和gpfdist协议，让数据并行地、点对点地写入各个Segment，而不是全部挤在Master这一个入口。

打个比方：JDBC写入像是所有快递都送到一个总仓，再由总仓分拣到各个网点，总仓必然拥堵；而Connector的做法是让每个网点直接派车到发货点拉货，总仓只负责登记一下地址信息。这个"登记地址"的动作，就是Connector先在Greenplum里创建一张可写的外部表（Writable External Table），指向Spark侧启动的gpfdist服务；然后Spark的各个Executor通过gpfdist把数据分片推送到对应的Segment。整个过程Master只参与元数据操作，数据流完全不经过Master。

这个架构带来的直接好处是：写入吞吐量随Segment数量线性扩展，Master压力极小，而且因为绕开了逐条INSERT，锁竞争问题也基本消失。同样是那个2000万行的场景，换成Connector之后，写入时间压到了6分钟左右，提升非常明显。

1.3 读和写是两条不同的路径

需要特别说明的是，Connector的读和写走的是两套机制。读Greenplum的时候，Connector同样利用gpfdist，但方向反过来——Greenplum的Segment把数据通过gpfdist推给Spark的Executor，Spark侧并行接收。写的时候则是Spark主动通过gpfdist把数据推给Segment。

这两条路径的配置参数、调优重点都不一样。读的时候更关注分区裁剪和并行度匹配，写的时候更关注批次大小、gpfdist服务数量和事务提交策略。后面我会分开讲，避免混在一起说不清楚。

2. 环境准备与依赖配置的关键细节

2.1 版本匹配是第一条铁律

Connector对Spark和Greenplum的版本都有要求，这一点必须先确认，否则后面全是玄学问题。根据我的经验，主流的对应关系大致是这样的：

这里有个坑我踩过：Spark 3.x默认用的是Scala 2.12，而早期一些Connector包是拿Scala 2.11编译的，直接扔进去会报NoSuchMethodError或者ClassNotFoundException。所以下载jar包的时候一定要看清楚编译时的Scala版本，宁可多花两分钟核对，也别等到运行时报错再回头查。

另外，Greenplum 7.x在外部表语法上有一些调整，如果用的是老版本Connector配新版本Greenplum，可能会在创建外部表那一步失败。稳妥的做法是Connector版本不低于2.x。

2.2 依赖引入的两种方式

如果项目用Maven管理，直接在pom.xml里加依赖最省事：

<dependency> <groupId>com.example.greenplum</groupId> <artifactId>greenplum-spark-connector</artifactId> <version>2.1.0</version> </dependency>

注意这里的groupId和artifactId是示意，实际以你拿到的包为准。如果是离线环境或者不想走依赖管理，就把jar包手动放到Spark的jars目录，或者提交任务时用--jars指定。我一般倾向于后者，因为不同任务可能用不同版本，放在任务级别更灵活，不会污染全局环境。

spark-submit \ --jars /path/to/greenplum-spark-connector.jar \ --class com.example.MainJob \ your-app.jar

2.3 gpfdist服务的前置条件

Connector的读写都依赖gpfdist，所以运行Spark任务的节点上必须有gpfdist可执行文件，并且能被Spark Executor调用到。gpfdist通常随Greenplum客户端工具一起安装，路径一般在$GPHOME/bin/gpfdist。

这里有个容易被忽略的点：gpfdist服务监听的端口必须在Spark Executor所在机器上可访问，同时Greenplum的Segment节点也要能访问到这个地址。也就是说，如果Spark跑在独立的集群上，Greenplum的Segment和Spark的Executor之间网络必须打通。我见过有团队把Spark和Greenplum部署在两个完全隔离的网络里，结果Connector一直卡在连接超时，排查了半天才发现是网络策略没放行。

提示：部署前先用telnet或nc从Greenplum的Segment节点测试一下能否连通Spark Executor所在机器的gpfdist端口，这个动作能省掉后面大量排查时间。

2.4 权限与用户配置

写入Greenplum需要目标表的INSERT权限，以及创建外部表的权限（因为Connector会在写入过程中临时创建可写外部表）。如果用的是受限账号，很可能在创建外部表这一步被拒绝。建议给ETL任务单独配一个具备CREATE和INSERT权限的账号，别图省事直接用超级用户，安全上也不合适。

3. 高速写入的核心机制与参数调优

3.1 写入流程拆解：从DataFrame到Segment

理解写入流程，调优才有方向。一次完整的Connector写入大致经历这么几步：

1. Spark侧根据DataFrame的分区数，规划出若干个数据分片；
2. Connector在Greenplum中创建一张可写外部表，指定LOCATION为Spark侧gpfdist服务的地址列表；
3. Spark的每个Executor启动或复用gpfdist服务，等待数据推送；
4. Executor把各自分区的数据通过gpfdist协议推送到Greenplum；
5. Greenplum的Segment并行接收数据，写入外部表；
6. 数据全部推送完成后，Connector执行INSERT INTO 目标表 SELECT * FROM 外部表，把数据从外部表落到目标表；
7. 清理临时外部表。

关键在第6步——数据其实先进了外部表，再通过一条INSERT SELECT落到目标表。这意味着写入性能不仅取决于gpfdist的推送速度，还取决于这条INSERT SELECT在Greenplum内部的执行效率。所以调优要两头看：一头是Spark侧的并行推送，一头是Greenplum侧的内部写入。

3.2 并行度：不是越大越好

很多人以为并行度拉满就快，实际上并行度要和Greenplum的Segment数量匹配。假设你的Greenplum集群有16个Segment，那么Spark侧的有效并行度最好在16的整数倍附近，比如16、32、48。如果Spark分区数是100，而Segment只有16个，多出来的并行度并不能提升写入速度，反而会增加gpfdist服务的连接开销和Greenplum侧的调度负担。

我的做法是先查清楚集群的Segment数量：

SELECT COUNT(*) FROM gp_segment_configuration WHERE content >= 0 AND role = 'p';

然后据此设置Spark的并行度。如果是通过repartition控制，建议设成Segment数量的1到2倍。实测下来，1倍时资源利用率最均衡，2倍时在数据倾斜场景下容错性更好。

3.3 批次大小与gpfdist服务数

Connector写入时有两个关键参数：批次大小（batch size）和gpfdist服务数量。批次大小决定每次推送多少行数据，太小则网络往返次数多，太大则单次内存占用高、失败重传代价大。经验值是每批1万到5万行，具体看单行宽度。40个字段的宽表，我一般设2万行一批；字段少、行窄的表可以设到5万。

gpfdist服务数量决定了Spark侧同时开多少个推送通道。这个值不宜超过Executor数量，也不宜超过Segment数量。比如10个Executor、16个Segment，gpfdist服务数设8到10比较合适。设太多会导致端口占用和连接管理混乱，设太少则推送通道不够，Segment吃不饱。

val writeOptions = Map( "url" -> "jdbc:postgresql://gp-master:5432/mydb", "dbschema" -> "public", "dbtable" -> "target_table", "user" -> "etl_user", "password" -> "******", "batchsize" -> "20000", "numgpfdist" -> "8" ) df.write .format("greenplum") .options(writeOptions) .mode("append") .save()

上面这段是示意写法，实际参数名以你使用的Connector版本文档为准。核心是理解每个参数背后的含义，而不是死记硬背。

3.4 事务与提交策略

Connector写入默认是整批提交，也就是说所有数据推送完成后才执行最终的INSERT SELECT并提交。这个策略保证了原子性，但代价是如果中途失败，整批数据都要重来。对于超大规模数据，可以考虑分批写入——把DataFrame按某个业务键切成多批，每批独立提交。这样单批失败只影响那一批，重试成本低。

不过分批也有代价：每批都要创建和清理外部表，有额外的元数据开销。所以分批的粒度要权衡，一般单批控制在500万到1000万行比较合适。

3.5 数据倾斜的处理

写入慢有时候不是Connector的问题，而是数据倾斜。如果DataFrame按某个键分区后，某个分区的数据量远超其他分区，那个分区的Executor就会成为长尾，拖慢整体进度。表现是大部分Executor早早完成，就剩一两个还在跑。

解决办法是在写入前做一次重分区，打散倾斜。可以用repartition(n)做随机重分区，也可以用repartition(n, col)按列重分区。如果业务上允许，随机重分区最简单有效。代价是多一次shuffle，但相比长尾带来的等待，通常划算。

4. 读取路径的差异化配置

4.1 读Greenplum的并行机制

读的时候方向反过来：Greenplum的Segment通过gpfdist把数据推给Spark。Connector会根据Segment数量和Spark分区数协商并行度。读的核心调优点是分区裁剪——尽量只读需要的列和行，减少数据传输量。

如果目标表很大但每次只取增量数据，强烈建议在读取时带上过滤条件，让Greenplum在Segment侧就完成过滤，而不是把全表拉到Spark再filter。Connector支持通过query或dbtable配合子查询的方式下推过滤条件。

val readOptions = Map( "url" -> "jdbc:postgresql://gp-master:5432/mydb", "dbtable" -> "(SELECT * FROM big_table WHERE dt = '2024-01-01') t", "user" -> "etl_user", "password" -> "******", "partitionColumn" -> "id", "lowerBound" -> "1", "upperBound" -> "10000000", "numPartitions" -> "16" ) val df = spark.read .format("greenplum") .options(readOptions) .load()

注意partitionColumn必须是数值型或可比较类型，且分布相对均匀，否则分区会倾斜。如果表没有合适的数值列，可以考虑用哈希列或者干脆用单分区读取再在Spark侧重分区。

4.2 读取时的内存与超时设置

读大表时容易遇到Executor OOM或者gpfdist超时。前者是因为单批拉取的数据量太大，后者是因为网络或Segment响应慢。对应的调整是减小读取批次、增大超时时间。另外，读取时Executor的内存要给足，尤其是宽表场景，一行数据可能就有几KB，几万行一批就是上百MB。

4.3 读写混合场景的注意事项

有些任务需要先读后写，比如从Greenplum读一张表，处理后写回另一张表。这种场景下要注意gpfdist服务的端口冲突——读和写可能同时需要gpfdist，如果端口配置不当会互相干扰。稳妥的做法是读写使用不同的端口段，或者在任务设计上避免读写并发。

5. 常见问题排查与避坑实录

5.1 典型问题速查表

5.2 外部表残留这个坑要重点说

Connector在写入时会创建临时外部表，正常结束后会清理。但如果任务异常中断（比如被kill、Executor崩溃），清理逻辑可能没执行，外部表就残留在Greenplum里了。残留多了不仅占元数据空间，还可能导致后续同名外部表创建失败。

我的做法是在任务外层加一层兜底清理：任务开始前先查一下有没有同名的残留外部表，有就删掉；任务结束后无论成败都尝试清理一次。查询外部表的语句大致是查pg_class里relkind='f'或者查gp_dist_random相关视图，具体看Greenplum版本。这个兜底逻辑虽然多写几行代码，但能避免很多莫名其妙的失败。

5.3 密码明文的问题

示例里密码是明文写的，实际生产环境千万别这么干。可以用Spark的credential管理机制，或者从配置中心、环境变量读取。我见过有团队把密码硬编码在代码里提交到代码仓库，这是大忌。

5.4 大批量写入失败后的重试策略

前面提到分批写入能降低重试成本。除此之外，重试本身也要设计：不是所有失败都值得重试，网络抖动、临时锁冲突可以重试，权限问题、表结构不匹配重试多少次都没用。建议在重试逻辑里区分异常类型，对可重试异常做指数退避重试，对不可重试异常直接失败并告警。

5.5 监控与度量

写入任务跑起来之后，怎么知道它健康不健康？我一般关注几个指标：gpfdist的推送速率、Greenplum侧的Segment写入速率、外部表到目标表的INSERT SELECT耗时、以及各Executor的完成时间分布。前三个能看出瓶颈在哪一段，最后一个能看出有没有长尾。这些指标可以通过Spark的metrics系统采集，也可以在Greenplum侧查pg_stat_activity看当前执行的语句。

6. 一些实战中的经验体会

Connector这套东西，用熟了之后确实比JDBC省心太多，但它不是银弹。我最大的体会是：性能调优的前提是理解数据流走向。你得清楚数据从Spark的哪个分区出发，经过哪个gpfdist服务，落到Greenplum的哪个Segment，中间哪一段可能堵。想清楚这条链路，参数调整才有依据，否则就是瞎试。

另一个体会是版本管理要严格。Connector、Spark、Greenplum三者的版本组合最好固定下来，做成团队的标准镜像或依赖清单。我见过因为不同任务用了不同版本Connector，导致同一个集群上行为不一致，排查起来非常痛苦。

最后分享一个小技巧：新表第一次写入前，先用小批量数据跑一遍完整流程，确认权限、网络、外部表创建都正常，再上全量。这个"冒烟测试"花不了几分钟，但能提前暴露大部分环境问题，比全量跑到一半失败再回头查要划算得多。至于后续扩展，如果写入频率很高，可以考虑把gpfdist服务做成常驻的，避免每次任务都启停，能再省一点开销——不过这就涉及服务管理和端口规划，复杂度会上一个台阶，看具体场景是否值得。