CZY/MES-ETL
1
0
Fork 0
Code Pull Requests Releases Activity

整理代码

Browse Source
This commit is contained in:
lindj 2024-01-22 16:52:10 +08:00
parent 7235400aee
commit fcc0de5b2a
6 changed files with 28 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
ConsoleApp2/HostedServices/MainHostedService.cs
View File

@ -12,8 +12,8 @@ public class MainHostedService : IHostedService
private readonly ITransformService _transform;
private readonly IOutputService _output;
private readonly ProcessContext _context;
private readonly Timer? _bigTableTimer;
private readonly Timer? _smallTableTimer;
private readonly Timer? _bigTableTimer=null;
private readonly Timer? _smallTableTimer=null;
public MainHostedService(ILogger<MainHostedService> logger, IInputService input, ITransformService transform, IOutputService output, ProcessContext context)
{
_logger = logger;

4
ConsoleApp2/Program.cs
View File

@ -60,7 +60,7 @@ async Task RunProgram()
var host = Host.CreateApplicationBuilder(args);
var commandOptions = host.Configuration.GetSection("CmdOptions").Get<CommandOptions>() ?? new CommandOptions();
Console.WriteLine($"InputDir:{commandOptions?.InputDir}");
if (commandOptions == null) throw new ArgumentNullException("commandOptions is null");
var oldestTime = DateTime.ParseExact(commandOptions.OldestTime, "yyyyMM", System.Globalization.DateTimeFormatInfo.InvariantInfo);
host.Services.Configure<CsvOptions>(option =>
{
@ -415,7 +415,7 @@ async Task RunProgram()
// Password = "123456",
// MaximumPoolSize = 50, // 这个值应当小于 max_connections
//}.ConnectionString;
options.ConnectionString = new MySqlConnectionStringBuilder(host.Configuration.GetConnectionString("MySqlMaster"))
options.ConnectionString = new MySqlConnectionStringBuilder(host.Configuration.GetConnectionString("MySqlMaster")??"")
{
CharacterSet = "utf8",
AllowUserVariables = true,

11
ConsoleApp2/Services/CsvSource.cs
View File

@ -16,10 +16,10 @@ public class CsvSource:IDataSource
//protected readonly StreamReader _reader;
private readonly ILogger? _logger;
protected readonly string _tableName;
protected string? _sqlFilePath;
protected string _sqlFilePath=string.Empty;
protected readonly string? _sqlFileText;
protected string[]? headers;
protected string[]? csvFiles;
protected string[] headers=Array.Empty<string>();
protected string[] csvFiles = Array.Empty<string>();
public string? CurrentRaw { get; protected set; }
public string Delimiter { get; private set; }
public char QuoteChar { get; private set; }
@ -31,9 +31,6 @@ public class CsvSource:IDataSource
_logger = logger;
Delimiter = delimiter;
QuoteChar = quoteChar;
string pattern = $"^.*\\.{tableName}\\..*\\.sql$";
_sqlFilePath = Directory.GetFiles(_inputDir).FirstOrDefault(s => Regex.Match(s, pattern).Success);
}
@ -129,6 +126,8 @@ public class CsvSource:IDataSource
}
public virtual async Task GetHeaderAndCsvFiles()
{
string pattern = $"^.*\\.{_tableName}\\..*\\.sql$";
_sqlFilePath = Directory.GetFiles(_inputDir).FirstOrDefault(s => Regex.Match(s, pattern).Success) ?? "";
var text = await File.ReadAllTextAsync(_sqlFilePath);
headers = DumpDataHelper.GetCsvHeadersFromSqlFileAsync(text);
csvFiles = DumpDataHelper.GetCsvFileNamesFromSqlFileAsync(text, new Regex(@"'.+\.dat.zst'"));

22
ConsoleApp2/Services/JsvSource.cs
View File

@ -11,23 +11,23 @@ namespace ConsoleApp2.Services;
[Obsolete]
public class JsvSource:IDataSource
{
private readonly string _inputDir;
private readonly JsvStringSerializer _jsv;
private readonly StreamReader _reader;
//private readonly string _inputDir;
//private readonly JsvStringSerializer _jsv;
//private readonly StreamReader? _reader;
// ReSharper disable once PrivateFieldCanBeConvertedToLocalVariable
private readonly ILogger? _logger;
private readonly string _tableName;
//private readonly ILogger? _logger;
//private readonly string _tableName;
public DataRecord Current { get; protected set; } = null!;
public string[]? Headers { get; }
public bool EndOfSource => _reader.EndOfStream;
//public bool EndOfSource => _reader.EndOfStream;
public JsvSource(string inputDir,string tableName, ILogger? logger = null)
{
_inputDir = inputDir;
_tableName = tableName;
_jsv = new JsvStringSerializer();
_logger = logger;
//_inputDir = inputDir;
//_tableName = tableName;
//_jsv = new JsvStringSerializer();
//_logger = logger;
}
public Task DoEnqueue(Action<DataRecord> action)
{
@ -36,6 +36,6 @@ public class JsvSource:IDataSource
public void Dispose()
{
_reader.Dispose();
// _reader?.Dispose();
}
}

12
ConsoleApp2/Services/ZstSource.cs
View File

@ -9,8 +9,6 @@ namespace ConsoleApp2.Services
public ZstSource(string inputDir, string tableName, string delimiter = ",", char quoteChar = '"',
ILogger? logger = null) : base(inputDir, tableName, delimiter = ",", quoteChar = '"', logger = null)
{
string pattern = $"^.*\\.{tableName}\\..*\\.sql.zst$";
_sqlFilePath = Directory.GetFiles(_inputDir).FirstOrDefault(s => Regex.Match(s, pattern).Success);
}
private static async Task<string> DecompressFile(string filePath)
@ -32,6 +30,8 @@ namespace ConsoleApp2.Services
}
public override async Task GetHeaderAndCsvFiles()
{
string pattern = $"^.*\\.{_tableName}\\..*\\.sql.zst$";
_sqlFilePath = Directory.GetFiles(_inputDir).FirstOrDefault(s => Regex.Match(s, pattern).Success) ?? "";
var text = await DecompressFile(_sqlFilePath);
headers= DumpDataHelper.GetCsvHeadersFromSqlFileAsync(text);
csvFiles= DumpDataHelper.GetCsvFileNamesFromSqlFileAsync(text, new Regex(@"'.+\.dat.zst'"));
@ -65,15 +65,15 @@ namespace ConsoleApp2.Services
public override async Task<DataRecord?> GetTestRecord()
{
await GetHeaderAndCsvFiles();
var file = csvFiles.FirstOrDefault();
var file = csvFiles?.FirstOrDefault();
if (file != null)
{
var filePath = Path.Combine(_inputDir, file);
using (var input = File.OpenRead(filePath))
using var input = File.OpenRead(filePath);
{
using (var decopress = new DecompressionStream(input))
using var decopress = new DecompressionStream(input);
{
using (var reader = new StreamReader(decopress))
using var reader = new StreamReader(decopress);
{
var line = await reader.ReadLineAsync();
var fields = ParseRow2(line, QuoteChar, Delimiter);

2
ConsoleApp2/SimulationService/SimulationInputService.cs
View File

@ -53,7 +53,9 @@ namespace ConsoleApp2.SimulationService
var shareKeyIntervalCount = 0;
var source = _dataInputOptions.Value.CreateSource?.Invoke(tableName);
if (source == null) throw new NullReferenceException($"create table source:{tableName} failed!");
var testRecord = await source.GetTestRecord();
if(testRecord == null) throw new NullReferenceException($"create testRecord failed, tableName:{tableName}");
for (long i = 1; i <= dataCount; i++)
{
shareKeyIntervalCount++;