在Cloudera Data Platform CDP公有云中使用Nifi对Kafka进行读/写

Question

Nifi 和 Kafka 现已在 Cloudera Data Platform、CDP 公共云中可用。 Nifi 擅长与一切对话，而 Kafka 是主流消息总线，我只是想知道：

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

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

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

Answer 1

将来将会有多种形式可供选择，目前我假设您的环境包含 1 个带有 NiFi 的数据中心和 1 个带有 Kafka 的数据中心。 （如果两者都在同一个数据中心上，答案仍然有效）。

先决条件

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

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

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

在 Kafka 数据中心集群中：

1. 收集代理的 FQDN 链接以及使用的端口。

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

2. 以这种格式将链接组合在一起：FQDN：端口，FQDN：端口，FQDN：端口现在应该看起来像这样：

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

在 NiFi GUI 中：

3. 确保 NiFi 中有一些要生成的数据，例如使用

   GenerateFlowFile

   处理器
4. 选择写入kafka的相关处理器，例如

   PublishKafka_2_0

   ，配置如下：

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

5. 将您的

   GenerateFlowFile

   处理器连接到您的

   PublishKafka_2_0

   处理器并启动流程

这些是最少的步骤，可以在Cloudera文档中找到更广泛的解释。请注意，最佳实践是显式创建主题（此示例利用 Kafka 的功能，在生成主题时自动创建主题）。

这些步骤允许您在 CDP 公共云中通过 NiFi 使用来自 Kafka 的数据

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

在 NiFi GUI 中：

6. 创建一个Kafka消费处理器，例如

   ConsumeKafka_2_0

   ，配置其Properties如下：

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

7. 创建另一个处理器或漏斗来发送消息，并启动消费处理器。

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

---

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