为数据库中的多个表调度单个nifi进程组

Question

我在Nifi中创建了一个流程组，其要求如下：

从蜂巢表中提取数据，>>加密内容，>>上传到azure存储。

现在我有3000张表，上面的流程需要排定。是否有任何方法只对所有表使用单个流，而不是为每个表创建3000个流。

在这里输入图像描述

另外，我想为一些表执行蔚蓝存储，而不是对所有表执行。是否有任何方法在流中给出指示的条件，即表1应该只到gcloud，而不是Azure。类似地，我希望表2同时进入蔚蓝和gcloud。

提前谢谢

Answer 1

有几件事可以做，但我希望一些最优的方法仍然存在。

不需要3000个流量

SelectHiveQL可以接受一个可以包含SELECT查询的输入流文件，因此我想到的一种方法是在一个文件中使用3000个SELECT查询，并使用GetFile读取该文件，并使用SplitText将其按行拆分，然后将splits关系连接到SelectHiveQL处理器，并应用您的进一步处理逻辑，即EncryptContent，并在需要时将其写入WASB。

决策驱动路由

对于第二个查询，即决定将哪些文件放入Azure Blob，以及将哪些文件放入GCS，您可以使用RouteOnAttribute处理器。

SelectHiveQL写入一个属性query.input.tables，该属性包含以逗号分隔格式选择的表列表。如果您只查询了一个表，则仅为query.input.tables编写该表名。因此，在RouteOnAttribute中，您可以拥有一个动态关系属性，如：

• Destination.Azure : ${query.input.tables:in("Table_1", "Table_15")}
• Destination.GCS : ${query.input.tables:in("Table_2", "Table_5")}

然后将Destination.Azure连接到PutAzureBlobStorage处理器，将Destination.GCS连接到PutGCSObject。

在上面的示例中，我使用了in表达式语言提供的NiFi运算符，您可以根据表命名约定使用其他运算符。例如，如果希望将客户表放在Azure Blob中，而供应商表位于GCS中，则可以使用以下方法：

• Destination.Azure : ${query.input.tables:startsWith("customer")}
• Destination.GCS : ${query.input.tables:startsWith("supplier")}

有关NifFi表达式语言提供的操作符和功能的完整列表，请查看此文档。