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

DataRecord结构改进，更新2025年数据表转换规则

Browse Source
This commit is contained in:
陈梓阳 2024-12-27 15:18:08 +08:00
parent 77a3909160
commit 8037fc74de
8 changed files with 166 additions and 125 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
MesETL.App/Const/TableNames.cs
View File

@ -25,7 +25,7 @@ public static class TableNames
public const string OrderProcessStep = "order_process_step";
public const string OrderProcessStepItem = "order_process_step_item";
public const string OrderScrapBoard = "order_scrap_board";
public const string OrderWaveGroup = "order_wave_group";
public const string OrderExtraList = "order_extra_list";
public const string ProcessGroup = "process_group";
public const string ProcessInfo = "process_info";
public const string ProcessItemExp = "process_item_exp";

173
MesETL.App/DataRecord.cs
View File

@ -2,46 +2,6 @@
public class DataRecord : ICloneable
{
/// <summary>
/// 尝试获取一条记录的某个字段值
/// </summary>
/// <param name="record"></param>
/// <param name="columnName"></param>
/// <param name="value"></param>
/// <returns></returns>
/// <exception cref="InvalidOperationException"></exception>
public static bool TryGetField(DataRecord record, string columnName, out string value)
{
value = string.Empty;
if (record.Headers is null)
throw new InvalidOperationException("Cannot get field when headers of a record have not been set.");
var idx = IndexOfIgnoreCase(record.Headers, columnName);
if (idx == -1)
return false;
value = record.Fields[idx];
return true;
}
/// <summary>
/// 获取一条记录的某个字段值
/// TODO: 最好能优化至O(1)
/// </summary>
/// <param name="record"></param>
/// <param name="columnName"></param>
/// <returns></returns>
/// <exception cref="InvalidOperationException"></exception>
/// <exception cref="IndexOutOfRangeException"></exception>
public static string GetField(DataRecord record, string columnName)
{
if (record.Headers is null)
throw new InvalidOperationException("Headers have not been set.");
var idx = IndexOfIgnoreCase(record.Headers, columnName);
if (idx is -1)
throw new IndexOutOfRangeException(
$"Column name '{columnName}' not found in this record, table name '{record.TableName}'.");
return record.Fields[idx];
}
private static int IndexOfIgnoreCase(IList<string> list, string value)
{
var idx = -1;
@ -57,45 +17,54 @@ public class DataRecord : ICloneable
return idx;
}
private readonly List<string> _fields;
private readonly List<string> _headers;
/// <summary>
/// 字段列表
/// </summary>
public IList<string> Fields { get; }
public IReadOnlyList<string> Fields => _fields;
/// <summary>
/// 表头列表
/// </summary>
public IList<string> Headers { get; }
public IReadOnlyList<string> Headers => _headers;
/// <summary>
/// 来源表名
/// </summary>
public string TableName { get; }
/// <summary>
/// 需要输出的数据库
/// </summary>
public string? Database { get; set; }
/// <summary>
/// 所有字段的总字符数量
/// </summary>
public long FieldCharCount { get; }
/// <summary>
/// 忽略这个记录，不会被输出
/// </summary>
public bool Ignore { get; set; }
public DataRecord(IEnumerable<string> fields, string tableName, IEnumerable<string> headers, string? database = null)
public DataRecord(IEnumerable<string> fields, string tableName, IEnumerable<string> headers,
string? database = null)
{
Fields = fields.ToList();
_fields = fields.ToList();
TableName = tableName;
Headers = headers.ToList();
_headers = headers.ToList();
Database = database;
if (Fields.Count != Headers.Count)
if (_fields.Count != _headers.Count)
throw new ArgumentException(
$"The number of fields does not match the number of headers. Expected: {Headers.Count} Got: {Fields.Count} Fields: {string.Join(',', Fields)}",
$"The number of fields does not match the number of headers. Expected: {_headers.Count} Got: {_fields.Count} Fields: {string.Join(',', _fields)}",
nameof(fields));
FieldCharCount = Fields.Sum(x => (long)x.Length);
FieldCharCount = _fields.Sum(x => (long)x.Length);
}
/// <summary>
@ -104,8 +73,8 @@ public class DataRecord : ICloneable
/// <param name="index"></param>
public string this[int index]
{
get => Fields[index];
set => Fields[index] = value;
get => _fields[index];
set => _fields[index] = value;
}
/// <summary>
@ -114,61 +83,101 @@ public class DataRecord : ICloneable
/// <param name="columnName"></param>
public string this[string columnName]
{
get => GetField(this, columnName);
get => GetField(columnName);
set => SetField(columnName, value);
}
/// <summary>
/// 尝试获取字段值
/// 尝试获取某个字段值
/// </summary>
/// <param name="columnName"></param>
/// <param name="value"></param>
/// <returns></returns>
public bool TryGetField(string columnName, out string value) => TryGetField(this, columnName, out value);
/// <summary>
/// 为一个字段赋值
/// </summary>
/// <param name="columnName"></param>
/// <param name="value"></param>
/// <returns></returns>
public bool SetField(string columnName, string value) => SetField(this, columnName, value);
/// <summary>
///
/// </summary>
/// <param name="record"></param>
/// <param name="columnName"></param>
/// <param name="value"></param>
/// <returns></returns>
/// <exception cref="InvalidOperationException"></exception>
/// <exception cref="IndexOutOfRangeException"></exception>
public static bool SetField(DataRecord record, string columnName, string value)
public bool TryGetField(string columnName, out string value)
{
// 表头检查
if (record.Headers is null)
value = string.Empty;
if (_headers is null)
throw new InvalidOperationException("Cannot get field when headers of a record have not been set.");
var idx = IndexOfIgnoreCase(_headers, columnName);
if (idx == -1)
return false;
value = _fields[idx];
return true;
}
/// <summary>
/// 获取一条记录的某个字段值
/// TODO: 最好能优化至O(1)
/// </summary>
/// <param name="columnName"></param>
/// <returns></returns>
/// <exception cref="InvalidOperationException"></exception>
/// <exception cref="IndexOutOfRangeException"></exception>
public string GetField(string columnName)
{
if (_headers is null)
throw new InvalidOperationException("记录的表头尚未设置，无法赋值");
var idx = IndexOfIgnoreCase(record.Headers, columnName);
var idx = IndexOfIgnoreCase(_headers, columnName);
if (idx is -1)
throw new IndexOutOfRangeException(
$"列 '{columnName}' 不存在于该纪录中，表名 '{record.TableName}");
record.Fields[idx] = value;
return true;
$"列 '{columnName}' 不存在于该纪录中，表名 '{TableName}");
return _fields[idx];
}
/// <summary>
/// 为记录的一个字段赋值，如果该字段名不存在则会抛出异常
/// </summary>
/// <param name="columnName">列名</param>
/// <param name="value">值</param>
/// <returns></returns>
/// <exception cref="InvalidOperationException">该记录的表头尚未初始化，你可能在错误的阶段调用了该方法</exception>
/// <exception cref="IndexOutOfRangeException">输入的字段名不存在于该记录中</exception>
public void SetField(string columnName, string value)
{
// 表头检查
if (_headers is null)
throw new InvalidOperationException("记录的表头尚未设置，无法赋值");
var idx = IndexOfIgnoreCase(_headers, columnName);
if (idx is -1)
throw new IndexOutOfRangeException(
$"列 '{columnName}' 不存在于该纪录中，表名 '{TableName}");
_fields[idx] = value;
}
/// <summary>
/// 在记录中追加一个字段
/// </summary>
/// <param name="columnName">字段名</param>
/// <param name="value">字段值</param>
/// <exception cref="InvalidOperationException">记录的表头尚未初始化，你可能在错误的阶段调用了此方法</exception>
/// <exception cref="ArgumentException">提供的字段名已存在于该记录中</exception>
public void AppendField(string columnName, string value)
{
if (_headers is null)
throw new InvalidOperationException("记录的表头尚未设置，无法赋值");
var idx = IndexOfIgnoreCase(_headers, columnName);
if (idx is > 0)
throw new ArgumentException($"字段名 '{columnName}' 已存在于该记录中，无法重复添加", nameof(columnName));
_headers.Add(columnName);
_fields.Add(value);
}
public void RemoveField(string columnName)
{
var idx = IndexOfIgnoreCase(Headers, columnName);
var idx = IndexOfIgnoreCase(_headers, columnName);
if (idx == -1)
throw new InvalidOperationException($"{TableName}: 列名 '{columnName}' 不存在");
Fields.RemoveAt(idx);
Headers.Remove(columnName);
_fields.RemoveAt(idx);
_headers.Remove(columnName);
}
public bool HeaderExists(string columnName) => IndexOfIgnoreCase(Headers, columnName) != -1;
public bool HeaderExists(string columnName) => IndexOfIgnoreCase(_headers, columnName) != -1;
public object Clone()
{
return new DataRecord(new List<string>(Fields), TableName, new List<string>(Headers), Database);
return new DataRecord(new List<string>(_fields), TableName, new List<string>(_headers), Database);
}
}

