C#带多组标签的Snowflake SQL查询批量数据导出程序 - IT评测·应用市场-qidao123.com

// AppConfig.cs
public class AppConfig
{
public SnowflakeConnectionConfig SnowflakeConnection { get; set; }
public List<QueryConfig> Queries { get; set; }
}
public class SnowflakeConnectionConfig
{
public string Account { get; set; }
public string User { get; set; }
public string Password { get; set; }
public string Warehouse { get; set; }
public string Database { get; set; }
public string Schema { get; set; }
public string Role { get; set; }
}
public class QueryConfig
{
public string Label { get; set; }
public string Sql { get; set; }
}

复制代码

// Logger.cs
public class Logger
{
private readonly string _logPath;
private readonly object _lock = new object();
public Logger(string logPath)
{
_logPath = logPath;
InitializeLogFile();
}
private void InitializeLogFile()
{
lock (_lock)
{
File.AppendAllText(_logPath, $"{"Timestamp",-25}|{"Status",-8}|{"Label",-20}|{"StartTime",-20}|{"Duration(s)",-12}|{"Rows",-10}|{"Error"}\n");
}
}
public void LogSuccess(string label, DateTime startTime, long rowCount, TimeSpan duration)
{
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";
WriteLogEntry(entry);
}
public void LogError(string label, DateTime startTime, string error, TimeSpan duration)
{
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";
WriteLogEntry(entry);
}
private void WriteLogEntry(string entry)
{
lock (_lock)
{
File.AppendAllText(_logPath, entry);
}
Console.Write(entry);
}
}

复制代码

// ExportProcessor.cs
using CsvHelper;
using Snowflake.Data.Client;
using System.Globalization;
public class ExportProcessor
{
private readonly string _connectionString;
private readonly Logger _logger;
public ExportProcessor(string connectionString, Logger logger)
{
_connectionString = connectionString;
_logger = logger;
}
public async Task ExportQueryAsync(QueryConfig query, string outputDir, CancellationToken cancellationToken = default)
{
var startTime = DateTime.UtcNow;
var sw = Stopwatch.StartNew();
long rowCount = 0;
string filePath = Path.Combine(outputDir, $"{query.Label}.csv");
try
{
using (var conn = new SnowflakeDbConnection())
{
conn.ConnectionString = _connectionString;
await conn.OpenAsync(cancellationToken);
using (var cmd = conn.CreateCommand())
{
cmd.CommandText = query.Sql;
using (var reader = await cmd.ExecuteReaderAsync(cancellationToken))
using (var writer = new StreamWriter(filePath, append: false))
using (var csv = new CsvWriter(writer, CultureInfo.InvariantCulture))
{
// Write header
for (int i = 0; i < reader.FieldCount; i++)
{
csv.WriteField(reader.GetName(i));
}
await csv.NextRecordAsync();
// Write rows
while (await reader.ReadAsync(cancellationToken))
{
for (int i = 0; i < reader.FieldCount; i++)
{
csv.WriteField(reader.GetValue(i));
}
await csv.NextRecordAsync();
rowCount++;
}
}
}
}
sw.Stop();
_logger.LogSuccess(query.Label, startTime, rowCount, sw.Elapsed);
}
catch (Exception ex)
{
sw.Stop();
_logger.LogError(query.Label, startTime, ex.Message, sw.Elapsed);
SafeDeleteFile(filePath);
throw; // Re-throw if using retry logic
}
}
private void SafeDeleteFile(string path)
{
try { File.Delete(path); }
catch { /* Ignore deletion errors */ }
}
}

复制代码

// Program.cs
using System.CommandLine;
class Program
{
static async Task Main(string[] args)
{
var configOption = new Option<FileInfo>(
name: "--config",
description: "Path to configuration file");
var outputOption = new Option<DirectoryInfo>(
name: "--output",
description: "Output directory for CSV files");
var rootCommand = new RootCommand
{
configOption,
outputOption
};
rootCommand.Description = "Snowflake Data Exporter";
rootCommand.SetHandler(async (config, outputDir) =>
{
await RunExport(config, outputDir);
}, configOption, outputOption);
await rootCommand.InvokeAsync(args);
}
static async Task RunExport(FileInfo configFile, DirectoryInfo outputDir)
{
// Read configuration
var config = JsonSerializer.Deserialize<AppConfig>(
File.ReadAllText(configFile.FullName),
new JsonSerializerOptions { PropertyNameCaseInsensitive = true });
// Create output directory
Directory.CreateDirectory(outputDir.FullName);
// Initialize logger
var logger = new Logger(Path.Combine(outputDir.FullName, "export.log"));
// Build connection string
var connBuilder = new SnowflakeDbConnectionStringBuilder
{
Account = config.SnowflakeConnection.Account,
User = config.SnowflakeConnection.User,
Password = config.SnowflakeConnection.Password,
Warehouse = config.SnowflakeConnection.Warehouse,
Db = config.SnowflakeConnection.Database,
Schema = config.SnowflakeConnection.Schema,
Role = config.SnowflakeConnection.Role
};
// Initialize processor
var processor = new ExportProcessor(
connBuilder.ToString(),
logger);
// Parallel execution with throttling
var parallelOptions = new ParallelOptions
{
MaxDegreeOfParallelism = Environment.ProcessorCount
};
await Parallel.ForEachAsync(
config.Queries,
parallelOptions,
async (query, cancellationToken) =>
{
await processor.ExportQueryAsync(
query,
outputDir.FullName,
cancellationToken);
});
}
}

复制代码

{
"snowflakeConnection": {
"account": "your_account",
"user": "your_user",
"password": "your_password",
"warehouse": "COMPUTE_WH",
"database": "PROD_DB",
"schema": "PUBLIC",
"role": "SYSADMIN"
},
"queries": [
{
"label": "customers",
"sql": "SELECT * FROM CUSTOMERS"
},
{
"label": "orders",
"sql": "SELECT * FROM ORDERS"
}
]
}

复制代码

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

复制代码