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

修改

Browse Source
This commit is contained in:
lindj
2024-01-12 16:50:37 +08:00
parent eab3695f53
commit 0984853c79
28 changed files with 1115 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
ConsoleApp2/HostedServices/Abstractions/IDataSource.cs Normal file
View File

@@ -0,0 +1,13 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace ConsoleApp2.HostedServices.Abstractions
{
public interface IDataSource:IDisposable
{
public Task DoEnqueue(Action<DataRecord> action);
}
}

52
ConsoleApp2/HostedServices/InputService.cs
View File

@@ -15,54 +15,52 @@ namespace ConsoleApp2.HostedServices;
public class InputService : IInputService
{
private readonly ILogger _logger;
private readonly IOptions<CsvOptions> _csvOptions;
private readonly IOptions<DataInputOptions> _dataInputOptions;
private readonly IOptions<InputTableOptions> _tableOptions;
private readonly DataRecordQueue _producerQueue;
private readonly ProcessContext _context;
public InputService(ILogger<InputService> logger,
IOptions<CsvOptions> csvOptions,
[FromKeyedServices(ProcessStep.Producer)]DataRecordQueue producerQueue,
ProcessContext context)
IOptions<DataInputOptions> dataInputOptions,
IOptions<InputTableOptions> tableOptions,
[FromKeyedServices(ProcessStep.Producer)] DataRecordQueue producerQueue,
ProcessContext context)
{
_logger = logger;
_csvOptions = csvOptions;
_dataInputOptions = dataInputOptions;
_tableOptions = tableOptions;
_producerQueue = producerQueue;
_context = context;
}
public async Task ExecuteAsync(CancellationToken cancellationToken)
{
var inputDir = _csvOptions.Value.InputDir;
var inputDir = _dataInputOptions.Value.InputDir;
_logger.LogInformation("***** Csv input service start, working dir: {InputDir}, thread id: {ThreadId} *****", inputDir, Environment.CurrentManagedThreadId);
var files = Directory.GetFiles(inputDir).Where(s => s.EndsWith(".sql") && !s.Contains("schema")).ToArray();
var files = Directory.GetFiles(inputDir);
if (files.Length == 0)
{
_logger.LogInformation("No sql files found in {InputDir}", inputDir);
_logger.LogInformation("No source files found in {InputDir}", inputDir);
return;
}
foreach (var sqlPath in files)
var count = 0;
foreach (var tableName in _tableOptions.Value.TableInfoConfig.Keys)
{
_logger.LogInformation("Working sql file: {SqlPath}", sqlPath);
var headers = await DumpDataHelper.GetCsvHeadersFromSqlFileAsync(sqlPath);
var csvFiles = await DumpDataHelper.GetCsvFileNamesFromSqlFileAsync(sqlPath);
foreach (var csvFile in csvFiles)
_logger.LogInformation("Working table: {tableName}", tableName);
var source = _dataInputOptions.Value.CreateSource?.Invoke(tableName);
await source.DoEnqueue((record) =>
{
var csvPath = Path.Combine(inputDir, csvFile);
// var source = new JsvSource(csvPath, headers, _logger);
var source = new CsvSource(csvPath, headers, _csvOptions.Value.Delimiter, _csvOptions.Value.QuoteChar, _logger);
_context.AddInput();
_producerQueue.Enqueue(record);
count++;
while (await source.ReadAsync())
{
_context.AddInput();
_producerQueue.Enqueue(source.Current);
if (cancellationToken.IsCancellationRequested)
return;
}
});
if (_context.GetExceptions().Count > 0)
{
_logger.LogInformation("***** Csv input service is canceled *****");
return;
}
_logger.LogInformation("File '{File}' input completed", Path.GetFileName(sqlPath));
_logger.LogInformation("table:'{tableName}' input completed", tableName);
}
_context.CompleteInput();

57
ConsoleApp2/HostedServices/MainHostedService.cs
View File

@@ -1,30 +1,73 @@
using ConsoleApp2.HostedServices.Abstractions;
using ConsoleApp2.Services;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using System.Threading.Tasks;
namespace ConsoleApp2.HostedServices;
public class MainHostedService : BackgroundService
{
private readonly ILogger _logger;
private readonly IInputService _input;
private readonly ITransformService _transform;
private readonly IOutputService _output;
private readonly ProcessContext _context;
public MainHostedService(IInputService input, ITransformService transform, IOutputService output)
public MainHostedService(ILogger<MainHostedService> logger, IInputService input, ITransformService transform, IOutputService output, ProcessContext context)
{
_logger = logger;
_input = input;
_transform = transform;
_output = output;
_context = context;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
var tasks = new List<Task>()
var inputTask = Task.Factory.StartNew(async () =>
{
try
{
await _input.ExecuteAsync(stoppingToken);
}
catch (Exception ex)
{
_context.AddException(ex);
_logger.LogError("Exception occurred on inputService:{Message},{StackTrace}", ex.Message, ex.StackTrace);
}
});
var transformTask = Task.Factory.StartNew(async () =>
{
Task.Run(async () => await _input.ExecuteAsync(stoppingToken), stoppingToken),
Task.Run(async () => await _transform.ExecuteAsync(stoppingToken), stoppingToken),
Task.Run(async () => await _output.ExecuteAsync(stoppingToken), stoppingToken),
};
await Task.WhenAll(tasks);
try
{
await _transform.ExecuteAsync(stoppingToken);
}
catch (Exception ex)
{
_context.AddException(ex);
_logger.LogError("Exception occurred on transformService:{Message},{StackTrace}", ex.Message, ex.StackTrace);
}
});
var outputTask = Task.Factory.StartNew(async () =>
{
try
{
await _output.ExecuteAsync(stoppingToken);
}
catch (Exception ex)
{
_context.AddException(ex);
_logger.LogError("Exception occurred on outputService:{Message},{StackTrace}", ex.Message, ex.StackTrace);
}
});
// await Task.Run(async () => await _output.ExecuteAsync(stoppingToken), stoppingToken);
}
}