7
MesETL.App/HostedServices/MainHostedService.cs
View File

@ -77,6 +77,8 @@ public class MainHostedService : BackgroundService
await Task.WhenAll(inputTask, transformTask, outputTask);
_stopwatch.Stop();
_logger.LogInformation("***** 所有传输任务均已完成 *****");
if (_context.HasException)
_logger.LogError("***** 传输过程中有错误发生 *****");
_logger.LogInformation("***** 耗时：{Time}", (_stopwatch.ElapsedMilliseconds / 1000f).ToString("F3"));
await Task.Delay(5000, stoppingToken);
@ -114,7 +116,10 @@ public class MainHostedService : BackgroundService
{
var connStr = _databaseOptions.Value.ConnectionString
?? throw new ApplicationException("分库配置中没有配置数据库");
_logger.LogWarning("已开启MySQL延迟写入功能并禁用重做日志，请注意数据安全");
if (enable)
_logger.LogWarning("已开启MySQL延迟写入功能并禁用重做日志，请注意数据安全");
else _logger.LogInformation("不安全变量已关闭");
if (enable)
{
await DatabaseHelper.NonQueryAsync(connStr,

23
MesETL.App/Program.cs
View File

@ -1,4 +1,4 @@
#define USE_TEST_DB // 如果使用测试库运行，则加上USE_TEST_DB预处理器指令
// #define USE_TEST_DB // 如果使用测试库运行，则加上USE_TEST_DB预处理器指令
using System.Text;
using MesETL.App;
@ -12,6 +12,7 @@ using MesETL.App.Options;
using MesETL.App.Services.ErrorRecorder;
using MesETL.App.Services.Loggers;
using MesETL.App.Services.Seq;
using MesETL.Shared.Compression;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
@ -217,6 +218,15 @@ async Task RunProgram()
break;
}
// 将JsonStr列转换为Data列，添加CompressionType列
case TableNames.OrderModuleExtra:
{
record.AppendField("CompressionType", "1");
record.AppendField("Data",
Convert.ToHexString(DeflateArchive.Compress(Convert.FromHexString(record["JsonStr"]))));
record.RemoveField("JsonStr");
break;
}
// 删除ID列，让数据库自行递增
// TODO: 数据表改进，删除ID列或是替换为流水号
case TableNames.ProcessStepEfficiency:
@ -234,14 +244,13 @@ async Task RunProgram()
record.RemoveField("Key");
break;
}
#if USE_TEST_DB
// 测试环境忽略PlaceData列，生产环境会提前将其移除
// 移除PlaceData列（如果存在的话，生产库已经删除）
case TableNames.SimplePlanOrder:
{
record.RemoveField("PlaceData");
if(record.HeaderExists("PlaceData"))
record.RemoveField("PlaceData");
break;
}
#endif
default: break;
}
@ -281,7 +290,7 @@ async Task RunProgram()
var id = seq.AddCachedSeq(SeqConfig.OrderWaveGroupID);
var orderWaveGroup = new DataRecord(
[id.ToString(), ..headers.Select(c => record[c])],
TableNames.OrderWaveGroup,
TableNames.OrderExtraList,
["ID", "OrderNo", "ShardKey", "Type", "ConfigJson", "CompanyID"]);
resultList.Add(orderWaveGroup);
return resultList;
@ -368,7 +377,7 @@ async Task RunProgram()
host.Services.AddHostedService<MainHostedService>();
host.Services.AddSingleton<IInputService, FileInputService>();
host.Services.AddSingleton<ITransformService, TransformService>();
host.Services.AddSingleton<IOutputService, VoidOutputService>();
host.Services.AddSingleton<IOutputService, OutputService>();
host.Services.AddSingleton<TaskMonitorService>();
// host.Services.AddRedisCache(redisOptions);
host.Services.AddSingleton<ICacher, MemoryCache>();

