问题标签 [apache-nifi]

当我尝试使用 Avro 在具有我各自架构的数据上运行 Kafka Consumer 时，它返回错误 "AvroRuntimeException: Malformed data. Length is negative: -40" 。我看到其他人在将字节数组转换为 json、Avro write and read以及Kafka Avro Binary *coder时遇到了类似的问题。我还引用了这个Consumer Group Example，它们都很有帮助，但是到目前为止对这个错误没有帮助。它一直工作到这部分代码（第 73 行）

解码器解码器 = DecoderFactory.get().binaryDecoder(byteArrayInputStream, null);

我已经尝试过其他解码器并打印出 byteArrayInputStream 变量的内容，该变量看起来像我认为您期望序列化的 avro 数据的样子（在消息中我可以看到架构和一些数据以及一些格式错误的数据）我打印出了使用 .available() 方法可用的字节数，该方法返回 594。我无法理解为什么会发生此错误。Apache Nifi 用于从 hdfs 生成具有相同模式的 Kafka 流。我将不胜感激任何帮助。

Using Apache Nifi, I created a flow that read a Json file and splits it line by line in order to verify if the content is correct. After that I have 2 outputs: 1 - for successful line and 2-for unsuccessful ones and the output is a Json file.

For the moment, all the lines are stored into separate files but what I want to do is to store each "good" line into 1 file and each "bad" one in another.

What processor should I use?

是否可以选择在特定流文件内容的循环中执行处理器操作？

作为一个具体的例子，使用时EvaluateXPath我想Phone使用以下方法检查每个子元素的值：

我想使用 Apache NiFi 的内置处理器“SplitJson”来拆分 JSON 数组。我以前做过，效果很好，但现在我似乎找不到正确的 json 路径表达式来拆分我的 json。我得到的 JSON 只是来自 github：

https://api.github.com/events

我想在单独的流文件中获取每个事件。在处理器的描述中，它指出：

一个 JsonPath 表达式，指示要拆分为 JSON/标量片段的数组元素。

因此，因为它是我要拆分的根数组，所以我尝试使用 $ 或 @ 但这不起作用。同样*或[*]其他我认为是错误的事情，例如$.[*]等，都不起作用。在某些情况下，它只是不拆分任何东西，有时它会给出一个错误，只是说它找不到数组。

提前感谢您的任何建议或帮助！

对于一个项目，我需要开发一个 ETL 过程（提取转换负载），该过程从一个（遗留）工具中读取数据，该工具在 REST API 上公开其数据。此数据需要存储在 amazon S3 中。

我真的很喜欢用 apache nifi 试试这个，但老实说，我还不知道如何连接 REST API，以及在哪里/如何实现一些业务逻辑来与源系统“对话正确的协议”。例如，我喜欢跟踪到目前为止已写入的数据，以便它可以继续加载它离开的地方。

到目前为止，我一直在阅读 nifi 文档，并且我可以更好地了解该工具提供/包含的内容。但是，尚不清楚如何在 nifi 架构中实现该任务。

希望有人能给我一些指导吗？

谢谢，保罗

我有一个看起来像的 json 数组

}，{“名称”：“ABCDEF”，“详细信息”：{“代码”：null，“描述”：null，“范围”：{“fromSerialNo”：“00061”，“toSerialNo”：“00173”}} ] }

我希望输出为 { "name": "ABCDEF", "details": { "code": null, "description": null, "range": { "fromSerialNo": "00001", "toSerialNo": " 99999"

但问题是我不能将变量放在 EvaluateJsonPath 中，因为它不接受 nifi 表达式语言。我必须遍历所有值，然后获得最终范围。有谁知道这是否可以在nifi中完成。

如何在 Apache Nifi ExecuteSQL 处理器（带有 in 和 out 参数）中使用过程调用专门用于说出参数。

有没有可能使用 NiFi 处理器将 facebook 数据获取到 hdfs 的方法？到目前为止，我什至没有找到关于这个案例的单一来源或样本……如果可能的话？程序是什么？

在 ExecuteScript 处理器中使用 ProcessorLog 的最佳方法是什么？如果在处理过程中遇到错误，我计划向我的 groovy ExecuteScript 添加一些日志记录。但是，尝试添加日志没有成功。可能忽略了一些明显的东西。

代码片段：

错误

我正在从 api 访问天气数据，并想使用 Apache nifi 来获取第一个 api 返回的 json 数组中所有城市的天气数据？

基本上我通过 2 个 API 访问数据：

这个 api 返回一个 JSON 数组，其中包含 JSON 数组元素中美国所有城市的列表：

现在我想遍历这个数组并使用要在另一个 API 中使用的州和城市信息，如下所示：

所以基本上我必须为我从第一个 API 获得的所有州和城市组合循环运行第二个 api。

到目前为止，我所做的基本 NiFi 流程与这个问题一样，但它不起作用。有人可以帮我弄清楚我在这里做错了什么吗？

HDP 2.4 沙盒上的 Hortonworks 数据流

SplitJson 处理器配置