使用 ParquetRecordSetWriter 时，NiFi 合并记录处理器不遵守最小 bin 大小

Question

我正在尝试在 NiFi (1.11.4) 中构建一个流程，该流程从 AMQ 读取 Avro 消息，使用 Merge Records 处理器累积它们，然后将合并的 parquet 文件写入 HDFS。

问题是当我尝试在 Merge Record 处理器中使用 ParquetRecordSetWriter 时（与 AvroReader 结合使用） - 合并的内容是从未基于最小箱大小阈值设置发出 - 我试图设置非常低的值 - 它只是不起作用。同时 Max Bin Age 阈值工作得很好。

另外，如果我使用 AvroRecordSetWriter - 最小尺寸阈值就可以了。 所以我尝试使用 AvroRecordSetWriter 后跟 PutParquet （或 ConvertAvroToParquet）并面临另一个问题： 如果我为 parquet 文件设置行组大小（例如 128 MB）- 则永远不会写入小文件。

看起来它在内存中缓冲内容，但它真的应该这样做吗？因为在我为测试 AvroParquetWriter（与 NiFi 使用的基本相同）而编写的简单 Java 程序中 - 我能够编写具有巨大行组大小集的小文件。

在 NiFi 中是否存在与拼花写入相关的任何已知问题？ 我对所描述的行为感到非常困惑。 任何帮助表示赞赏。

提前致谢。

Answer 1

我在 1.12.0 上运行类似的流程，但我发出的是 record count 和 max bin age，而不是 min bin size。工作得很好。这里可能对您有用的一件事是，我注意到 Parquet 输出大约是 Avro 二进制文件大小的 50%，在我们的数据集中非常一致。因此，如果您根据 Avro 来猜测 bin 内存大小，那么 Parquet 对您来说可能是错误的。