Python实现Oracle导出CSV转换为指定XML格式的高效方案

当前需求为将Oracle数据库导出的CSV格式数据，转换为按Id字段分组嵌套的指定结构XML，单文件数据量15~20万条，属于每日定时批处理任务。

输入示例 #

CSV输入结构：

Id,SubID,Rank,Size
1,123,1,0.1
1,234,2,0.2
2,456,1,0.1
2,123,2,0.2

期望输出格式 #

目标XML结构：

<AA_ITEMS>
<Id ID="1">
    <SubId ID="123">
        <Rank>1</Rank>
        <Size>0.1</Size>
    </SubId>
    <SubId ID="234">
        <Rank>2</Rank>
        <Size>0.2</Size>
    </SubId>
</Id>
<Id ID="2">
    <SubId ID="456">
        <Rank>1</Rank>
        <Size>0.1</Size>
    </SubId>
    <SubId ID="123">
        <Rank>2</Rank>
        <Size>0.2</Size>
    </SubId>
</Id>
</AA_ITEMS>

核心性能原则：20万行数据量级下，禁止使用全量加载内存的DOM类XML生成方式，优先选择流式读写方案，内存占用可控制在百MB级别，单任务处理耗时通常在10秒内。

• 方案1：Python流式处理（部署成本最低）
  不需要额外搭建重型服务，仅用Python标准库即可实现，全程逐行读写不缓存全量数据。如果Oracle导出时已经按Id排序，可以跳过预排序步骤进一步提速。
  参考实现代码：

  import csv
  
  # 可选：XML特殊字符转义，字段存在特殊字符时启用
  def xml_escape(s):
      return s.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;").replace('"', "&quot;")
  
  def csv_to_xml(csv_path, xml_path):
      with open(csv_path, 'r', encoding='utf-8') as f_csv, open(xml_path, 'w', encoding='utf-8') as f_xml:
          reader = csv.DictReader(f_csv)
          f_xml.write('<AA_ITEMS>\n')
          current_id = None
          for row in reader:
              row_id = xml_escape(row['Id'])
              # 匹配到新的Id分组时，先闭合上一个Id节点
              if row_id != current_id:
                  if current_id is not None:
                      f_xml.write('</Id>\n')
                  f_xml.write(f'<Id ID="{row_id}">\n')
                  current_id = row_id
              # 逐行写入SubId子节点
              sub_id = xml_escape(row["SubID"])
              rank = xml_escape(row["Rank"])
              size = xml_escape(row["Size"])
              f_xml.write(f'    <SubId ID="{sub_id}">\n')
              f_xml.write(f'        <Rank>{rank}</Rank>\n')
              f_xml.write(f'        <Size>{size}</Size>\n')
              f_xml.write(f'    </SubId>\n')
          # 补全最后一个节点和根节点闭合标签
          if current_id is not None:
              f_xml.write('</Id>\n')
          f_xml.write('</AA_ITEMS>')
  
  if __name__ == '__main__':
      csv_to_xml('input.csv', 'output.xml')
  

  如果原始CSV未按Id排序，提前加一步按Id排序的逻辑即可，20万行数据排序内存占用极低。

• 方案2：Java StAX流式处理（性能最优，适配Java技术栈）
  若现有批处理体系基于Java搭建，直接使用JDK自带的StAX事件驱动写入API即可，不要用DOM4J、JDOM这类内存树模型的库，StAX方案处理速度比Python快30%左右，全程内存占用不超过100MB，核心逻辑和Python方案一致：逐行读CSV、按Id分组判断、流式写入XML节点，无需全量加载数据。

• 方案3：ETL工具可视化配置（适配低代码运维场景）
  若日常运维基于ETL工具实现，直接拖拽组件配置流程即可：

  1. 用CSV文件输入组件读取源文件，配置按Id字段分组
  2. 用XML输出组件配置嵌套结构：根节点为AA_ITEMS，第一层子节点Id取Id字段作为ID属性，第二层子节点SubId取SubID字段作为ID属性，下挂Rank、Size两个文本节点
  3. 开启组件流式写入开关，关闭全量缓存选项，即可稳定处理20万行级数据。

• 不要用全量读取数据转对象/字典再整体拼接XML的写法，嵌套结构复杂时极易触发内存溢出，处理速度也远低于流式方案
• 字段包含<、>、&、"等XML特殊字符时，必须做转义处理，否则会生成格式非法的XML文件
• 日批任务可增加轻量校验：写入完成后统计SubId节点数量和CSV原始行数是否一致，避免漏写数据

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