标签: hortonworks-dataflow

用户指南https://nifi.apache.org/docs/nifi-docs/html/user-guide.html具有优先顺序的以下详细信息,请您帮助我了解这些不同之处并提供任何实时示例.

FirstInFirstOutPrioritizer:给定两个FlowFiles,首先处理首先到达连接的FlowFiles.

OldestFlowFileFirstPrioritizer:给定两个FlowFiles,将首先处理数据流中最旧的FlowFiles.'如果没有选择优先顺序,则使用此默认方案.

我在 Kafka 主题中有大约 2000000 条消息，我想使用 NiFi 将这些记录放入 HDFS，所以我PutHDFS为此使用处理器，但ConsumeKafka_0_10它在 HDFS 中生成小文件，所以我使用合并内容处理器来合并记录在推送文件之前。 如果配置需要更改，请提供帮助 这对于少量消息来说效果很好，但当涉及到具有大量数据的主题时，会为每条记录写入一个文件。

谢谢你！！

为了快速测试和开发，有一种方法可以清除处理器右上角的公告确实很有帮助。一旦我看到错误并修复它，我想在重新启动之前清除它。

如果当前版本可以，请指导。

所述DBCPConnectionPool Service要求5的连接参数来建立到数据库的连接如图所示在下面的图片 [标记为黄色]

我曾经UpdateAttribute Processor手动添加这5个连接参数并给它们各自的值,如下图所示 [标记为黄色]

现在,当我试图DBCPConnectionPool Service通过这些属性读取连接参数的值时(如下图所示),我无法读取它们.

要知道之所以DBCPConnectionPool Service无法读取Flowfile属性,我继续检查两个源代码DBCPConnectionPool Service和UpdateAttribute Processor.

https://github.com/apache/nifi/blob/master/nifi-nar-bundles/nifi-standard-services/nifi-dbcp-service-bundle/nifi-dbcp-service/src/main/java/org/阿帕奇/ nifi/DBCP/DBCPConnectionPool.java

https://github.com/apache/nifi/blob/master/nifi-nar-bundles/nifi-update-attribute-bundle/nifi-update-attribute-processor/src/main/java/org/apache/nifi/处理器/属性/ UpdateAttribute.java

Souce代码DBCPConnectionPool Service:

Souce代码UpdateAttribute Processor:

因此,我开始知道它无法从FlowFile属性中读取值的原因.这是因为ExpressionLanguageScope它受到限制VARIABLE_REGISTRY而不受限制FLOWFILE_ATTRIBUTES.

现在,我的问题是,为什么ExpressionLanguageScope对DBCPConnectionPool Service被限制为VARIABLE_REGISTRY.这种限制的原因是什么？我问这个问题的原因是因为我想通过FlowFile属性读取连接参数的值.

我正在从 MySQL 表中导入数据（仅适用于选定的列）并将其放入 HDFS。完成后，我想在 Hive 中创建一个表。

为此，我有一个schema.sql包含整个表的 CREATE TABLE 语句的文件，我只想为我导入的列生成新的 CREATE TABLE 语句。

类似于我grep在下面的示例中所做的事情。

我FetchFile一起使用ExtractText但无法使其工作。如果我将整体模式放入一个属性中，我如何使用 NiFi 处理器甚至表达式语言来实现这一点？

或者有没有更好的方法在导入的数据上创建表？

根据我在使用 NiFi 构建一些数据库摄取 PoC 后的理解，整个数据流作为流文件流运行。在任何特定时间，执行控制可以同时在一个或多个处理器上。

所以我对如何调试复杂的数据流以解决任何故障感到非常困惑。

我的 PoC 工作流程本身如下所示。

当我们使用生产用例时，它可能会变得比这复杂得多。所以我有几个问题。

1. 如何知道数据流的状态。如果假设 10 个分叉流文件中有 4 个GenerateTableFetch因数据库池错误而失败，我如何知道哪些文件失败以及如何快速重播它们，而无需逐一查找数据来源。

2. 有没有一种方法可以通过查看数据流来了解哪个处理器上的哪个流文件发生了故障。

我对使用 NiFi 调试数据流有很多疑问/困惑，如果有人可以给我指出一些文档或分享最佳实践，那将会很有帮助。

谢谢。

我目前的理解是NiFi处理器属性特定于该处理器。因此，向处理器添加新属性将仅在该处理器中可见，而不会传递给以后的处理器块？

这就是为什么UpdateAttribute有必要添加在流文件中遍历数据流时保留在其中的元数据的原因：

那么，允许用户在处理器中添加自定义属性（超出该处理器执行所定义和要求的属性）的价值是什么？它类似于创建可以在其他属性中使用的变量吗？

我正在 Apache NiFi 中创建数据摄取工作流，使用 Kafka 作为缓冲系统。我有一个运行相同工作流的 3 节点集群设置，每个节点有 4 个核心。

我依赖于将数据移入和移出不同 Kafka 主题的多个实例，这是工作流程中最慢的部分，并且在性能方面非常不一致，因为两个相同的测试可能会增加 100% 的持续时间。

我们的 Publish 和 Consume Kafka 处理器在所有三个节点上运行，我们的 Kafka 主题有 3 个分区，跨越三个代理。

有没有人知道什么会导致这种不一致以及我可以做些什么来减轻它并加快工作流程？

我正在尝试学习如何创建自定义NiFi控制器服务。首先，我想DBCPConnectionPool通过简单地复制原始DBCPConnectionPool服务源代码来模仿控制器服务。为此，我从“ nifi - service -bundle -archetype” 生成了一个Maven原型，并获得了以下项目结构

然而，当我从“nifi-生成的原型处理器 -bundle-原型，我得到了以下的结构： -

我知道在处理器的情况下，我只需要MyProceesor.java在nifi-ListDbTableDemo-processors文件夹下编写当前代码，然后从中创建一个nar文件。但是在控制器服务的情况下，我生成了4个文件夹。我可以看到两个Java文件，即

1. StandardMyService.java存在于资料nifi-DbcpServiceDemo夹下

2. MyService.java存在于资料nifi-DbcpServiceDemo-api夹下

现在，为什么在自定义控制器服务的情况下生成两个Java文件，而在自定义处理器的情况下为什么仅生成一个Java文件。另外，由于我试图模仿该DBCPConnectionPool服务，因此应在其中两个Java文件中复制服务的原始源代码DBCPConnectionPool。

请从头开始指导我，创建与该服务等效的自定义服务所需遵循的步骤DBCPConnectionPool。