如何用Python结合JDBC驱动迭代Oracle大数据集并写入新表

我之前处理过类似的超大规模Oracle数据迁移场景，你的需求完全可以通过流式分批读取+逐行处理+批量写入的方案来实现，既不会把所有数据加载到内存，又能保证处理效率。下面是具体的实现细节：

避免全量加载的关键是让数据库端分批返回结果集，而不是一次性把所有数据拉到Python内存中。结合JayDeBeApi和Oracle JDBC驱动的特性，我们可以通过设置fetchsize来控制每次从数据库获取的行数，然后逐行处理，最后攒够一定数量的处理结果后批量插入新表，减少数据库交互开销。

1. 数据库连接配置 #

首先确保你的Python环境已经安装了JayDeBeApi，并且有Oracle的JDBC驱动（比如ojdbc8.jar）。连接时要注意指定正确的JDBC URL、用户名和密码。

import jaydebeapi
import re
from typing import List, Tuple

# 数据库连接参数
jdbc_driver = "oracle.jdbc.driver.OracleDriver"
jdbc_url = "jdbc:oracle:thin:@//your-oracle-host:1521/your-service-name"
db_user = "your_username"
db_password = "your_password"
jdbc_jar_path = "/path/to/ojdbc8.jar"

# 建立连接
conn = jaydebeapi.connect(
    jdbc_driver,
    jdbc_url,
    [db_user, db_password],
    jdbc_jar_path,
)

2. 设置分批读取（关键！） #

Oracle JDBC驱动默认的fetchsize很小（通常是10），而且如果不设置，有些情况下会一次性拉取全量数据。所以我们需要给查询游标设置合适的fetchsize，比如1000或5000（根据你的内存情况调整），这样每次从数据库只获取指定行数的数据到客户端内存。

# 创建查询游标，并设置fetchsize
select_cursor = conn.cursor()
# 设置每次从数据库获取1000行，避免全量加载
select_cursor.setfetchsize(1000)

# 执行查询，尽量不要加ORDER BY（除非业务强制要求），避免数据库端排序占用额外资源
select_sql = "SELECT varchar_column, blob_column FROM original_table"
select_cursor.execute(select_sql)

3. 逐行处理数据 #

现在可以迭代游标来逐行获取数据（实际上是分批从数据库拉取，然后逐行返回），对每一行执行正则处理和Blob操作：

# 预编译正则表达式，提升处理效率
pattern = re.compile(r'old_pattern')

# 定义处理函数
def process_row(varchar_val: str, blob_val) -> Tuple[str, bytes]:
    # 对varchar列执行正则替换操作
    processed_varchar = pattern.sub(r'new_pattern', varchar_val)
    # 处理Blob列：如果是非空Blob，读取二进制内容；空值则返回None
    if blob_val is not None:
        # 对于超大Blob，建议用getBinaryStream()流式读取，避免一次性加载到内存
        processed_blob = blob_val.getBytes(1, blob_val.length())
    else:
        processed_blob = None
    return processed_varchar, processed_blob

# 批量插入的批次大小，比如每1000条提交一次，平衡性能和事务风险
batch_size = 1000
batch_data = []

# 逐行读取并处理
for row in select_cursor:
    varchar_col, blob_col = row
    processed_varchar, processed_blob = process_row(varchar_col, blob_col)
    # 将处理后的数据加入批量列表
    batch_data.append((processed_varchar, processed_blob))
    
    # 当批量列表达到设定大小，执行批量插入
    if len(batch_data) >= batch_size:
        insert_cursor = conn.cursor()
        insert_sql = "INSERT INTO new_table (processed_varchar, processed_blob) VALUES (?, ?)"
        insert_cursor.executemany(insert_sql, batch_data)
        # 提交事务
        conn.commit()
        # 清空批量列表
        batch_data = []
        insert_cursor.close()

4. 处理剩余数据 #

循环结束后，可能还有剩余的不足一个批次的数据，需要单独处理：

# 处理最后一批未提交的数据
if batch_data:
    insert_cursor = conn.cursor()
    insert_sql = "INSERT INTO new_table (processed_varchar, processed_blob) VALUES (?, ?)"
    insert_cursor.executemany(insert_sql, batch_data)
    conn.commit()
    insert_cursor.close()

5. 资源清理 #

最后记得关闭游标和连接，避免资源泄漏：

select_cursor.close()
conn.close()

• fetchsize的调整：如果内存充足，可以适当调大fetchsize（比如5000）减少数据库交互次数；内存紧张则调小到500或1000。
• Blob处理优化：对于超大Blob，不要用getBytes()一次性读取，改用getBinaryStream()流式处理，避免内存过载。
• 正则效率优化：一定要预编译正则表达式，避免每次处理都重新编译，浪费CPU资源。
• 事务控制：批量提交的批次不要过大，否则事务日志会占用过多磁盘，出错时回滚代价也更高。
• 避免不必要的排序：查询时尽量不要加ORDER BY，否则数据库会先全量排序再返回，反而占用更多数据库资源。

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