Parquet.NET读取Parquet文件内存占用过高问题求助

使用Parquet.NET库时，以下代码因一次性读取整个Parquet文件到Table实例导致内存占用过高：

using (ParquetReader reader = await ParquetReader.CreateAsync(filePath, null))
{
    string tableName = GetTableName(filePath);
    Table parquetTable = await reader.ReadAsTableAsync(); // 内存占用过高的核心原因
    DataTable dataTable = new DataTable();

    string sql = $"CREATE TABLE {tableName} (";
    foreach(Field field in parquetTable.Schema.Fields)
    {
        DataField? ptField = field as DataField;
        string columnName = ptField.Name;
        Type columnType = ptField.ClrType;
        dataTable.Columns.Add(columnName, columnType);
        sql += $"[{columnName}] {GetSqlDataType(columnType, field)},";
    }

需求：在不过度占用内存的前提下读取Parquet文件并导入SQL Server。

我用C#开发了一个程序，功能是读取目录下的Parquet文件，将数据转换为DataTable后通过SqlBulkCopy导入SQL Server 2019本地数据库。程序采用了并行处理，但我是C#和并行计算的新手，代码主要通过ChatGPT和搜索资料完成。

当前实现流程 #

1. 获取文件列表：读取目标目录下所有Parquet文件到字符串数组

string[] fileList = GetParquetFiles(activeDirectory[0]);

2. 并行处理文件：用SemaphoreSlim限制并发文件处理数，并行处理每个文件

public static async Task ProcessParquetFileAsync(string[] fileList, string databaseName)
{
    int numberOfConcurrentFiles = 2;
    using (SemaphoreSlim semaphore = new SemaphoreSlim(numberOfConcurrentFiles))
    {
        List<Task> tasks = new List<Task>();
        foreach (var file in fileList)
        {
            await semaphore.WaitAsync();
            tasks.Add(Task.Run(async () =>
            {
                try
                {
                    await ReadAndDeployParquetFile(file, databaseName);
                }
                finally
                {
                    semaphore.Release();
                }
            }));
        }
        await Task.WhenAll(tasks);
    }
}

3. 单个文件处理逻辑：
   一次性读取整个Parquet文件为Table实例，复制其Schema到空DataTable，计算批次大小后拆分Table，再用SemaphoreSlim并行处理每个数据批次：

public static async Task ReadAndDeployParquetFile(string filePath, string databasename)
{
    using (ParquetReader reader = await ParquetReader.CreateAsync(filePath, null))
    {
        string tableName = GetTableName(filePath);
        Table parquetTable = await reader.ReadAsTableAsync();
        DataTable dataTable = new DataTable();

        string sql = $"CREATE TABLE {tableName} (";
        foreach(Field field in parquetTable.Schema.Fields)
        {
            DataField? ptField = field as DataField;
            string columnName = ptField.Name;
            Type columnType = ptField.ClrType;
            dataTable.Columns.Add(columnName, columnType);
            sql += $"[{columnName}] {GetSqlDataType(columnType, field)},";
        }
        sql = sql.Trim(',') + ')';
        SQLConnection conn = new SQLConnection();
        conn.ExecuteSqlCommand(sql, tableName, databasename);

        int rowCount = parquetTable.Count;
        int batchSize = 100000;
        decimal parts = Decimal.Ceiling((decimal)rowCount / (decimal)batchSize);
        
        SemaphoreSlim semaphore = new SemaphoreSlim(Environment.ProcessorCount);
        List<Task> tasks = new List<Task>();
        Console.WriteLine($"File {tableName} has total batch {(int)parts}");
        for (int i= 0; i < (int)parts; i++)
        {
            await semaphore.WaitAsync();
            int currentPart = i;
            tasks.Add(Task.Run (() =>
            {
                try
                {
                    ProcessBatch(parquetTable, dataTable.Clone(), currentPart, batchSize, tableName, databasename);
                }
                finally
                {
                    semaphore.Release();
                }
            }));
        }
        await Task.WhenAll(tasks);
    }
}

4. 数据批次处理逻辑：
   逐行从parquetTable取出数据填充到克隆的DataTable中，再导入SQL Server：

public static void ProcessBatch(Table parquetTable, DataTable partTable, int currentPart, int batchSize, string tableName, string databaseName)
{
    SQLConnection conn = new SQLConnection();
    int columnCount = parquetTable.Schema.Fields.Count;
    for (int i = currentPart * batchSize; (i < ((currentPart + 1) * batchSize)) && (i < parquetTable.Count); i++)
    {
        var row = parquetTable[i];
        var dataRow = partTable.NewRow();
        for (int j = 0; j < columnCount; j++)
        {
            if (row[j] != null)
            {
                dataRow[j] = row[j] ?? DBNull.Value;
            }
        }
        partTable.Rows.Add(dataRow);
    }
    conn.InsertTable(tableName, partTable, databaseName, currentPart);
    partTable.Dispose();
}

遇到的问题 #

一个包含200万行、文件大小仅24MB的Parquet文件，处理时程序内存占用飙升至3GB，且内存持续上升从未下降。使用dotMemory检测发现内存未被正常释放，需要分析内存泄漏原因并优化代码以降低内存占用。

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