如何从SQL Server动态表查询为Parquet创建动态Schema？

当然可以实现动态创建Schema，以下是具体的实现思路和代码示例：

1. 先获取完整的列元数据（含类型） #

只获取列名不够，还需要知道每个列对应的SQL数据类型，才能映射到.NET类型。修改你的SQL查询：

SELECT 
    c.name AS ColumnName,
    t.name AS SqlTypeName,
    c.is_nullable AS IsNullable
FROM sys.columns c
JOIN sys.types t ON c.system_type_id = t.system_type_id
WHERE c.object_id = OBJECT_ID(@TableName)
  AND c.name <> N'ID';

2. 建立SQL类型与.NET类型的映射表 #

根据你的数据库实际使用的类型，扩展这个映射：

var sqlTypeToClrType = new Dictionary<string, Type>
{
    ["int"] = typeof(int),
    ["nvarchar"] = typeof(string),
    ["varchar"] = typeof(string),
    ["datetime"] = typeof(DateTime),
    ["bit"] = typeof(bool),
    ["decimal"] = typeof(decimal),
    ["bigint"] = typeof(long)
    // 按需添加更多类型映射
};

3. 动态构建DataField集合 #

通过反射创建泛型的DataField<T>实例，这是构建Schema的核心：

string tableName = "你的表名";
string connString = "你的数据库连接字符串";
List<DataColumn> dataColumns = new List<DataColumn>();

// 第一步：获取列的元数据
List<(string ColumnName, Type ClrType)> columnTypes = new List<(string, Type)>();
using (var conn = new SqlConnection(connString))
{
    conn.Open();
    string metaSql = @"
SELECT 
    c.name AS ColumnName,
    t.name AS SqlTypeName,
    c.is_nullable AS IsNullable
FROM sys.columns c
JOIN sys.types t ON c.system_type_id = t.system_type_id
WHERE c.object_id = OBJECT_ID(@TableName)
  AND c.name <> N'ID';";

    using (var cmd = new SqlCommand(metaSql, conn))
    {
        cmd.Parameters.AddWithValue("@TableName", tableName);
        using (var reader = cmd.ExecuteReader())
        {
            while (reader.Read())
            {
                string colName = reader.GetString(0);
                string sqlType = reader.GetString(1);
                bool isNullable = reader.GetBoolean(2);

                if (!sqlTypeToClrType.TryGetValue(sqlType, out Type baseType))
                {
                    // 未知类型默认用string兜底
                    baseType = typeof(string);
                }

                // 处理可空类型
                Type finalType = isNullable && baseType.IsValueType 
                    ? typeof(Nullable<>).MakeGenericType(baseType) 
                    : baseType;

                columnTypes.Add((colName, finalType));
            }
        }
    }
}

4. 读取数据库数据并构建DataColumn #

用DataTable加载数据，方便逐列转换为对应类型的数组：

using (var conn = new SqlConnection(connString))
{
    conn.Open();
    string dataSql = $"SELECT {string.Join(", ", columnTypes.Select(c => $"[{c.ColumnName}]"))} FROM [{tableName}]";
    using (var cmd = new SqlCommand(dataSql, conn))
    {
        DataTable dt = new DataTable();
        dt.Load(cmd.ExecuteReader());

        foreach (var (colName, clrType) in columnTypes)
        {
            // 创建泛型DataField
            Type dataFieldType = typeof(DataField<>).MakeGenericType(clrType);
            DataField dataField = (DataField)Activator.CreateInstance(dataFieldType, colName);

            // 将DataTable列数据转换为对应类型的数组
            Array dataArray = Array.CreateInstance(clrType, dt.Rows.Count);
            for (int i = 0; i < dt.Rows.Count; i++)
            {
                object value = dt.Rows[i][colName];
                dataArray.SetValue(value == DBNull.Value ? null : value, i);
            }

            // 添加到DataColumn集合
            dataColumns.Add(new DataColumn(dataField, dataArray));
        }
    }
}

5. 生成Parquet文件 #

用动态构建的Schema写入文件：

var schema = new Schema(dataColumns.Select(dc => dc.Field).ToArray());

using (Stream fileStream = System.IO.File.OpenWrite("c:\\dynamic_output.parquet"))
{
    using (var parquetWriter = new ParquetWriter(schema, fileStream))
    {
        using (ParquetRowGroupWriter groupWriter = parquetWriter.CreateRowGroup())
        {
            foreach (var dc in dataColumns)
            {
                groupWriter.WriteColumn(dc);
            }
        }
    }
}

注意事项 #

• 如果处理超大数据集，不建议用DataTable加载全部数据，可以改成逐行读取并分批写入Parquet的行组
• 对于复杂类型（比如JSON、XML），可以统一映射为string类型处理
• 可根据实际需求扩展SQL类型与.NET类型的映射表，确保类型转换准确

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