100天跟着CP学PostgreSQL+AI,第12天 : 用 PostgreSQL 构建 AI 训练数据管道（ETL 全流程）

# 安装Flink 1.17.1
wget https://archive.apache.org/dist/flink/flink-1.17.1/flink-1.17.1-bin-scala_2.12.tgz
tar -zxvf flink-1.17.1-bin-scala_2.12.tgz
# 添加PostgreSQL CDC依赖
cp flink-sql-connector-postgres-cdc-2.3.0.jar flink/lib/

CREATE TABLE source.click_log (
    log_id BIGINT PRIMARY KEY,
    user_id BIGINT,
    item_id BIGINT,
    click_time TIMESTAMP,
    client_ip VARCHAR(32)
);

CREATE TABLE target.click_log (
    log_id BIGINT,
    user_id BIGINT,
    item_id BIGINT,
    click_time TIMESTAMP,
    etl_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    op_type VARCHAR(10)  -- 数据操作类型：INSERT/UPDATE/DELETE
);

CREATE TABLE postgres_source (
    log_id BIGINT,
    user_id BIGINT,
    item_id BIGINT,
    click_time TIMESTAMP,
    __deleted BOOLEAN,  -- CDC删除标记
    PRIMARY KEY (log_id) NOT ENFORCED
) WITH (
    'connector' = 'postgres-cdc',
    'hostname' = 'source-postgres',
    'port' = '5432',
    'database-name' = 'recommendation_db',
    'schema-name' = 'public',
    'table-name' = 'click_log',
    'username' = 'admin',
    'password' = 'password'
);
INSERT INTO target.click_log (log_id, user_id, item_id, click_time, op_type)
SELECT 
    log_id, 
    user_id, 
    item_id, 
    click_time,
    CASE WHEN __deleted THEN 'DELETE' ELSE 'INSERT' END  -- 简化UPDATE处理，实际需处理字段级变更
FROM postgres_source;

CREATE TABLE target.item_dim (
    item_id BIGINT,
    item_name VARCHAR(100),
    category_id BIGINT,
    brand VARCHAR(50),
    version INT,  -- 版本号，自增序列
    effective_start TIMESTAMP,  -- 生效时间
    effective_end TIMESTAMP,    -- 失效时间，默认NULL表示当前版本
    is_current BOOLEAN DEFAULT TRUE,  -- 是否当前有效版本
    PRIMARY KEY (item_id, version)
);

public class SCD2Sink extends RichSinkFunction<Row> {
    private Connection conn;
    @Override
    public void open(Configuration parameters) throws Exception {
        conn = DriverManager.getConnection("jdbc:postgresql://target-postgres:5432/recommendation_db", "admin", "password");
    }
    @Override
    public void invoke(Row row, Context context) throws Exception {
        long itemId = row.getFieldAs("item_id");
        String itemName = row.getFieldAs("item_name");
        // 查询当前最新版本
        PreparedStatement selectStmt = conn.prepareStatement(
            "SELECT version, effective_end FROM item_dim WHERE item_id = ? AND is_current = true"
        );
        selectStmt.setLong(1, itemId);
        ResultSet rs = selectStmt.executeQuery();

        try (Connection txnConn = conn.getConnection();
             PreparedStatement updateStmt = txnConn.prepareStatement(
                 "UPDATE item_dim SET is_current = false, effective_end = NOW() WHERE item_id = ? AND version = ?"
             );
             PreparedStatement insertStmt = txnConn.prepareStatement(
                 "INSERT INTO item_dim (item_id, item_name, category_id, brand, version, effective_start) " +
                 "VALUES (?, ?, ?, ?, ?, NOW())"
             ) {
            if (rs.next()) {
                long currentVersion = rs.getLong("version");
                // 关闭旧版本
                updateStmt.setLong(1, itemId);
                updateStmt.setLong(2, currentVersion);
                updateStmt.executeUpdate();
            }
            // 生成新版本号（实际需使用数据库序列）
            long newVersion = rs.next() ? rs.getLong("version") + 1 : 1;
            insertStmt.setLong(1, itemId);
            insertStmt.setString(2, itemName);
            // 填充其他字段...
            insertStmt.executeUpdate();
            txnConn.commit();
        }
    }
}

-- 获取用户点击时的商品有效版本
SELECT cl.*, id.item_name, id.brand
FROM target.click_log cl
JOIN target.item_dim id 
ON cl.item_id = id.item_id 
AND cl.click_time BETWEEN id.effective_start AND COALESCE(id.effective_end, NOW())

flink run -m yarn-cluster -c com.recommendation.PipelineJob /opt/jar/recommendation-pipeline.jar

import psycopg2
conn = psycopg2.connect("dbname=recommendation_db user=admin")
cur = conn.cursor()
cur.execute("INSERT INTO click_log (user_id, item_id, click_time) VALUES (1001, 2001, NOW())")
conn.commit()

UPDATE source.item_info SET category_id = 1002 WHERE item_id = 2001;  -- 类目调整

-- 变更前版本
SELECT * FROM item_dim WHERE item_id = 2001 AND version = 1;
-- 变更后新版本
SELECT * FROM item_dim WHERE item_id = 2001 AND version = 2 AND is_current = true;
验证结果：旧版本effective_end更新为变更时间，新版本effective_start为变更时间，属性值正确更新

SELECT COUNT(*) 
FROM click_log cl
LEFT JOIN item_dim id 
ON cl.item_id = id.item_id 
AND cl.click_time BETWEEN id.effective_start AND COALESCE(id.effective_end, NOW())
WHERE id.item_id IS NULL;  -- 应返回0条空关联数据

DELETE FROM item_dim WHERE effective_end < NOW() - INTERVAL '180 days';