在Azure Data Factory中将嵌套XML键值数组转换为单行

根据你的需求，需要将XML中每行发票数据（包含line属性、product属性及动态的pos_record/data键值对）转换为SQL表的单行记录，以下是几种常用的实现方式：

1. SQL Server 内置XML直接处理 #

如果XML已存储在SQL Server表中，可通过XQuery+动态SQL实现动态列转换：

步骤与代码示例 #

假设XML存储在InvoiceData表的XmlContent列：

-- 1. 提取所有唯一的data key，生成动态列名（带product_pos_前缀）
DECLARE @Columns NVARCHAR(MAX)
SELECT @Columns = STRING_AGG(DISTINCT QUOTENAME('product_pos_' + c.value('@key', 'NVARCHAR(100)')), ', ')
FROM InvoiceData
CROSS APPLY XmlContent.nodes('/invoice/lines/line/product/pos_record/data') AS t(c)

-- 2. 动态生成行转列查询
DECLARE @Query NVARCHAR(MAX) = N'
SELECT 
    ROW_NUMBER() OVER(ORDER BY (SELECT NULL)) AS id,
    line.value(''@quantity'', ''DECIMAL(10,3)'') AS qty,
    line.value(''@unit'', ''NVARCHAR(10)'') AS unit,
    product.value(''@name'', ''NVARCHAR(255)'') AS name,
    product.value(''@product_type'', ''NVARCHAR(50)'') AS product_type,
    ' + @Columns + '
FROM InvoiceData
CROSS APPLY XmlContent.nodes(''/invoice/lines/line'') AS t(line)
CROSS APPLY line.nodes(''product'') AS p(product)
CROSS APPLY product.nodes(''pos_record/data'') AS d(data)
PIVOT (
    MAX(data.value(''@value'', ''NVARCHAR(255)''))
    FOR data.value(''@key'', ''NVARCHAR(100)'') IN (' + REPLACE(@Columns, 'product_pos_', '') + ')
) AS PivotTable'

EXEC sp_executesql @Query

2. SSIS 数据流实现 #

若用SSIS做ETL，可通过以下步骤实现动态转换：

• XML源组件：读取XML，设置XPath为/invoice/lines/line，提取line的quantity/unit、product的name/product_type，以及pos_record下data的key/value。
• 脚本组件（动态列处理）：由于SSIS透视转换默认需固定列，用C#脚本实现动态列生成与赋值：

using System;
using System.Data;
using System.Xml.Linq;
using Microsoft.SqlServer.Dts.Pipeline.Wrapper;

[Microsoft.SqlServer.Dts.Pipeline.SSISScriptComponentEntryPointAttribute]
public class ScriptMain : UserComponent
{
    private string[] _dynamicColumns;

    public override void PreExecute()
    {
        base.PreExecute();
        // 从XML中提取所有唯一的data key，生成带前缀的列名
        var xml = XDocument.Parse(Variables.InvoiceXml);
        var keys = xml.Descendants("data").Select(d => (string)d.Attribute("key")).Distinct();
        _dynamicColumns = keys.Select(k => "product_pos_" + k).ToArray();
        
        // 动态添加输出列
        foreach (var col in _dynamicColumns)
        {
            Output0Buffer.AddColumn(col, DataType.DT_WSTR, 255);
        }
    }

    public override void Input0_ProcessInputRow(Input0Buffer Row)
    {
        var lineXml = XElement.Parse(Row.LineXml);
        // 填充固定列
        Row.id = Row.Id;
        Row.qty = (decimal)lineXml.Attribute("quantity");
        Row.unit = (string)lineXml.Attribute("unit");
        var product = lineXml.Element("product");
        Row.name = (string)product.Attribute("name");
        Row.product_type = (string)product.Attribute("product_type");
        
        // 填充动态列
        var posData = product.Element("pos_record").Elements("data")
            .ToDictionary(d => "product_pos_" + (string)d.Attribute("key"), d => (string)d.Attribute("value"));
        
        foreach (var col in _dynamicColumns)
        {
            Output0Buffer.SetString(col, posData.TryGetValue(col, out var val) ? val : string.Empty);
        }
    }
}

3. Python 脚本自定义处理 #

适合灵活的自定义数据流场景，用xml.etree解析+pandas透视：

import xml.etree.ElementTree as ET
import pandas as pd

# 解析XML文件
tree = ET.parse('invoice.xml')
root = tree.getroot()

# 提取每行发票数据
rows = []
for line in root.findall('.//line'):
    # 基础字段
    row = {
        'qty': float(line.get('quantity')),
        'unit': line.get('unit'),
        'name': line.find('product').get('name'),
        'product_type': line.find('product').get('product_type')
    }
    # 提取pos_record动态键值对
    pos_record = line.find('.//pos_record')
    if pos_record:
        for data in pos_record.findall('data'):
            key = f'product_pos_{data.get("key")}'
            row[key] = data.get('value')
    rows.append(row)

# 转换为DataFrame，自动适配动态列
df = pd.DataFrame(rows)
df['id'] = df.index + 1  # 添加自增id

# 输出到SQL表（需安装sqlalchemy）
# from sqlalchemy import create_engine
# engine = create_engine('mssql+pyodbc://server/database?driver=ODBC+Driver+17+for+SQL+Server')
# df.to_sql('InvoiceLines', engine, if_exists='replace', index=False)

print(df)

核心注意事项 #

• 动态列兼容性：所有方案都会自动收集所有出现过的data key，即使部分行无对应key也会填充空值，适配未来key的变化。
• 数据类型转换：根据SQL表需求，需将数值型字段（如quantity、tax）转换为对应类型，避免存储为字符串。
• 多发票处理：确保提取逻辑会遍历所有发票的line节点，不会混淆不同发票的行数据。

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