如何用Python将Oracle表中嵌套JSON数据转换为CSV文件？

核心思路 #

针对嵌套JSON中的对象直接扁平化提取字段，数组类型则做行拆分（每条数组元素对应原记录的一行），同时保留原表的idAsString和created_date字段。

示例代码实现 #

假设data_extract的JSON结构如下：

{
  "user_info": {
    "name": "Alice",
    "age": 30
  },
  "Technology": [
    {"name": "Python", "version": "3.10"},
    {"name": "Oracle", "version": "19c"}
  ]
}

步骤1：读取Oracle数据并解析JSON #

使用cx_Oracle连接Oracle，结合json模块解析嵌套结构：

import cx_Oracle
import json
import csv

# 初始化Oracle连接
conn = cx_Oracle.connect("username/password@host:port/service_name")
cursor = conn.cursor()

# 查询表数据
cursor.execute("SELECT idAsString, data_extract, created_date FROM Table1")

# 准备CSV写入器
with open("output.csv", "w", newline="", encoding="utf-8") as csvfile:
    # 定义CSV表头：原表字段 + 扁平化后的JSON字段 + 数组元素字段
    fieldnames = ["idAsString", "created_date", "user_name", "user_age", "tech_name", "tech_version"]
    writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
    writer.writeheader()

    # 处理每条记录
    for row in cursor:
        id_str, data_json_str, created_date = row
        data = json.loads(data_json_str)
        
        # 扁平化嵌套对象
        user_info = data.get("user_info", {})
        user_name = user_info.get("name")
        user_age = user_info.get("age")
        
        # 处理Technology数组，拆分每行
        tech_list = data.get("Technology", [])
        if not tech_list:
            # 数组为空时写入空值行
            writer.writerow({
                "idAsString": id_str,
                "created_date": created_date,
                "user_name": user_name,
                "user_age": user_age,
                "tech_name": None,
                "tech_version": None
            })
        else:
            # 数组非空时，每个元素对应一行
            for tech in tech_list:
                writer.writerow({
                    "idAsString": id_str,
                    "created_date": created_date,
                    "user_name": user_name,
                    "user_age": user_age,
                    "tech_name": tech.get("name"),
                    "tech_version": tech.get("version")
                })

# 关闭连接
cursor.close()
conn.close()

常见问题修正 #

• 数组未拆分：检查是否遍历了数组元素，确保每个元素单独生成一行，而非将整个数组作为一个字段写入。
• 嵌套对象提取失败：使用dict.get()方法避免KeyError，若JSON结构有变动，需调整字段映射路径。
• 日期格式问题：Oracle的created_date需转换为Python可识别的格式，可在查询时用TO_CHAR(created_date, 'YYYY-MM-DD HH24:MI:SS')转为字符串。
• 写入Oracle表替代CSV：将CSV写入部分替换为Oracle插入语句，使用批量插入提升效率：

  # 示例：批量插入到新表Table1_flat
  insert_sql = """
  INSERT INTO Table1_flat (idAsString, created_date, user_name, user_age, tech_name, tech_version)
  VALUES (:1, :2, :3, :4, :5, :6)
  """
  # 收集所有待插入数据
  insert_data = []
  # ...（前面的解析逻辑）
  for tech in tech_list:
      insert_data.append((id_str, created_date, user_name, user_age, tech.get("name"), tech.get("version")))
  # 批量执行
  cursor.executemany(insert_sql, insert_data)
  conn.commit()
  

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