Cloudera数据平台CDP公共云中用Nifi对Kafka的读写

Question

Nifi和Kafka现在都可以在Cloudera数据平台，CDP公共云中使用。Nifi擅长与所有事物交谈，Kafka是一种主流的信息总线，我只是想知道：

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

理想情况下，我会寻找在任何云中工作的步骤，例如Amazon和Microsoft。

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

Answer 1

将来会有多种形式的因素可用，现在我假设您的环境包含一个带有NiFi的数据中心，以及一个带有Kafka的数据集线器。(如果两者都位于同一个数据中心，则答案仍然有效)。

先决条件

• 使用NiFi和Kafka的数据中心
• 访问这些(例如添加处理器、创建Kafka主题)的权限
• 知道您的工作负载用户名(Cdp管理console>Click您的名称(左下角)>单击配置文件)
• 您应该将工作负载密码设置在同一个位置。

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

除非另有说明，否则我已经将所有内容保持在默认设置中。

卡夫卡数据中心集群中的：

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

• 如果您有Streams消息传递管理器:转到brokers选项卡，查看已经在一起的FQDN和端口
• 如果不能使用Streams消息传递管理器:使用Kafka转到Data的硬件选项卡，并获取相关节点的FQDN。(目前这些被称为代理)。然后添加:每一个后面的端口号。默认端口为9093。

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

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

In NiFi GUI:

1. 确保要生成NiFi中的一些数据，例如使用GenerateFlowFile处理器
2. 选择用于写入kafka的相关处理器，例如PublishKafka_2_0，将其配置如下：

• 设置
  • 自动终止关系:勾选成功和失败

• 属性
  • 卡夫卡经纪人:我们之前创建的合并列表
  • 安全协议: SASL_SSL
  • SASL机制:平原
  • SSL上下文服务:默认的NiFi SSL上下文服务
  • 用户名:您的工作负载用户名(见上面的先决条件)
  • 密码:您的工作负载密码
  • 题目名称: dennis
  • 使用事务:假
  • 最大元数据等待时间:30秒

1. 将GenerateFlowFile处理器连接到PublishKafka_2_0处理器并启动流

这些是最小的步骤，在Cloudera文件中可以找到更广泛的解释。注意，最好的做法是显式地创建主题(这个示例利用Kafka的特性，它可以在生成主题时自动创建主题)。

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

一次很好的检查，看看数据是否被写入卡夫卡，正在再次消费。

In NiFi GUI:

1. 例如，创建一个Kafka消费处理器( ConsumeKafka_2_0 )，将其属性配置为：

• Kafka Brokers、安全协议、SASL机制、SSL上下文服务、用户名、密码、主题名称:与上面的生产者示例相同
• 消费集团:1
• 偏移复位:最早

1. 创建另一个处理器，或一个向其发送消息的漏斗，然后启动消费处理器。

就是这样，在30秒内，你应该会看到你发布给卡夫卡的数据现在又流入了NiFi。

完全披露:我是Cloudera的雇员，Cloudera是Nifi背后的驱动力。