如何用C# Azure Function将Azure Data Lake中的嵌套XML转为CSV？

前置准备 #

1. 安装必要的NuGet包：

   • Azure.Storage.Files.DataLake：用于访问Azure Data Lake Storage
   • System.Xml.Linq：用于解析XML
   • CsvHelper（可选）：简化CSV生成（若不想依赖第三方包，可使用手动拼接CSV的方式）

2. 配置ADLS连接字符串：
   在Azure Function应用的设置中添加名为ADLS_CONNECTION_STRING的应用设置，值为你的ADLS账户连接字符串。

实现步骤 #

1. 读取ADLS中的XML文件 #

使用DataLakeServiceClient连接到ADLS，获取文件客户端并读取文件内容到流中。

2. 解析XML数据 #

提取XML中的根节点属性、level1节点数据，以及每个level_record节点的内容，将这些数据映射到一个实体类中（每个level_record对应CSV的一行，父节点数据会重复填充）。

3. 生成CSV内容 #

可以选择使用CsvHelper自动生成CSV，或者手动拼接CSV字符串（需处理特殊字符转义）。

4. 将CSV保存回ADLS #

将生成的CSV内容写入ADLS的指定路径。

完整代码示例（HTTP触发器） #

using System;
using System.Collections.Generic;
using System.Globalization;
using System.IO;
using System.Text;
using System.Threading.Tasks;
using Azure.Storage.Files.DataLake;
using CsvHelper;
using CsvHelper.Configuration.Attributes;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.Http;
using Microsoft.Extensions.Logging;
using System.Xml.Linq;

public class XmlToCsvFunction
{
    [FunctionName("XmlToCsv")]
    public async Task<IActionResult> Run(
        [HttpTrigger(AuthorizationLevel.Function, "get", "post", Route = null)] HttpRequest req,
        ILogger log)
    {
        log.LogInformation("开始处理XML转CSV任务");

        // 从应用设置获取ADLS连接字符串
        string connectionString = Environment.GetEnvironmentVariable("ADLS_CONNECTION_STRING");
        if (string.IsNullOrEmpty(connectionString))
        {
            return new BadRequestObjectResult("未配置ADLS_CONNECTION_STRING应用设置");
        }

        // 获取请求中的输入输出文件路径
        string inputFilePath = req.Query["inputFilePath"];
        if (string.IsNullOrEmpty(inputFilePath))
        {
            return new BadRequestObjectResult("请提供inputFilePath查询参数，指定ADLS中的XML文件路径");
        }

        string outputFilePath = req.Query["outputFilePath"] ?? "output/converted.csv";

        try
        {
            // 连接ADLS并读取XML文件
            var dataLakeClient = new DataLakeServiceClient(connectionString);
            var fileSystemName = Path.GetDirectoryName(inputFilePath).TrimStart('/');
            var fileName = Path.GetFileName(inputFilePath);
            var fileSystemClient = dataLakeClient.GetFileSystemClient(fileSystemName);
            var xmlFileClient = fileSystemClient.GetFileClient(fileName);

            using var xmlStream = new MemoryStream();
            await xmlFileClient.ReadToAsync(xmlStream);
            xmlStream.Position = 0;

            // 解析XML
            XDocument xmlDoc = XDocument.Load(xmlStream);
            var root = xmlDoc.Root;
            if (root == null)
            {
                return new BadRequestObjectResult("XML格式错误：未找到根节点");
            }

            // 提取根节点属性
            string rootId = root.Attribute("id")?.Value;
            string createdDate = root.Attribute("created_date")?.Value;
            string asOfDate = root.Attribute("asof_date")?.Value;

            // 提取level1节点数据
            var level1 = root.Element("level1");
            if (level1 == null)
            {
                return new BadRequestObjectResult("XML格式错误：未找到level1节点");
            }

            string data1 = level1.Element("data1")?.Value;
            string data2 = level1.Element("data2")?.Value;

            // 提取所有level_record节点
            var level2 = level1.Element("level2");
            if (level2 == null)
            {
                return new BadRequestObjectResult("XML格式错误：未找到level2节点");
            }

            var levelRecords = level2.Elements("level_record");
            if (!levelRecords.Any())
            {
                return new OkObjectResult("未找到任何level_record节点");
            }

            // 映射到实体类
            List<Record> csvRecords = new List<Record>();
            foreach (var record in levelRecords)
            {
                csvRecords.Add(new Record
                {
                    Id = rootId,
                    CreatedDate = createdDate,
                    AsOfDate = asOfDate,
                    Data1 = data1,
                    Data2 = data2,
                    GroupId = record.Element("groupid")?.Value,
                    GroupName = record.Element("groupname")?.Value,
                    GroupDate = record.Element("groupdate")?.Value,
                    GroupValue = record.Element("groupvalue")?.Value,
                    GroupKey = record.Element("groupkey")?.Value
                });
            }

            // 生成CSV内容（使用CsvHelper）
            string csvContent;
            using (var stringWriter = new StringWriter())
            {
                using (var csvWriter = new CsvWriter(stringWriter, CultureInfo.InvariantCulture))
                {
                    csvWriter.WriteRecords(csvRecords);
                }
                csvContent = stringWriter.ToString();
            }

            // 保存CSV到ADLS
            var outputFileSystemClient = dataLakeClient.GetFileSystemClient(Path.GetDirectoryName(outputFilePath).TrimStart('/'));
            var csvFileClient = outputFileSystemClient.GetFileClient(Path.GetFileName(outputFilePath));
            using var csvStream = new MemoryStream(Encoding.UTF8.GetBytes(csvContent));
            await csvFileClient.UploadAsync(csvStream, overwrite: true);

            return new OkObjectResult($"CSV文件已成功保存到：{outputFilePath}");
        }
        catch (Exception ex)
        {
            log.LogError(ex, "XML转CSV过程中发生错误");
            return new StatusCodeResult(StatusCodes.Status500InternalServerError);
        }
    }

