Oracle与AWS S3数据湖并行Ingestion时主键同步最优方案咨询

针对流式JSON数据并行写入S3与Oracle且需主键同步的场景，以下是三种可落地的设计方案，各有适用场景：

方案1：先写Oracle获取主键，再同步至S3 #

核心思路是先将流式数据写入Oracle，利用其序列生成主键后，再将主键与原始数据合并写入S3。确保主键的唯一性由Oracle原生序列保障，同步逻辑简单直接。

实现步骤 #

1. 流式数据预处理：用PySpark清洗、转换原始JSON数据，保留业务字段。
2. 批量写入Oracle并返回主键：通过foreachBatch处理每个微批，在每个分区内调用Oracle的INSERT ... RETURNING语法，插入数据的同时获取生成的主键。
3. 关联主键并写入S3：将主键与原始业务字段合并为新的DataFrame，以Parquet/ORC等格式写入S3数据湖。

代码示例 #

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("StreamSyncOracleS3").getOrCreate()

def process_batch(df, batch_id):
    def insert_get_id(row):
        import cx_Oracle
        # 建立Oracle连接（建议用连接池优化性能）
        conn = cx_Oracle.connect("username/password@oracle-host:1521/service-name")
        cursor = conn.cursor()
        # 插入并返回主键，假设表为user_data，序列为user_seq
        cursor.execute(
            "INSERT INTO user_data (name, email) VALUES (:name, :email) RETURNING id INTO :id",
            name=row.name, email=row.email, id=cx_Oracle.NUMBER
        )
        generated_id = cursor.fetchone()[0]
        conn.commit()
        cursor.close()
        conn.close()
        return (generated_id, row.name, row.email)
    
    # 转换为RDD处理每个分区的记录
    rdd_with_id = df.rdd.map(insert_get_id)
    df_with_id = spark.createDataFrame(rdd_with_id, schema="id LONG, name STRING, email STRING")
    
    # 写入S3（按日期分区优化查询）
    df_with_id.write.mode("append").partitionBy("dt").parquet("s3://your-bucket/lake/user_data/")

# 读取流式JSON源（比如Kafka、文件流）
stream_df = spark.readStream.format("json").schema("name STRING, email STRING, dt DATE").load("stream-source-path")

# 启动流式处理
stream_df.writeStream.foreachBatch(process_batch).option("checkpointLocation", "s3://your-bucket/checkpoint/").start().awaitTermination()

优缺点 #

• 优点：主键完全由Oracle管控，一致性最高；无需额外状态管理逻辑。
• 缺点：增加了一次Oracle写入的网络往返，高吞吐量场景下可能有性能瓶颈；需处理Oracle写入失败后的重试与数据回滚。

方案2：预批量获取Oracle序列主键，双写S3与Oracle #

核心思路是提前从Oracle序列批量获取一段主键，在PySpark流式处理中为每条数据分配主键，同时写入S3与Oracle。减少单次写入的数据库交互，提升吞吐量。

实现步骤 #

1. 批量预取主键段：通过Oracle SQL批量获取序列值（比如一次取1000个），避免频繁调用序列。
2. 分布式主键分配：利用PySpark的状态管理（flatMapGroupsWithState）维护主键池，为每个微批的记录分配唯一主键。
3. 双写操作：将带主键的DataFrame同时写入Oracle（指定主键字段，不再依赖序列自动生成）与S3。

代码示例 #

from pyspark.sql.functions import lit
from pyspark.sql.streaming import GroupState

def fetch_seq_batch(seq_name, batch_size=1000):
    import cx_Oracle
    conn = cx_Oracle.connect("username/password@oracle-host:1521/service-name")
    cursor = conn.cursor()
    # 批量获取序列值
    cursor.execute(f"SELECT {seq_name}.nextval FROM dual CONNECT BY level <= {batch_size}")
    ids = [row[0] for row in cursor.fetchall()]
    conn.commit()
    cursor.close()
    conn.close()
    return ids

def assign_primary_key(state, rows):
    # 主键池为空时批量补充
    if not state.exists or len(state.get()) == 0:
        new_ids = fetch_seq_batch("user_seq")
        state.update(new_ids)
    id_pool = state.get()
    # 为每条记录分配主键
    assigned_rows = []
    for row in rows:
        if id_pool:
            assigned_rows.append((id_pool.pop(0), row.name, row.email, row.dt))
    # 更新状态中的主键池
    state.update(id_pool)
    return assigned_rows

# 读取流式数据
stream_df = spark.readStream.format("json").schema("name STRING, email STRING, dt DATE").load("stream-source-path")

# 用固定键分组，确保所有记录进入同一个状态处理器
keyed_df = stream_df.withColumn("group_key", lit(1))

# 分配主键
df_with_id = keyed_df.groupBy("group_key").flatMapGroupsWithState(
    outputMode="append",
    stateTimeout="noTimeout",
    func=assign_primary_key
).toDF("id", "name", "email", "dt")

def write_to_both(df, batch_id):
    # 写入Oracle（指定主键字段，避免序列重复生成）
    df.write.jdbc(
        url="jdbc:oracle:thin:@oracle-host:1521/service-name",
        table="user_data",
        mode="append",
        properties={"user": "username", "password": "password"}
    )
    # 写入S3
    df.write.mode("append").partitionBy("dt").parquet("s3://your-bucket/lake/user_data/")

# 启动流式处理
df_with_id.writeStream.foreachBatch(write_to_both).option("checkpointLocation", "s3://your-bucket/checkpoint/").start().awaitTermination()

优缺点 #

• 优点：减少数据库交互次数，提升流式处理吞吐量；主键分配逻辑在Spark侧完成，灵活可控。
• 缺点：需维护主键池的状态，任务失败时需确保未使用的主键不会丢失；需避免多个并行任务同时预取主键导致重复（可通过分布式锁或单节点预取解决）。

方案3：基于Oracle CDC同步主键至S3 #

核心思路是先将原始JSON数据写入S3，同时写入Oracle生成主键，再通过Oracle的CDC（变更数据捕获）功能捕获插入事件，将主键同步回S3的对应记录。适合已有CDC基础设施的场景。

实现步骤 #

1. 双写原始数据：将未带主键的原始JSON数据同时写入S3与Oracle（Oracle侧生成主键）。
2. CDC捕获主键：启用Oracle CDC（如GoldenGate、Oracle CDC for Kafka），捕获user_data表的插入操作，提取主键与业务唯一标识（如原始数据中的uuid）。
3. 更新S3数据：利用Delta Lake的ACID特性或S3分区覆盖，将主键关联到S3中的对应记录。

关键注意事项 #

• 原始数据必须包含业务唯一标识（如UUID），否则无法匹配CDC事件与S3中的记录。
• 若使用Delta Lake，可通过MERGE语句高效更新主键字段；若用普通Parquet，需按分区覆盖对应文件。

优缺点 #

• 优点：不修改原有双写流程，对业务侵入小；适合已有CDC链路的架构。
• 缺点：增加了CDC链路的复杂度；存在主键同步的延迟，无法做到实时一致。

通用注意事项 #

1. 幂等性保障：通过Spark的checkpointLocation跟踪处理进度，避免重复处理导致主键重复或数据重复写入。
2. 异常处理：针对Oracle写入失败的情况，需实现重试机制；若S3写入失败，需回滚Oracle的插入操作（可通过数据库事务实现）。
3. 性能优化：使用Oracle连接池减少连接开销；批量写入Oracle与S3，提升处理效率；对S3数据按时间/业务字段分区，优化查询性能。

内容的提问来源于stack exchange，提问作者Vrishank