70
ConsoleApp2/HostedServices/OutputService.cs
View File

@@ -5,6 +5,8 @@ using ConsoleApp2.Services;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using MySqlConnector;
using System.Threading;
namespace ConsoleApp2.HostedServices;
@@ -32,36 +34,40 @@ public class OutputService : IOutputService
_taskManager = taskManager;
}
public async Task ExecuteAsync(CancellationToken stoppingToken)
public async Task ExecuteAsync(CancellationToken cancellationToken)
{
_logger.LogInformation("***** Mysql output service started *****");
var records = new List<DataRecord>();
while (!_context.IsTransformCompleted || _consumerQueue.Count > 0)
var count = 0;
_taskManager.CreateTasks(async () =>
{
if (!_consumerQueue.TryDequeue(out var record)) continue;
records.Add(record);
if (records.Count >= _options.Value.FlushCount)
var records = new List<DataRecord>();
while (!_context.IsTransformCompleted || _consumerQueue.Count > 0)
{
var recordsCopy = records;
_taskManager.CreateTask(async () => await FlushAsync(recordsCopy), stoppingToken);
records = [];
if (!_consumerQueue.TryDequeue(out var record)) continue;
records.Add(record);
count++;
//_logger.LogInformation(@"*****OutputCount: {count} *****",count);
if (records.Count >= _options.Value.FlushCount)
{
await FlushAsync(records);
records.Clear();
}
if (_context.GetExceptions().Count>0)
{
_logger.LogInformation("***** Csv output service is canceled *****");
return;
}
}
if (_taskManager.TaskCount >= _options.Value.MaxTask)
if (_context.IsTransformCompleted && records.Count > 0)
{
await _taskManager.WaitAll();
_taskManager.ClearTask();
await FlushAsync(records);
records.Clear();
_context.CompleteOutput();
_logger.LogInformation("***** Mysql output service completed *****");
}
}
}, _options.Value.TaskCount);
await _taskManager.WaitAll();
await FlushAsync(records);
_context.CompleteOutput();
_logger.LogInformation("***** Mysql output service completed *****");
}
private async Task FlushAsync(IEnumerable<DataRecord> records)
@@ -69,15 +75,29 @@ public class OutputService : IOutputService
var count = 0;
await using var output = new MySqlDestination(
_options.Value.ConnectionString ?? throw new InvalidOperationException("Connection string is required"),
_logger, true);
_logger, _context,true);
//if (records == null || records.Count() == 0) return;
//var dbName = $"cferp_test_1";
//if (records != null && records.Count() > 0)
//{
// dbName = $"cferp_test_{records.FirstOrDefault()?.CompanyID}";
//}
//await using var output = new MySqlDestination(new MySqlConnectionStringBuilder
//{
// Server = "127.0.0.1",
// Port = 34309,
// Database = dbName,
// UserID = "root",
// Password = "123456",
// MaximumPoolSize = 50,
//}.ConnectionString, _logger,true);
foreach (var record in records)
{
await output.WriteRecordAsync(record);
count++;
}
await output.FlushAsync();
await output.FlushAsync(_options.Value.MaxAllowedPacket);
_context.AddOutput(count);
}
}

1
ConsoleApp2/HostedServices/TaskMonitorService.cs
View File

