Apache Nifi:计算 csv 文件中字符串中的数字
Apache Nifi: Counting numbers in a string inside a csv file
我有一个包含两列的 csv。但是 download
和 upload
列有一个字符串,可以有多个值,用逗号分隔。
download,upload
"12,3,43","2,33,7,10"
"2,1","4,5,6,23,1"
我需要在字符串中添加每个值,并在 Nifi 工作流程之后使用 addition.So 形成一个新的 csv,我的输出 csv 必须是:
download,upload
57,52
3,39
基本上应该是每个字符串中的值相加,如下所示。希望您能给我建议要使用的处理器和要更改的配置,以便从输入 csv 获得此输出 csv,如下所示。
您可以使用 JoltTransformJSON 处理器实现此目的,该处理器转换 JSON。它提供了一个 split
和 sum
转换,你需要它来获得你想要的输出。
策略是:
- 检查 CSV 文件的每一行
- 从 CSV 转换为 JSON
- 应用 Jolt 转换,将字符串拆分、转换为整数,最后求和
- 将记录转换回 CSV
- 将记录合并到 CSV 文件中
总流量:
生成流文件:
SplitRecord 将获取每个 CSV 行并将其转换为 JSON:
创建 CSVReader
和 JsonRecordSetWriter
。设置 CSVReader 以将第一行用作 header 行。保留默认属性。将每次拆分的记录设置为 1。
使用 JoltTransformJson
处理器并提供以下震动规格:
[
{
"operation": "modify-default-beta",
"spec": {
"downloadSplit": "=split(',', @(2,download))",
"uploadSplit": "=split(',', @(2,upload))"
}
}, {
"operation": "modify-overwrite-beta",
"spec": {
"downloadSplit": ["=toInteger", 0],
"uploadSplit": ["=toInteger", 0]
}
},
{
"operation": "modify-overwrite-beta",
"spec": {
"download": "=intSum(@(1,downloadSplit))",
"upload": "=intSum(@(1,uploadSplit))"
}
}, {
"operation": "shift",
"spec": {
"download": "download",
"upload": "upload"
}
}
]
- 分裂
- 转换为整数
- 求和
- 用转换后的download/uploaded覆盖原来的
将记录转换回 CSV:
保留 reader 和 writer 的默认属性。最后将每条记录合并回 CSV 文件:
使用 CSV reader 和具有默认属性的编写器。您可以通过调整 MergeRecord
的属性来控制每个 CSV 文件的记录数。详细了解属性 here.
的含义
输出:
请注意,原始行的顺序已更改。第一行对应于输入中的最后一行。
我有一个包含两列的 csv。但是 download
和 upload
列有一个字符串,可以有多个值,用逗号分隔。
download,upload
"12,3,43","2,33,7,10"
"2,1","4,5,6,23,1"
我需要在字符串中添加每个值,并在 Nifi 工作流程之后使用 addition.So 形成一个新的 csv,我的输出 csv 必须是:
download,upload
57,52
3,39
基本上应该是每个字符串中的值相加,如下所示。希望您能给我建议要使用的处理器和要更改的配置,以便从输入 csv 获得此输出 csv,如下所示。
您可以使用 JoltTransformJSON 处理器实现此目的,该处理器转换 JSON。它提供了一个 split
和 sum
转换,你需要它来获得你想要的输出。
策略是:
- 检查 CSV 文件的每一行
- 从 CSV 转换为 JSON
- 应用 Jolt 转换,将字符串拆分、转换为整数,最后求和
- 将记录转换回 CSV
- 将记录合并到 CSV 文件中
总流量:
生成流文件:
SplitRecord 将获取每个 CSV 行并将其转换为 JSON:
创建 CSVReader
和 JsonRecordSetWriter
。设置 CSVReader 以将第一行用作 header 行。保留默认属性。将每次拆分的记录设置为 1。
使用 JoltTransformJson
处理器并提供以下震动规格:
[
{
"operation": "modify-default-beta",
"spec": {
"downloadSplit": "=split(',', @(2,download))",
"uploadSplit": "=split(',', @(2,upload))"
}
}, {
"operation": "modify-overwrite-beta",
"spec": {
"downloadSplit": ["=toInteger", 0],
"uploadSplit": ["=toInteger", 0]
}
},
{
"operation": "modify-overwrite-beta",
"spec": {
"download": "=intSum(@(1,downloadSplit))",
"upload": "=intSum(@(1,uploadSplit))"
}
}, {
"operation": "shift",
"spec": {
"download": "download",
"upload": "upload"
}
}
]
- 分裂
- 转换为整数
- 求和
- 用转换后的download/uploaded覆盖原来的
将记录转换回 CSV:
保留 reader 和 writer 的默认属性。最后将每条记录合并回 CSV 文件:
使用 CSV reader 和具有默认属性的编写器。您可以通过调整 MergeRecord
的属性来控制每个 CSV 文件的记录数。详细了解属性 here.
输出:
请注意,原始行的顺序已更改。第一行对应于输入中的最后一行。