C#带多组标签的Snowflake SQL查询批量数据导出程序

设计一个基于多个带标签Snowflake SQL语句作为json设置文件的C#代码程序，实现根据差别的输入参数自动批量地将Snowflake数据库的数据导出为CSV文件到当地目录上，标签加扩展名.csv为导出数据文件名，文件已经存在则覆盖原始文件。需要考虑SQL结果集是大数据量分批数据导出的情况，通过多线程和异步操作来进步程序性能，程序需要异常处理和输出，输出出错时的错误信息，每次每个查询导出数据的运行状态和表数据行数以及运行时间戳，导出时间，输出每个文件记载数量的日记。
步调1：创建设置文件类

1. // AppConfig.cs
2. public class AppConfig
3. {
4.     public SnowflakeConnectionConfig SnowflakeConnection { get; set; }
5.     public List<QueryConfig> Queries { get; set; }
6. }
8. public class SnowflakeConnectionConfig
9. {
10.     public string Account { get; set; }
11.     public string User { get; set; }
12.     public string Password { get; set; }
13.     public string Warehouse { get; set; }
14.     public string Database { get; set; }
15.     public string Schema { get; set; }
16.     public string Role { get; set; }
17. }
19. public class QueryConfig
20. {
21.     public string Label { get; set; }
22.     public string Sql { get; set; }
23. }

复制代码

步调2：实现日记记载器

1. // Logger.cs
2. public class Logger
3. {
4.     private readonly string _logPath;
5.     private readonly object _lock = new object();
7.     public Logger(string logPath)
8.     {
9.         _logPath = logPath;
10.         InitializeLogFile();
11.     }
13.     private void InitializeLogFile()
14.     {
15.         lock (_lock)
16.         {
17.             File.AppendAllText(_logPath, $"{"Timestamp",-25}|{"Status",-8}|{"Label",-20}|{"StartTime",-20}|{"Duration(s)",-12}|{"Rows",-10}|{"Error"}\n");
18.         }
19.     }
21.     public void LogSuccess(string label, DateTime startTime, long rowCount, TimeSpan duration)
22.     {
23.         var entry = $"{DateTime.UtcNow:yyyy-MM-dd HH:mm:ss}|{"SUCCESS",-8}|{label,-20}|{startTime:HH:mm:ss}|{duration.TotalSeconds,-12:N2}|{rowCount,-10}|{"-"}\n";
24.         WriteLogEntry(entry);
25.     }
27.     public void LogError(string label, DateTime startTime, string error, TimeSpan duration)
28.     {
29.         var entry = $"{DateTime.UtcNow:yyyy-MM-dd HH:mm:ss}|{"ERROR",-8}|{label,-20}|{startTime:HH:mm:ss}|{duration.TotalSeconds,-12:N2}|{"-",-10}|{error}\n";
30.         WriteLogEntry(entry);
31.     }
33.     private void WriteLogEntry(string entry)
34.     {
35.         lock (_lock)
36.         {
37.             File.AppendAllText(_logPath, entry);
38.         }
39.         Console.Write(entry);
40.     }
41. }

复制代码

步调3：实现数据导出处理器

1. // ExportProcessor.cs
2. using CsvHelper;
3. using Snowflake.Data.Client;
4. using System.Globalization;
6. public class ExportProcessor
7. {
8.     private readonly string _connectionString;
9.     private readonly Logger _logger;
11.     public ExportProcessor(string connectionString, Logger logger)
12.     {
13.         _connectionString = connectionString;
14.         _logger = logger;
15.     }
17.     public async Task ExportQueryAsync(QueryConfig query, string outputDir, CancellationToken cancellationToken = default)
18.     {
19.         var startTime = DateTime.UtcNow;
20.         var sw = Stopwatch.StartNew();
21.         long rowCount = 0;
22.         string filePath = Path.Combine(outputDir, $"{query.Label}.csv");
24.         try
25.         {
26.             using (var conn = new SnowflakeDbConnection())
27.             {
28.                 conn.ConnectionString = _connectionString;
29.                 await conn.OpenAsync(cancellationToken);
31.                 using (var cmd = conn.CreateCommand())
32.                 {
33.                     cmd.CommandText = query.Sql;
35.                     using (var reader = await cmd.ExecuteReaderAsync(cancellationToken))
36.                     using (var writer = new StreamWriter(filePath, append: false))
37.                     using (var csv = new CsvWriter(writer, CultureInfo.InvariantCulture))
38.                     {
39.                         // Write header
40.                         for (int i = 0; i < reader.FieldCount; i++)
41.                         {
42.                             csv.WriteField(reader.GetName(i));
43.                         }
44.                         await csv.NextRecordAsync();
46.                         // Write rows
47.                         while (await reader.ReadAsync(cancellationToken))
48.                         {
49.                             for (int i = 0; i < reader.FieldCount; i++)
50.                             {
51.                                 csv.WriteField(reader.GetValue(i));
52.                             }
53.                             await csv.NextRecordAsync();
54.                             rowCount++;
55.                         }
56.                     }
57.                 }
58.             }
60.             sw.Stop();
61.             _logger.LogSuccess(query.Label, startTime, rowCount, sw.Elapsed);
62.         }
63.         catch (Exception ex)
64.         {
65.             sw.Stop();
66.             _logger.LogError(query.Label, startTime, ex.Message, sw.Elapsed);
67.             SafeDeleteFile(filePath);
68.             throw; // Re-throw if using retry logic
69.         }
70.     }
72.     private void SafeDeleteFile(string path)
73.     {
74.         try { File.Delete(path); }
75.         catch { /* Ignore deletion errors */ }
76.     }
77. }

