FlinkServer对接Kafka

操作场景

本章节介绍Kafka作为source表或者sink表的DDL定义，以及创建表时使用的WITH参数和代码示例，并指导如何在FlinkServer作业管理页面操作。

本示例以安全模式Kafka为例。

前提条件

• 集群中已安装HDFS、Yarn、Kafka和Flink服务。
• 包含Kafka服务的客户端已安装，例如安装路径为：/opt/Bigdata/client
• 参考基于用户和角色的鉴权创建一个具有“FlinkServer管理操作权限”的用户用于访问Flink WebUI，如：flink_admin。

操作步骤

1. 使用flink_admin登录Manager，选择“集群 > 服务 > Flink”，在“Flink WebUI”右侧，单击链接，访问Flink的WebUI。
2. 参考新建作业，新建Flink SQL流作业，在作业开发界面进行作业开发，配置完成后启动作业。
   需勾选“运行参数”中的“开启CheckPoint”，“时间间隔（ms）”可设置为“60000”，“模式”可使用默认值。

   CREATE TABLE KafkaSource (
     `user_id` VARCHAR,
     `user_name` VARCHAR,
     `age` INT
   ) WITH (
     'connector' = 'kafka',
     'topic' = 'test_source',
     'properties.bootstrap.servers' = 'Kafka的Broker实例业务IP:Kafka端口号',
     'properties.group.id' = 'testGroup',
     'scan.startup.mode' = 'latest-offset',
     'format' = 'csv',
     'properties.sasl.kerberos.service.name' = 'kafka',
     'properties.security.protocol' = 'SASL_PLAINTEXT',
     'properties.kerberos.domain.name' = 'hadoop.系统域名'
   );
   CREATE TABLE KafkaSink(
     `user_id` VARCHAR,
     `user_name` VARCHAR,
     `age` INT
   ) WITH (
     'connector' = 'kafka',
     'topic' = 'test_sink',
     'properties.bootstrap.servers' = 'Kafka的Broker实例业务IP:Kafka端口号',
     'scan.startup.mode' = 'latest-offset',
     'value.format' = 'csv',
     'properties.sasl.kerberos.service.name' = 'kafka',
     'properties.security.protocol' = 'SASL_PLAINTEXT',
     'properties.kerberos.domain.name' = 'hadoop.系统域名'
   );
   Insert into
     KafkaSink
   select
     *
   from
     KafkaSource;

   

   Kafka端口号：

   • 安全模式为“sasl.port”的值，默认为“21007”。
   • 非安全模式为“port”的值，默认为“9092”。如果配置端口号为9092，则需要配置“allow.everyone.if.no.acl.found”参数为true，具体操作如下：

     登录FusionInsigh Manager系统，选择“集群 > 服务 > Kafka > 配置 > 全部配置”，搜索“allow.everyone.if.no.acl.found”配置，修改参数值为true，保存配置即可。

3. 查看作业管理界面，作业状态为“运行中”。
4. 参考管理Kafka主题中的消息，查看Topic并向Kafka中写入数据。

   ./kafka-topics.sh --list --zookeeper ZooKeeper的quorumpeer实例业务IP:ZooKeeper客户端端口号/kafka

   sh kafka-console-producer.sh --broker-list Kafka角色实例所在节点的IP地址:Kafka端口号 --topic 主题名称 --producer.config 客户端目录/Kafka/kafka/config/producer.properties

   例如本示例使用主题名称为test_source：sh kafka-console-producer.sh --broker-list Kafka角色实例所在节点的IP地址:Kafka端口号 --topic test_source --producer.config /opt/Bigdata/client/Kafka/kafka/config/producer.properties

   输入消息内容：

   1,clw,33

   输入完成后按回车发送消息。

   

   • ZooKeeper的quorumpeer实例业务IP：

     ZooKeeper服务所有quorumpeer实例业务IP。登录FusionInsight Manager，选择“集群 > 服务 > ZooKeeper > 实例”，可查看所有quorumpeer实例所在主机业务IP地址。

   • ZooKeeper客户端端口号：

     登录FusionInsight Manager，选择“集群 > 服务 > ZooKeeper”，在“配置”页签查看“clientPort”的值。默认为24002。

5. 执行以下命令查看Sink表中是否接收到数据，即查看Kafka topic是否正常写入数据。

   sh kafka-console-consumer.sh --topic 主题名称 --bootstrap-server Kafka的Broker实例业务IP:Kafka端口号 --consumer.config /opt/Bigdata/client/Kafka/kafka/config/consumer.properties

WITH主要参数说明