01 / 一句话讲清楚

实时数据接入是指 数据产生后秒级到达 Greenplum ，无需等待每日 ETL 批次。

Greenplum 支持多种实时接入方式：

• gp_kafka ：官方 Kafka 连接器

• gpcopy ：实时同步

• 外部表 + PXF ：流式读取

• JDBC 批量写入 ：应用直连

📌 实时接入的核心价值：从“昨天发生了什么”升级到“刚才发生了什么”。

02 / 四种实时接入方式对比

03 / 方式一：gp_kafka（官方推荐）

3.1 架构原理

Kafka Topic → gp_kafka Connector → Greenplum 表

                  ↓

            并行读取、分区映射

3.2 安装配置

# 1. 下载 gp_kafka 连接器

# 从 Greenplum 官方仓库下载对应版本

# 2. 配置 Kafka 连接

# $GPHOME/connectors/kafka/etc/kafka.properties

bootstrap.servers=kafka1:9092,kafka2:9092,kafka3:9092

group.id=greenplum-consumer

auto.offset.reset=earliest

3.3 创建目标表

-- 创建接收实时数据的表

CREATE TABLE realtime_orders (

    order_id bigint,

    user_id int,

    amount numeric(10,2),

    product_id int,

    order_time timestamp,

    load_time timestamp DEFAULT now()

) WITH (appendonly=true, orientation=column, compresstype=zstd)

DISTRIBUTED BY (order_id)

PARTITION BY RANGE (order_time);

3.4 启动实时消费

# 启动 Kafka 消费者，将数据写入 Greenplum

gp_kafka_consumer \

    --kafka-topic orders_topic \

    --gp-table realtime_orders \

    --gp-host gp-master \

    --gp-port 5432 \

    --gp-db mydb \

    --gp-user gpadmin \

    --batch-size 10000 \

    --flush-interval 5

3.5 处理 JSON 数据

-- 如果 Kafka 消息是 JSON 格式

CREATE TABLE realtime_logs (

    log_data jsonb,

    load_time timestamp DEFAULT now()

) DISTRIBUTED RANDOMLY;

-- 消费时自动解析 JSON

-- gp_kafka_consumer 支持 JSON 到列的映射

04 / 方式二：PXF + Kafka

4.1 创建 Kafka 外部表

-- 使用 PXF 创建 Kafka 外部表

CREATE EXTERNAL TABLE ext_kafka_orders (

    order_id bigint,

    user_id int,

    amount numeric,

    order_time timestamp

)

LOCATION ('pxf://kafka:9092/orders_topic?PROFILE=Kafka')

FORMAT 'CUSTOM' (formatter='pxfwritable_import');

4.2 实时消费

-- 定期将外部表数据写入本地表

-- 可以配合 pg_cron 定时执行

INSERT INTO realtime_orders

SELECT * FROM ext_kafka_orders

WHERE order_time > (SELECT max(order_time) FROM realtime_orders);

05 / 方式三：JDBC 批量写入

5.1 应用层批量写入

// Java 示例：使用批量提交

String sql = "INSERT INTO realtime_orders (order_id, user_id, amount, order_time) VALUES (?, ?, ?, ?)";

try (PreparedStatement pstmt = connection.prepareStatement(sql)) {

    for (Order order : orders) {

        pstmt.setLong(1, order.getId());

        pstmt.setInt(2, order.getUserId());

        pstmt.setBigDecimal(3, order.getAmount());

        pstmt.setTimestamp(4, order.getTime());

        pstmt.addBatch();

        // 每 1000 条提交一次

        if (++count % 1000 == 0) {

            pstmt.executeBatch();

        }

    }

    pstmt.executeBatch();

}

5.2 使用 COPY 协议（更快）

// 使用 PostgreSQL JDBC 的 CopyManager

CopyManager copyManager = connection.unwrap(PGConnection.class).getCopyAPI();

CopyIn copyIn = copyManager.copyIn("COPY realtime_orders FROM STDIN");

for (Order order : orders) {

    String line = order.getId() + "," + order.getUserId() + "," + order.getAmount() + "," + order.getTime() + "\n";

    copyIn.writeToCopy(line.getBytes());

}

copyIn.endCopy();

06 / 性能优化建议

写在最后

实时数据接入的选型建议：

从 T+1 到秒级，差的不是技术，是架构思维。实时不是目的，让数据更快产生价值才是。