    // CSV实体类，对应CSV列名
    private class Record
    {
        [Name("id")]
        public string Id { get; set; }

        [Name("created_date")]
        public string CreatedDate { get; set; }

        [Name("asof_date")]
        public string AsOfDate { get; set; }

        [Name("data1")]
        public string Data1 { get; set; }

        [Name("data2")]
        public string Data2 { get; set; }

        [Name("groupid")]
        public string GroupId { get; set; }

        [Name("groupname")]
        public string GroupName { get; set; }

        [Name("groupdate")]
        public string GroupDate { get; set; }

        [Name("groupvalue")]
        public string GroupValue { get; set; }

        [Name("groupkey")]
        public string GroupKey { get; set; }
    }
}

手动生成CSV的替代方案 #

如果不想使用CsvHelper，可以用以下代码替换CSV生成部分：

// 手动生成CSV内容
StringBuilder csvBuilder = new StringBuilder();
// 写入表头
csvBuilder.AppendLine("id,created_date,asof_date,data1,data2,groupid,groupname,groupdate,groupvalue,groupkey");

foreach (var rec in csvRecords)
{
    // 转义特殊字符（逗号、引号、换行）
    csvBuilder.AppendLine($"{EscapeCsvField(rec.Id)},{EscapeCsvField(rec.CreatedDate)},{EscapeCsvField(rec.AsOfDate)},{EscapeCsvField(rec.Data1)},{EscapeCsvField(rec.Data2)},{EscapeCsvField(rec.GroupId)},{EscapeCsvField(rec.GroupName)},{EscapeCsvField(rec.GroupDate)},{EscapeCsvField(rec.GroupValue)},{EscapeCsvField(rec.GroupKey)}");
}

string csvContent = csvBuilder.ToString();

// CSV字段转义辅助方法
private string EscapeCsvField(string value)
{
    if (string.IsNullOrEmpty(value))
        return string.Empty;

    // 若包含特殊字符，用双引号包裹并转义内部的双引号
    if (value.Contains(',') || value.Contains('"') || value.Contains('\n') || value.Contains('\r'))
    {
        return $"\"{value.Replace("\"", "\"\"")}\"";
    }

    return value;
}

注意事项 #

• 确保XML文件格式正确：你的示例XML中存在拼写错误（</groudate>应为</groupdate>），需修正后再处理。
• 可以将HTTP触发器替换为Data Lake Storage触发器，实现文件上传后自动转换。
• 处理大文件时，建议使用流式处理而非一次性加载到内存，避免内存溢出。

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