2
MesETL.App/Services/ProcessContext.cs
View File

@ -45,7 +45,7 @@ public class ProcessContext
public void CompleteTransform() => IsTransformCompleted = true;
public void CompleteOutput() => IsOutputCompleted = true;
public bool AddException(Exception e) => _hasException = true;
public bool AddException(Exception e) => _hasException = true; // 没打算存起来，暂时先加个标记
public void AddInput() => Interlocked.Increment(ref _inputCount);

39
MesETL.Shared/Compression/DeflateArchive.cs Normal file
View File

@ -0,0 +1,39 @@
using System.IO.Compression;
namespace MesETL.Shared.Compression;
/// <summary>
/// Deflate压缩工具类
/// </summary>
public static class DeflateArchive
{
/// <summary>
/// 解压Deflate
/// </summary>
/// <param name="input"></param>
/// <returns></returns>
public static byte[] Decompress(byte[] input)
{
using var msi = new MemoryStream(input);
using var mso = new MemoryStream();
using var ds = new DeflateStream(msi, CompressionMode.Decompress);
ds.CopyTo(mso);
ds.Flush();
return mso.ToArray();
}
/// <summary>
/// 压缩Deflate
/// </summary>
/// <param name="input"></param>
/// <returns></returns>
public static byte[] Compress(byte[] input)
{
using var msi = new MemoryStream(input);
using var mso = new MemoryStream();
using var ds = new DeflateStream(mso, CompressionMode.Compress);
msi.CopyTo(ds);
ds.Flush();
return mso.ToArray();
}
}

