在 Cloudera Data Platform CDP 公共云中使用 Nifi 读取/写入 Kafka

Posted 2023-04-18

【中文标题】在 Cloudera Data Platform CDP 公共云中使用 Nifi 读取/写入 Kafka

【英文标题】：Read/Write with Nifi to Kafka in Cloudera Data Platform CDP public cloud

【发布时间】：2020-07-17 14:57:46

【问题描述】：

Nifi 和 Kafka 现在都可以在 Cloudera 数据平台、CDP 公共云中使用。 Nifi 擅长与一切对话，而 Kafka 是主流的消息总线，我只是想知道：

在 CDP 公共云中从 Apache Nifi 向 Kafka 生产/消费数据所需的最少步骤是什么

理想情况下，我会寻找适用于任何云的步骤，例如 Amazon AWS 和 Microsoft Azure。

我对遵循最佳实践并使用平台默认配置的答案感到满意，但如果有常见的替代方案，也欢迎使用。

【参考方案1】：

未来将有多种外形尺寸可用，现在我假设您的环境包含 1 个带有 NiFi 的数据中心和 1 个带有 Kafka 的数据中心。 （如果两者都在同一个数据中心，答案仍然有效）。

先决条件

带有 NiFi 和 Kafka 的数据中心 访问这些的权限（例如添加处理器、创建 Kafka 主题） 了解您的工作负载用户名（Cdp 管理控制台>单击您的姓名（左下角）> 单击配置文件） 您应该在同一位置设置工作负载密码

这些步骤允许您在 CDP 公共云中将数据从 NiFi 生成到 Kafka

除非另有说明，否则我将所有内容都保留为默认设置。

在 Kafka 数据中心集群中：

收集代理的 FQDN 链接和使用的端口。

如果您有 Streams Messaging Manager：转到代理选项卡以查看 FQDN 和端口已经在一起 如果您无法使用 Streams Messaging Manager：使用 Kafka 转到 Data Hub 的硬件选项卡并获取相关节点的 FQDN。 （目前这些被称为经纪人）。然后在每个后面添加 :portnumber。默认端口为 9093。

以这种格式将链接组合在一起：FQDN:port,FQDN:port,FQDN:port 现在应该如下所示：

broker1.abc:9093,broker2.abc:9093,broker3.abc:9093

在 NiFi GUI 中：

确保您在 NiFi 中有一些数据要生成，例如使用 GenerateFlowFile 处理器 选择写入kafka的相关处理器，例如PublishKafka_2_0，配置如下：

设置 自动终止关系：勾选成功和失败 属性 Kafka Brokers：我们之前创建的组合列表 安全协议：SASL_SSL SASL 机制：普通 SSL 上下文服务：默认 NiFi SSL 上下文服务 用户名：您的工作负载用户名（请参阅上面的先决条件） 密码：您的工作负载密码 主题名称：丹尼斯 使用事务：false 最大元数据等待时间：30 秒

将您的 GenerateFlowFile 处理器连接到您的 PublishKafka_2_0 处理器并启动流程

这些是最少的步骤，更详细的解释可以在Cloudera Documentation 中找到。请注意，最佳实践是显式创建主题（此示例利用 Kafka 的特性，即在生成主题时自动创建主题）。

这些步骤允许您使用 NiFi 从 CDP 公共云中的 Kafka 消费数据

很好地检查数据是否已写入 Kafka，正在再次使用它。

在 NiFi GUI 中：

创建一个Kafka消费处理器，例如ConsumeKafka_2_0，配置其Properties如下：

Kafka 代理、安全协议、SASL 机制、SSL 上下文服务、用户名、密码、主题名称：与上面的生产者示例中的所有内容相同 消费群体：1 偏移重置：最早

创建另一个处理器或将消息发送到的漏斗，并启动消费处理器。

就是这样，在 30 秒内，您应该会看到您发布到 Kafka 的数据现在再次流入 NiFi。

---

全面披露：我是 Cloudera 的员工，是 Nifi 背后的驱动力。