Nifi - 如何将数据放入 Hive 数据库？

Question

我正在构建一个 Nifi 流以从 kafka 获取 json 元素并将它们写入 Have 表中。

但是，关于处理器以及如何使用它们的文档很少甚至没有。

我打算做的如下：

kafka consume --> ReplaceText --> PutHiveQL

使用 kafka 主题效果很好。我收到一个 json 字符串。

我想提取 json 数据（使用 replaceText）并将它们放入配置单元表（PutHiveQL）中。

但是，我完全不知道该怎么做。文档没有帮助，也没有处理器使用的精确示例（或者我找不到）。

• 我的理论解决方案有效吗？
• 如何提取 json 数据，构建 HQL 查询并将其发送到我的本地 hive 数据库？

Answer 1

基本上你想将你的记录从 kafka 转换为 HQL 请求，然后将请求发送到 putHiveQl 处理器。

我不确定转换 kafka 记录 -> putHQL 可以通过替换文本来完成（接缝有点困难/棘手）。一般来说，我使用自定义 groovy 脚本处理器来执行此操作。

---

编辑

全球概览：

评估JsonPath

这会提取我的 Json 流文件的属性 timestamp 和 uuid 并将它们作为流文件的属性。

替换文本

这将流文件字符串设置为空字符串，并将其替换为 replacement value 属性，我在其中构建查询。

Answer 2

您可以使用 Puthivestreaming 进程直接注入流数据。 创建一个结构与流匹配的 ORC 表，并将流传递给它工作的 PUTHIVE3STreaming 处理器。