Cinchoo ETL——基于深度嵌套数组属性拆分大型JSON文件1、简介2. 要求3. 如何使用

46 阅读 0 评论 31 点赞

我是靠谱客的博主微笑心锁，最近开发中收集的这篇文章主要介绍Cinchoo ETL——基于深度嵌套数组属性拆分大型JSON文件1、简介2. 要求3. 如何使用，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

下载 Cinchoo ETL 源码
下载 Cinchoo ETL 二进制文件 (.NET Core)
下载 Cinchoo ETL 二进制文件 (.NET Framework)
工作示例 1（.NET fiddle）

1、简介

ChoETL是一个用于.NET的开源ETL（提取、转换和加载）框架。它是一个基于代码的库，用于在.NET环境中从多个来源提取数据、转换并加载到您自己的数据仓库中。您可以立即在数据仓库中获取数据。

本文讨论使用Cinchoo ETL框架基于深度嵌套的数组属性拆分大型JSON文件。此方法通过将大型JSON文件拆分为多个较小的文件，有助于解析大型JSON文件，而不会出现内存不足异常。

2. 要求

该框架库是使用.NET 4.5/.NET Core 3.x框架用C#编写的。

3. 如何使用

3.1 样本数据

让我们从查看下面的示例JSON文件开始。假设JSON文件很大，在这里，InstrumentData节点重复并包含数千个元素。本练习的目的是生成包含所有根节点的拆分文件，其中每个节点都有一个InstrumentData。

清单 3.1.1 输入JSON文件(sample.json)

{
"Job": {
"Keys": {
"JobID": "test123",
"DeviceID": "TEST01"
},
"Props": {
"FileType": "Measurements",
"InstrumentDescriptions": [
{
"InstrumentID": "1723007",
"InstrumentType": "Actual1",
"Name": "U",
"DataType": "Double",
"Units": "degC"
},
{
"InstrumentID": "2424009",
"InstrumentType": "Actual2",
"Name": "VG03",
"DataType": "Double",
"Units": "Pa"
}
]
},
"Steps": [
{
"Keys": {
"StepID": "START",
"StepResult": "NormalEnd"
},
"InstrumentData": [
{
"Keys": {
"InstrumentID": "1723007"
},
"Measurements": [
{
"DateTime": "2021-11-16 21:18:37.000",
"Value": 540
},
{
"DateTime": "2021-11-16 21:18:37.100",
"Value": 539
},
{
"DateTime": "2021-11-16 21:18:37.200",
"Value": 540
},
{
"DateTime": "2021-11-16 21:18:37.300",
"Value": 540
},
]
},
{
"Keys": {
"InstrumentID": "2424009"
},
"Measurements": [
{
"DateTime": "2021-11-16 21:18:37.000",
"Value": 1333.22
},
{
"DateTime": "2021-11-16 21:18:37.100",
"Value": 1333.22
},
]
}
]
}
]
}
}

第 1 步：在上面，Job.Keys和Job.Props是所有元素的共同因素，需要包含所有新的拆分文件。

第 2 步：由于输入文件自带Job.Steps[*].InstrumentData[*]节点，因此需要2级解析来按InstrumentData级别才能拆分文件。

第 3 步：按每个Steps[*]节点分解文件（又名Steps_0.json、Steps_1.json等）

第 4 步：然后获取每个StepsFiles并按每个InstrumentData[*]节点级别对其进行分解。（又名InstrumentData_0.json、InstrumentData_1.json等）

最终预期的拆分文件应如下所示。

清单 3.1.2 输出拆分JSON文件(InstrumentData_0.json)

{
"Job": {
"Keys": {
"JobID": "test123",
"DeviceID": "TEST01"
},
"Props": {
"FileType": "Measurements",
"InstrumentDescriptions": [
{
"InstrumentID": "1723007",
"InstrumentType": "Actual1",
"Name": "U",
"DataType": "Double",
"Units": "degC"
},
{
"InstrumentID": "2424009",
"InstrumentType": "Actual2",
"Name": "VG03",
"DataType": "Double",
"Units": "Pa"
}
]
},
"Steps": {
"Keys": {
"InstrumentID": "1723007"
},
"Measurements": [
{
"DateTime": "2021-11-16 21:18:37.000",
"Value": 540
},
{
"DateTime": "2021-11-16 21:18:37.100",
"Value": 539
},
{
"DateTime": "2021-11-16 21:18:37.200",
"Value": 540
},
]
}
}
}

首先要做的是安装ChoETL.JSON/ChoETL.JSON.NETStandard nuget包。为此，请在包管理器控制台中运行以下命令。

.NET Framework

Install-Package ChoETL.JSON

.NET Core


Install-Package ChoETL.JSON.NETStandard

现在将ChoETL命名空间添加到程序中。

using ChoETL;

3.2 拆分操作

由于JSON文件很大，我们需要考虑反序列化流模型中的InstrumentData节点，而不是将整个文件加载到内存中以避免内存压力。

首先，按Steps节点拆分文件，如下所示：

