Python执行PL/SQL代码批量操作Oracle数据库无数据变更问题排查

嗨Jegor，我来帮你一步步排查这个问题，从你贴的代码来看，有几个关键错误导致数据没有被正确插入/更新，咱们逐个梳理：

1. 致命的缩进错误（批量插入逻辑完全未执行） #

看你Python代码里的批量处理部分，缩进明显错误：

if len(data) % batch_size == 0:
cursor.executemany(sql_insert, data)
    data = []

cursor.executemany这一行没有缩进，导致这个批量插入的逻辑永远不会触发！只有当最后剩余的data不为空时，才会执行一次插入操作；如果你的数据量刚好是batch_size的整数倍，最后data是空的，连这最后一次执行都不会触发，自然看不到任何数据变更。

2. 错误跳过第一行有效数据 #

你明确说明S3里的文件没有表头，但代码里却执行了：

headings = next(csv_reader)

这会直接跳过第一行有效数据，如果你测试的文件数据量很少，可能直接导致没有数据被处理。

3. PL/SQL的UPDATE语句缺少WHERE条件（逻辑错误） #

你的PL/SQL代码里，UPDATE部分没有指定过滤条件：

Update %(oracle_table)s
Set STRING_COLUMN = :STRING_COLUMN;

这意味着只要counter>0（即表中存在任意一条ID匹配的记录），就会更新整个表的所有行的STRING_COLUMN，而不是当前要处理的那条ID对应的行！正确的写法必须加上WHERE条件：

Update %(oracle_table)s
Set STRING_COLUMN = :STRING_COLUMN
Where ID = :ID;

另外，用PL/SQL手动判断存在性其实不如Oracle原生的MERGE语句高效且安全，推荐直接替换成MERGE语句，避免手动逻辑的错误：

MERGE INTO %(oracle_table)s t
USING (SELECT :ID AS ID, :STRING_COLUMN AS STRING_COLUMN FROM DUAL) s
ON (t.ID = s.ID)
WHEN MATCHED THEN
  UPDATE SET t.STRING_COLUMN = s.STRING_COLUMN
WHEN NOT MATCHED THEN
  INSERT (ID, STRING_COLUMN) VALUES (s.ID, s.STRING_COLUMN)

这个语句能自动完成“存在则更新，不存在则插入”的逻辑，代码更简洁也更可靠。

4. 提交时机不完整 #

当前代码里，只有最后处理剩余数据时才调用connection.commit()，而前面每批次的executemany执行后都没有提交。虽然Oracle默认关闭自动提交，但最好每批次执行后都提交，避免事务过大导致性能问题或回滚风险。

修正后的完整代码示例 #

Python代码修正版 #

....
# SQL Statement
sql_insert = config.get('SQL_STATEMENT', 'sql_insert')

# Get File from S3 Bucket
obj = s3.Object(CDH_S3_Bucket, CDH_Path + '/' + s3_filename)
body = obj.get()['Body'].read().decode('utf-8').splitlines()

cursor.setinputsizes(None, 25)

batch_size = 10000
data = []  # 初始化data列表，避免未定义报错

# ----------------------------------------------------------
print("Import Data...")
print("------------------------------")
csv_reader = csv.reader(body, delimiter=',')
# 没有表头，注释掉跳过行的代码
# headings = next(csv_reader)

for line in csv_reader:
    data.append(line)
    if len(data) % batch_size == 0:
        cursor.executemany(sql_insert, data)
        connection.commit()  # 每批次执行后提交
        data = []

if data:
    cursor.executemany(sql_insert, data)
    connection.commit()
#------------------------------------------------------------
# Close Cursor and Connection
cursor.close()
connection.close()
....

.ini文件修正版（使用MERGE替代PL/SQL） #

[SQL_STATEMENT]
oracle_table = TEST_TABLE
sql_insert = """
MERGE INTO %(oracle_table)s t
USING (SELECT :ID AS ID, :STRING_COLUMN AS STRING_COLUMN FROM DUAL) s
ON (t.ID = s.ID)
WHEN MATCHED THEN
  UPDATE SET t.STRING_COLUMN = s.STRING_COLUMN
WHEN NOT MATCHED THEN
  INSERT (ID, STRING_COLUMN) VALUES (s.ID, s.STRING_COLUMN)
"""

如果你坚持使用原来的PL/SQL逻辑，记得修正UPDATE的WHERE条件：

[SQL_STATEMENT]
oracle_table = TEST_TABLE
sql_insert = """Declare
       counter number;
    Begin
       Select
          Count(*) into counter
       From %(oracle_table)s
       Where ID  = :ID;

       If counter = 0 Then
          insert into %(oracle_table)s (ID,STRING_COLUMN) 
          values (:ID,:STRING_COLUMN);
       Else
          Update %(oracle_table)s
          Set STRING_COLUMN = :STRING_COLUMN
          Where ID = :ID;  -- 新增WHERE条件
       End If;
    End;"""

备注：内容来源于stack exchange，提问作者Jegor Wieler