22
MesETL.Shared/Helper/Helper.Compress.cs
View File

@ -1,22 +0,0 @@
using System.IO.Compression;
namespace MesETL.Shared.Helper;
public class CompressHelper
{
/// <summary>
///
/// </summary>
/// <param name="data"></param>
/// <returns></returns>
public static byte[] CompressDeflate(byte[] data)
{
using var src = new MemoryStream(data);
using var outStream = new MemoryStream();
using var gzip = new DeflateStream(outStream, CompressionMode.Compress);
src.CopyTo(gzip);
gzip.Flush();
return outStream.ToArray();
}
}

3
MesETL.Test/DatabaseToolBox.cs
View File

@ -12,7 +12,7 @@ namespace TestProject1;
public class DatabaseToolBox
{
private readonly ITestOutputHelper _output;
public const string ConnStr = "Server=127.0.0.1;Port=3306;UserId=root;Password=123456;";
public const string ConnStr = "Server=192.168.1.246;Port=33025;UserId=root;Password=123456;";
public DatabaseToolBox(ITestOutputHelper output)
{
@ -177,6 +177,7 @@ public class DatabaseToolBox
[InlineData("mesdb_3")]
[InlineData("mesdb_4")]
[InlineData("mesdb_5")]
[InlineData("mesdb_6")]
public async Task TruncateAllTable(string database)
{
var tables = await DatabaseHelper.QueryTableAsync(ConnStr,