NiFi 示例将 MySQL 中数据导出并转换为 JSON 发送到 Kafka

添加 ExecuteSQL 处理器

1. 添加 ExecuteSQL 到面板

   image

2. 配置 ExecuteSQL

   image

3. 配置 DBCPConnectionPool

   更改 DBCPConnectionPool 配置，直到将 state 从 Invalid 变为 Disable

   
   
   image

   数据库连接池配置信息

   
   
   image

   启用 DBCPConnectionPool

   
   
   image
   image
   image
   image

任务成功或失败时的处理方式

image

image

.

添加 ConvertAvroToJSON 处理器

添加 ConvertAvroToJSON 到面板，无需修改任何配置

image

添加 PublishKafka 处理器

1. 添加 PublishKafka_0_10 到面板

   image

2. 配置 PublishKafka_0_10

   1. PLAINTEXT
      

      image

   2. SSL

      image

   3. SASL_SSL

      连接阿里云 kafka 配置

      
      
      image

      注意:

      1. 修改 ${NIFi_HOME}/conf/bootstrap.conf 配置文件中 java.arg.16 ，然后重启 NiFi 服务。

         [root@cdh01 nifi-1.10.0]$ cat conf/bootstrap.conf | grep java.arg.16
         #java.arg.16=-Djavax.security.auth.useSubjectCredsOnly=true      # 默认
         java.arg.16=-Djava.security.auth.login.config=/root/kafka_conf/kafka_client_jaas.conf
         [root@cdh01 nifi-1.10.0]$ cat /root/kafka_conf/kafka_client_jaas.conf
         KafkaClient {
           org.apache.kafka.common.security.plain.PlainLoginModule required
           username="123456"
           password="123456";
         };
         

      2. 增加 sasl.mechanism 配置
         
         image

3. 配置 StandardSSLContextService

   提示: PLAINTEXT 无需配置此项

   更改 StandardSSLContextService 配置，直到将 state 从 Invalid 变为 Disable
   

   image
   
   配置 StandardSSLContextService
   如无 Keystore 则不用配置 Keystore 相关属性
   
   image

   启用 StandardSSLContextService
   

   image
   image

连接处理器

image

启动处理器

image