hadoop - Nifi 在特定的通用 header 上加入两个 CSV 流文件

Question

尝试以 csv 格式合并两个传入的流文件，并根据一些共同的 header 值准备一个组合的 csv 输出，其中包含来自这两个文件的数据。

需要根据“creation_Date”和“Hour_of_Day”连接行，并根据“source_count”和“hive_count”之间的计算差异合并两个流文件。如果 INPUT_2 缺少某个 creation_date/Hour_of_day 而 INPUT_1 有 我也需要包括它 在“差异”标题下可能有一个“NOT AVAILABLE”。

我如何在 Nifi 中实现这一点？有没有一种方法可以使用 MergeContent 加入并在输出文件上运行查询以生成差异？

如有任何帮助，我们将不胜感激。 :)

SOURCE_INPUT_1

creation_Date,Hour_of_day,source_count
2018-08-16,3,19934
2018-08-16,0,3278
2018-08-16,1,10680
2018-08-16,4,19705
2018-08-16,2,14199
2018-08-16,5,672

SOURCE_INPUT_2

creation_date,hour_of_day,hive_count
2018-08-16,0,3279
2018-08-16,1,10680
2018-08-16,3,19933
2018-08-16,2,14199
2018-08-16,4,19700

组合输出

creation_Date,hour_of_day,source_count,hive_count,difference
2018-08-16,0,3278,3279,-1
2018-08-16,1,10680,10680,0
2018-08-16,2,14199,14199,0
2018-08-16,3,19934,19933,1
2018-08-16,4,19705,19700,5
2018-08-16,5,672,0,(NotAvailable)

Answer 1

Apache NiFi 通常不执行此类流连接。

它可以执行的连接更像是一种查找连接，其中数据的一侧是固定的，而另一侧来自传入流。这就是查找服务的工作方式，我相信有 CSV 查找服务。

另一种选择是将数据插入数据库，然后使用 ExecuteSQL 发出 SQL 查询，将它们连接在一起并将结果写入单个流文件。

最后，您可以使用流处理系统，如 Spark、Flink、Storm 等，它们都具有真正的流连接。

MergeContent 旨在将“类似”数据合并在一起，因此如果所有数据都是具有相同列的 CSV，并且您使用该数据创建多个流文件，那么您可以将它们全部合并到一个大 CSV 中。