@@ -49,6 +49,7 @@ public class TaskMonitorService : BackgroundService
bool endCheck = false;
while (true)
{
if (_context.GetExceptions().Count>0) return;
EndCheck:
// var running = 0;
// var error = 0;

83
ConsoleApp2/HostedServices/TestInputService.cs Normal file
View File

@@ -0,0 +1,83 @@
using ConsoleApp2.HostedServices.Abstractions;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using ConsoleApp2.Const;
using ConsoleApp2.Options;
using ConsoleApp2.Services;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using System.Reflection.PortableExecutable;
using System.Collections.Concurrent;
using ConsoleApp2.SimulationService;
namespace ConsoleApp2.HostedServices
{
public class TestInputService : IInputService
{
private readonly ILogger _logger;
private readonly IOptions<CsvOptions> _csvOptions;
private readonly DataRecordQueue _producerQueue;
private readonly ProcessContext _context;
public TestInputService(ILogger<TestInputService> logger,
IOptions<CsvOptions> csvOptions,
[FromKeyedServices(ProcessStep.Producer)] DataRecordQueue producerQueue,
ProcessContext context)
{
_logger = logger;
_csvOptions = csvOptions;
_producerQueue = producerQueue;
_context = context;
}
public async Task ExecuteAsync(CancellationToken cancellationToken)
{
var tableName = "order_item";
var headers = new string[] { "ID","OrderNo","ItemNo","ItemType","RoomID","BoxID","DataID","PlanID","PackageID","Num","CompanyID","ShardKey" };
var dataCount = 1200000000L;
var tempCount = 80000;
var tempRecords=new List<DataRecord>();
var comanyID = 1;
short[] shareKeys = { 23040, 23070, 23100, 24000, 24040, 24070, 24100, 25000, 25040, 25070, 25100 };
int[] companyIds = { 1, 2, 3, 4 };
var sk = shareKeys.First();
var companyID = companyIds.First();
var shareKeyInterval = 20000;
var getShareKeyTimes = 0;
var getCompanyIDTimes = 0;
var shareKeyIntervalCount = 0;
for (long i = 1; i <= dataCount; i++)
{
shareKeyIntervalCount++;
if (shareKeyIntervalCount > shareKeyInterval) {
sk=DataHelper.GetShareKey(getShareKeyTimes);
getShareKeyTimes++;
shareKeyIntervalCount = 0;
}
var fields = new string[] { i.ToString(), "20220104020855", (220105981029 + i).ToString(), "1", "482278", "482279", "3768774", "0", "0", "1", companyID.ToString(), sk.ToString() };
var record = new DataRecord(fields, tableName, headers, comanyID);
tempRecords.Add(record);
if (tempRecords.Count >= tempCount)
{
foreach (var rc in tempRecords)
{
_context.AddInput();
_producerQueue.Enqueue(rc);
if (cancellationToken.IsCancellationRequested)
return;
}
tempRecords.Clear();
companyID = DataHelper. GetCompanyId(getCompanyIDTimes);
getCompanyIDTimes++;
}
}
_context.CompleteInput();
_logger.LogInformation("***** Csv input service completed *****");
}
}
}

43
ConsoleApp2/HostedServices/TransformService.cs
View File

@@ -36,12 +36,16 @@ public class TransformService : ITransformService
public async Task ExecuteAsync(CancellationToken cancellationToken)
{
_logger.LogInformation("***** Data transform service started, thread id: {ThreadId} *****", Environment.CurrentManagedThreadId);
while (!_context.IsInputCompleted || _producerQueue.Count > 0)
while ((!_context.IsInputCompleted || _producerQueue.Count > 0))
{
if (_context.GetExceptions().Count > 0)
{
_logger.LogInformation("***** Csv transform service is canceled *****");
return;
}
// var dbOptions = _options.Value.DatabaseFilter(record);
if (!_producerQueue.TryDequeue(out var record)) continue;
record.Database = _options.Value.DatabaseFilter?.Invoke(record);
for (var i = 0; i < record.Fields.Length; i++)
{
var field = record[i];
@@ -56,20 +60,41 @@ public class TransformService : ITransformService
switch (_options.Value.GetColumnType(record.TableName, record.Headers[i]))
{
case ColumnType.Blob or ColumnType.Text:
field = string.IsNullOrEmpty(field) ? "''" : $"0x{field}";
case ColumnType.Text:
field = string.IsNullOrEmpty(field) ? "''" : _options.Value.TransformBinary?.Invoke(field) ?? field; ;
break;
case ColumnType.Blob:
field = string.IsNullOrEmpty(field) ? "NULL" : $"0x{field}";
break;
default:
break;
}
Escape:
Escape:
record[i] = field;
}
// TODO: 数据处理/过滤/复制
//过滤不要的record
if (_options.Value.RecordFilter?.Invoke(record) == false) continue;
record.Database = _options.Value.DatabaseFilter?.Invoke(record);
//修改record
_options.Value.RecordModify?.Invoke(record);
//替换record
var replaceRecord = _options.Value.RecordReplace?.Invoke(record);
if (replaceRecord != null)
{
record = replaceRecord;
}
_consumerQueue.Enqueue(record);
//数据增加
var addRecords=_options.Value.RecordAdd?.Invoke(record);
if(addRecords != null)
{
foreach(var rc in addRecords)
{
_consumerQueue.Enqueue(rc);
}
}
_context.AddTransform();
}