基于SSIS将Oracle数据导出为多份JSON并生成清单的技术需求

借助SSIS从Oracle数据源读取数据，转换为JSON格式写入平面文件；因数据量庞大，需拆分至多个JSON文件；处理完成后生成包含总处理行数的清单文件。目前已配置数据流任务并选定Oracle源，需要对应的C#逻辑完成数据转JSON及文件写入操作。

SSIS脚本组件C#实现逻辑 #

以下代码实现从Oracle源读取数据、转换为JSON格式，并按指定行数拆分写入多个文件：

using System;
using System.IO;
using System.Text;
using Microsoft.SqlServer.Dts.Pipeline.Wrapper;
using Microsoft.SqlServer.Dts.Runtime.Wrapper;
using System.Collections.Generic;
using System.Reflection;

[Microsoft.SqlServer.Dts.Pipeline.SSISScriptComponentEntryPointAttribute]
public class ScriptMain : UserComponent
{
    private StreamWriter fileWriter;
    private int fileCounter = 1;
    private int rowCounter = 0;
    private int rowsPerFile = 1000;  // 配置每个文件的行数

    public override void PreExecute()
    {
        base.PreExecute();
        OpenNewFile();
    }

    public override void PostExecute()
    {
        base.PostExecute();
        fileWriter.Close();
    }

    public override void Input0_ProcessInputRow(Input0Buffer Row)
    {
        var rowData = new Dictionary<string, object>();

        foreach (IDTSInputColumn100 column in this.ComponentMetaData.InputCollection[0].InputColumnCollection)
        {
            var columnName = column.Name;
            var columnValue = Row.GetType().GetProperty(columnName).GetValue(Row, null);
            rowData.Add(columnName, columnValue);
        }

        string json = SerializeToJson(rowData);
        fileWriter.WriteLine(json);
        rowCounter++;

        if (rowCounter >= rowsPerFile)
        {
            fileWriter.Close();
            fileCounter++;
            rowCounter = 0;
            OpenNewFile();
        }
    }

    private void OpenNewFile()
    {
        string filePath = $"C:\\PathToYourFolder\\output_{fileCounter}.json";  // 修改为你的目标文件路径
        fileWriter = new StreamWriter(filePath, false, Encoding.UTF8);
    }

    private string SerializeToJson(Dictionary<string, object> data)
    {
        StringBuilder json = new StringBuilder();
        json.Append("{");

        foreach (var kvp in data)
        {
            json.Append($"\"{kvp.Key}\": \"{kvp.Value}\",");
        }

        if (json.Length > 1)
        {
            json.Length--;  // 移除末尾多余的逗号
        }

        json.Append("}");
        return json.ToString();
    }
}

清单文件生成C#代码 #

处理完成后，运行以下代码生成包含总处理行数和生成文件列表的XML清单：

using (XmlWriter writer = XmlWriter.Create(manifestFilePath, new XmlWriterSettings { Indent = true }))
{
    writer.WriteStartDocument();
    writer.WriteStartElement("manifest");

    writer.WriteElementString("processType", "DataIngest");
    writer.WriteElementString("totalRecordCount", totalRecordCount.ToString());

    writer.WriteStartElement("files");

    foreach (string fileName in fileNames)
    {
        writer.WriteStartElement("file");
        writer.WriteElementString("name", fileName);
        writer.WriteEndElement();
    }

    writer.WriteEndElement(); // 结束files节点
    writer.WriteEndElement(); // 结束manifest节点
    writer.WriteEndDocument();
}

清单文件格式示例 #

生成的清单文件为XML格式，示例如下：

<?xml version="1.0" ?>
<manifest>
 <processType>DataIngest</processType>
 <totalRecordCount>13000</totalRecordCount>
 <files>
  <file>
   <name>/AccountDataIngestion-20240624092345_123_1.json</name>
  </file>
  <file>
   <name>/AccountDataIngestion_20240624092345_234_2.json</name> 
  </file>
 </files>
</manifest>

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