捕获Job.Keys节点的值。
捕获Job.Props节点的值。
然后循环遍历每个Job.Steps节点，使用ChoJObjectWriter（实用程序类以流方式写入json值）生成输出拆分文件（又名Steps_0.json，Steps_1.json等）
最后，该方法返回拆分文件名的步骤列表，以便按InstrumentData节点拆分。

清单 3.2.1 按步骤节点拆分

static string[] SplitBySteps(string inputFilePath)
{
List<string> stepsFiles = new List<string>();
dynamic keys = null;
dynamic props = null;
//Capture Keys
using (var r = new ChoJSONReader(inputFilePath).WithJSONPath("$..Job.Keys"))
{
keys = r.FirstOrDefault();
}
//Capture props
using (var r = new ChoJSONReader(inputFilePath).WithJSONPath("$..Job.Props"))
{
props = r.FirstOrDefault();
}
int fileCount = 0;
//Loop thro Steps, write to individual files
using (var r = ChoJSONReader.LoadText(json).WithJSONPath("$..Job.Steps")
.NotifyAfter(1)
.Setup(s => s.RowsLoaded += (o, e) =>
$"Step Nodes loaded: {e.RowsLoaded} <- {DateTime.Now}".Print())
//Callback used to hook up to loader, stream the nodes to file
//(this avoids loading to memory)
.Configure(c => c.CustomJObjectLoader = (sr, s) =>
{
string outFilePath = $"Steps_{fileCount++}.json";
$"Writing to `{outFilePath}` file...".Print();
using (var topJo = new ChoJObjectWriter(outFilePath))
{
topJo.Formatting = Newtonsoft.Json.Formatting.Indented;
using (var jo = new ChoJObjectWriter("Job", topJo))
{
jo.WriteProperty("Keys", keys);
jo.WriteProperty("Props", props);
jo.WriteProperty("Steps", sr);
}
}
//File.ReadAllText(outFilePath).Print();
//"".Print();
stepsFiles.Add(outFilePath);
return ChoJSONObjects.EmptyJObject;
})
)
{
r.Loop();
}
return stepsFiles.ToArray();
}

下一步是使用上述生成的步骤一次拆分文件，按InstrumentData节点拆分它们。下面的代码显示了如何。

接下来，按InstrumentData节点拆分文件，如下所示：

捕获Job.Keys节点的值。
捕获Job.Prop节点的值。
捕获Job.Steps.Keys节点的值。
然后遍历每个Job.Steps.InstrumentData节点，使用ChoJObjectWriter生成输出分割文件（又名InstrumentData_0.json, InstrumentData_1.json等）。
最后，该方法返回InstrumentData拆分文件名列表。

清单 3.2.2 按InstrumentData节点拆分

static string[] SplitByInstrumentData(string stepsFilePath)
{
List<string> instrumentDataFiles = new List<string>();
dynamic keys = null;
dynamic props = null;
dynamic stepsKeys = null;
//Capture Keys
using (var r = new ChoJSONReader(stepsFilePath).WithJSONPath("$..Job.Keys"))
{
keys = r.FirstOrDefault();
}
//Capture props
using (var r = new ChoJSONReader(stepsFilePath).WithJSONPath("$..Job.Props"))
{
props = r.FirstOrDefault();
}
//Capture steps/keys
using (var r = new ChoJSONReader(stepsFilePath).WithJSONPath("$..Job.Steps.Keys"))
{
stepsKeys = r.FirstOrDefault();
}
int fileCount = 0;
//Loop thro InstrumentData, write to individual files
using (var r = ChoJSONReader.LoadText(json).WithJSONPath("$..Job.Steps.InstrumentData")
.NotifyAfter(1)
.Setup(s => s.RowsLoaded += (o, e) => $"InstrumentData Nodes loaded:
{e.RowsLoaded} <- {DateTime.Now}".Print())
//Callback used to hook up to loader, stream the nodes to file
//(this avoids loading to memory)
.Configure(c => c.CustomJObjectLoader = (sr, s) =>
{
string outFilePath = $"InstrumentData_{fileCount++}.json";
$"Writing to `{outFilePath}` file...".Print();
using (var topJo = new ChoJObjectWriter(outFilePath))
{
topJo.Formatting = Newtonsoft.Json.Formatting.Indented;
using (var jo = new ChoJObjectWriter("Job", topJo))
{
jo.WriteProperty("Keys", keys);
jo.WriteProperty("Props", props);
jo.WriteProperty("Steps", sr);
}
}
File.ReadAllText(outFilePath).Print();
"".Print();
instrumentDataFiles.Add(outFilePath);
return ChoJSONObjects.EmptyJObject;
})
)
{
r.Loop();
}
return instrumentDataFiles.ToArray();
}

最后，使用以上两种方法通过InstrumentData完成拆分过程，如下图所示。

清单 3.2.3 Main()方法

public static void Main()
{
var inputFilePath = "input.json";
foreach (var stepsFilePath in SplitBySteps(inputFilePath))
{
SplitByInstrumentData(stepsFilePath);
File.Delete(stepsFilePath);
}
}

fiddle示例：https ://dotnetfiddle.net/j3Y03m

https://www.codeproject.com/Tips/5322235/Cinchoo-ETL-Split-a-Large-JSON-File-Based-on-Deepl