复制代码

步调4：主程序实现

1. // Program.cs
2. using System.CommandLine;
4. class Program
5. {
6.     static async Task Main(string[] args)
7.     {
8.         var configOption = new Option<FileInfo>(
9.             name: "--config",
10.             description: "Path to configuration file");
11.         
12.         var outputOption = new Option<DirectoryInfo>(
13.             name: "--output",
14.             description: "Output directory for CSV files");
16.         var rootCommand = new RootCommand
17.         {
18.             configOption,
19.             outputOption
20.         };
22.         rootCommand.Description = "Snowflake Data Exporter";
23.         
24.         rootCommand.SetHandler(async (config, outputDir) =>
25.         {
26.             await RunExport(config, outputDir);
27.         }, configOption, outputOption);
29.         await rootCommand.InvokeAsync(args);
30.     }
32.     static async Task RunExport(FileInfo configFile, DirectoryInfo outputDir)
33.     {
34.         // Read configuration
35.         var config = JsonSerializer.Deserialize<AppConfig>(
36.             File.ReadAllText(configFile.FullName),
37.             new JsonSerializerOptions { PropertyNameCaseInsensitive = true });
39.         // Create output directory
40.         Directory.CreateDirectory(outputDir.FullName);
42.         // Initialize logger
43.         var logger = new Logger(Path.Combine(outputDir.FullName, "export.log"));
45.         // Build connection string
46.         var connBuilder = new SnowflakeDbConnectionStringBuilder
47.         {
48.             Account = config.SnowflakeConnection.Account,
49.             User = config.SnowflakeConnection.User,
50.             Password = config.SnowflakeConnection.Password,
51.             Warehouse = config.SnowflakeConnection.Warehouse,
52.             Db = config.SnowflakeConnection.Database,
53.             Schema = config.SnowflakeConnection.Schema,
54.             Role = config.SnowflakeConnection.Role
55.         };
57.         // Initialize processor
58.         var processor = new ExportProcessor(
59.             connBuilder.ToString(),
60.             logger);
62.         // Parallel execution with throttling
63.         var parallelOptions = new ParallelOptions
64.         {
65.             MaxDegreeOfParallelism = Environment.ProcessorCount
66.         };
68.         await Parallel.ForEachAsync(
69.             config.Queries,
70.             parallelOptions,
71.             async (query, cancellationToken) =>
72.             {
73.                 await processor.ExportQueryAsync(
74.                     query,
75.                     outputDir.FullName,
76.                     cancellationToken);
77.             });
78.     }
79. }

复制代码

步调5：设置文件示例（config.json）

1. {
2.   "snowflakeConnection": {
3.     "account": "your_account",
4.     "user": "your_user",
5.     "password": "your_password",
6.     "warehouse": "COMPUTE_WH",
7.     "database": "PROD_DB",
8.     "schema": "PUBLIC",
9.     "role": "SYSADMIN"
10.   },
11.   "queries": [
12.     {
13.       "label": "customers",
14.       "sql": "SELECT * FROM CUSTOMERS"
15.     },
16.     {
17.       "label": "orders",
18.       "sql": "SELECT * FROM ORDERS"
19.     }
20.   ]
21. }

复制代码

实现阐明

• 并行处理：
  
  
  
  • 使用Parallel.ForEachAsync进行并行查询处理
    
  • 默认并行度设置为处理器焦点数
    
  • 每个查询独立使用本身的数据库连接
    
  
  
• 大文件处理：
  
  
  
  • 使用CsvHelper进行流式写入
    
  • 采用异步I/O操作（ReadAsync/WriteAsync）
    
  • 逐行处理避免内存爆炸
    
  
  
• 错误处理：
  
  
  
  • 自动删除不完整文件
    
  • 具体错误日记记载
    
  • 异常传播与隔离设计
    
  
  
• 日记功能：
  
  
  
  • 布局化日记格式
    
  • 线程安全写入
    
  • 包含关键性能指标
    
  
  
• 性能优化：
  
  
  
  • 异步数据库操作
    
  • 并行查询实行
    
  • 流式结果集处理
    
  
  

使用阐明

• 安装依赖：
  
  1. dotnet add package Snowflake.Data
  2. dotnet add package CsvHelper
  3. dotnet add package System.CommandLine

  复制代码
• 编译运行：
  
  1. dotnet run -- --config ./config.json --output ./exports

  复制代码
• 输出布局：
  
  1. exports/
  2.   customers.csv
  3.   orders.csv
  4.   export.log

  复制代码

日记示例

1. 2023-09-20 14:30:45|SUCCESS |customers           |14:30:30|15.23      |1000000    |-
2. 2023-09-20 14:31:02|ERROR   |orders             |14:30:45|17.12      |-          |Timeout expired

复制代码

此实现提供了：

• 线程安全的并行处理
  
• 完整错误处理机制
  
• 具体的实行日记
  
• 高效的大数据处理能力
  
• 可设置的Snowflake连接参数
  
• 清楚的命令行界面
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。