问题标签 [apache-nifi]

我想知道是否有办法捕获出现在 Nifi UI 上的公告消息（基本上是错误）并将其存储在某个属性/文件中，以便以后查看。屏幕每 5 分钟刷新一次，如果任何处理器出现故障，我想知道它的原因。

我在这里并不是特别谈论日志记录部分。

是否可以将 ftp 主机、端口等动态属性传递给 GetFTP 处理器。我试过了，但不知何故 GetFTP 处理器不接受任何上游连接。基本上，我想将所有源的 ftp 配置保存在一个文件中，并仅使用一个 GetFTP 处理器来读取该信息。这不是正确的方法吗？

我目前正在尝试让 NiFi 启动并运行，它现在正在初始化自己，但我不太确定如何解决这些异常。我的第一个猜测是我没有正确设置 PATH 变量，并且 NiFi 无法访问所有 java。我知道第二个例外只是因为第一个例外，但我仍然不确定是什么参数导致了标志。

第一个例外：

第二个例外：

关于 Apache Nifi 的另一个问题，我如何获取处理器的组 ID，当您右键单击处理器然后单击统计信息时也会出现该组 ID？有没有办法将此组 ID 值存储为某些属性/流文件内容的一部分。

我需要组 ID，以便我可以使用来自其余 api 的公告消息，以获取公告错误。

http[s]://{host}:{port}/nifi-api/controller/process-groups/{process-group-id}/status?recursive=true

我一直在尝试将 HDF 1.2.0 Nifi 配置为与 kerberized Kafka 一起使用，但无济于事。这是我的zookeeper-jaas.conf：

启动 PutKafka 处理器时出现错误：

我已经添加了

到 zookeeper.properties 文件。也将其设置为 bootstrap.conf

我已经多次检查 zookeeper-jaas.conf 并无法识别错误。任何指针将不胜感激。

我正在为我们的 ETL 流程评估 Nifi。我想构建以下流程：从 SQL 数据库中获取大量数据 -> 拆分为块，每个块 1000 条记录 -> 计算每个块中的错误记录 -> 计算错误记录的总数 -> 如果超过阈值失败过程 - > 否则将每个块保存到数据库。

我无法解决的问题是如何等到所有块都得到验证。例如，如果我有 5 个验证任务同时工作，我需要某种屏障来等待所有块都被处理，然后才运行错误计数处理器，因为我不想保存无效数据并在达到阈值时将其删除.

我的另一个问题是，是否有可能在多个节点上并行运行此验证处理器，并且仍然有可能等到它们全部完成。

新来的尼菲！

我想知道 nifi 中是否有一种方法可以使用诸如“PutFile”之类的处理器并将其写入一个文件（将数据附加到该文件，或覆盖该文件中的数据） - 而不是创建多个不同的文件？我需要使用另一个处理器来完成此操作吗？

我正在使用 Groovy 脚本上的 Apache NiFi 0.5.1 以将传入的 Json 值替换为映射文件中包含的值。映射文件如下所示（它是一个简单的 .txt）：

我从以下开始：

这第一步工作得很好，尽管它是硬编码的，而且远非理想。我最初的想法是使用 ReplaceTextWithMapping 来执行替换，但是它不适用于复杂的映射文件（例如多列）。我想更进一步，但我不知道如何去做。首先，我不想传入整个硬编码的 JSON，而是想读取传入的流文件。NiFi怎么可能？在将脚本作为 ExecuteScript 的一部分运行之前，我已经通过 UpdateAttribute 输出了一个包含内容的 .Json 文件，其中文件名 = myResultingJSON.json。此外，我知道如何使用 Groovy ( String mappingContent= new File('/path/to/file').getText('UTF-8') 加载 .txt 文件，但是如何使用加载的文件执行替换，以便生成的 JSON 如下所示：

谢谢您的帮助，

我。

编辑：

对脚本的第一次修改确实允许我从 InputStream 中读取：

然后，我开始使用 ConfigSlurper 测试该方法，并在将逻辑注入 Groovy ExecuteScript 之前编写了一个通用类：

我返回一个 groovy.lang.MissinPropertyException ，这是因为映射不是那么简单。所有字段/属性（从字段 1 到字段 3）都以相同的值（例如）进入 InpuStream，这意味着每次字段 2 具有该值时，您可以确定它对其他两个属性有效。但是，我不能有一个映射“field2”：“someText”的映射字段，因为实际映射是由映射文件中的第一个值驱动的。这里有一个例子：

在我的映射文件中，我有：

但是，如果您愿意，field1 需要映射到 A（文件中的第一个值）或保持不变。Field2 需要映射到最后一列 (A2) 中的值，最后 Field3 需要映射到中间列中的“一些文本”。

你能帮忙吗？这是我可以用 Groovy 和 ExecuteScript 实现的吗？如果需要，我可以将配置文件分成两个。

另外，我快速浏览了另一个选项（PutDistributedMapCache），我不确定我是否了解如何将键值对加载到分布式地图缓存中。看起来您需要一个 DistributedMapCacheClient ，但我不确定这是否易于实现。

谢谢！

编辑2：

其他一些进展，我现在映射工作，但不知道为什么它在读取属性文件的第二行时失败：

我回来了：This is my builder {"field2":"someText","field3":"someText"}

知道为什么吗？

太感谢了

编辑 3（从下面移动）

我写了以下内容：

但是，我必须更改映射文件的结构，如下所示：

然后我将 ConfigSlurper “合并”到 ExecuteScript 脚本中，如下所示：

问题似乎是这样一个事实，即我无法通过使用类似于在我的 TestLoadingMappings 中创建的逻辑来真正复制原始映射文件中的逻辑。正如我之前的评论/编辑中提到的，映射应该以这种方式工作：

field2 = if A then 替换为“一些文本”

field3 = 如果 A 则替换为 A2

...

field2 = B 然后替换为“其他一些文本”

field3 = B 然后替换为 B2

和儿子。

简而言之，映射由 InputStream 中的传入值驱动（变化），根据 JSON 属性有条件地映射到不同的值。您能否推荐一种通过 Groovy/ExecuteScript 实现此映射的更好方法？我可以灵活地修改映射文件，您能看到我可以更改它以实现所需映射的方法吗？

谢谢

我有一个带有 nifi 的工作网络服务，在那里我得到一个 httprequest 并可以响应它。对于测试，我使用 ReplaceText 将特定内容添加到响应中。这可行，但我的目标是添加一个流文件（或它的值）作为 httpresponse 的内容。

用例是，我可以发送信息（来自 Kafka、HDFS 等）作为 httprequest（REST Web 服务）的响应。

我正在寻找和测试解决方案，但没有任何效果。是否可以使用 Apache NiFi (Hortonworks HDF) 的标准处理器？

最好的问候n3

我想使用 ec2-user 而不是默认的 nifi 用户运行我的 NiFi 应用程序。我在 bootstrap.conf 中更改了 run.as=ec2-user 但它不起作用。它不允许我启动 Nifi 应用程序在启动 nifi 服务时出现以下错误。

任何指向这个问题的指针？