Flink 使用之连接 Kafka

Flink 使用介绍相关文档目录

前言

本文介绍Flink 1.17.1版本连接Kafka的配置和操作方式。原文档Flink 使用之 SQL 连接 Kafka对应的Flink版本较老（1.10），已过期。请参考本文档。

环境信息

Flink 1.17.1
Kafka 1.1.1
Hadoop 3.x

Flink较新的版本（1.15.x以上），和本文验证1.17时的使用方法大同小异，均可参考本文。

环境依赖配置

首先讲解在Flink安装目录中引入依赖的方法。下载下方依赖到$Flink_HOME/lib目录中：

flink-connector-kafka-1.17.1.jar
flink-connector-base-1.17.1.jar
kafka-clients-3.2.3.jar

注意：这里使用的kafka-clients为flink-connector-kafka依赖的版本，尽管和集群中实际的Kafka不同，但是仍可正常工作。Flink官网也对此做出说明：Kafka | Apache Flink

如果没有在集群Flink安装目录中放置Kafka Connector相关jar包。我们还可以编译打包时将Kafka Connector合并入作业jar包，也能够运行。推荐将依赖放在$Flink_HOME/lib目录中，这样每个作业无需再重复引入依赖。
作业需要引入的依赖为：

<dependency>  
    <groupId>org.apache.flink</groupId>  
    <artifactId>flink-connector-kafka</artifactId>  
    <version>${flink.version}</version>  
</dependency>

执行环境

Flink操作Kafka既可以通过SQL client，又可以通过提交Flink作业。下面分别介绍两种操作方式和环境配置。

使用Flink SQL client

下面以最复杂的Kerberos 环境下使用Flink SQL操作Kafka为例，如果实际使用环境没有开启Kerberos，可忽略相关认证部分配置。

配置flink-conf.yaml启用Kerberos Kafka认证相关配置。涉及到的配置项如下所示：

security.kerberos.login.use-ticket-cache: true
security.kerberos.login.keytab: /etc/security/keytabs/paul.keytab
security.kerberos.login.principal: paul
security.kerberos.login.contexts: Client,KafkaClient

如果没有配置上述认证内容，会出现如下错误：

[ERROR] Could not execute SQL statement. Reason:
java.lang.IllegalArgumentException: Could not find a 'KafkaClient' entry in the JAAS configuration.

需要使用集群的Ranger，赋予paul用户读写HDFS权限，提交Yarn作业到目标队列的权限和访问Kafka的权限。

Flink SQL Client的使用参见Flink 使用之 SQL Client。

使用Flink作业

下面仍以较为复杂的Kerberos环境下使用为例。如果编写好的作业目标是在集群上运行，无需在代码中配置Kerberos认证部分，只需要在Flink集群中配置flink-conf.yaml启用Kerberos Kafka认证相关配置。

按照环境依赖配置所述配置完依赖之后，编写作业代码。示例代码如下：

EnvironmentSettings environmentSettings = EnvironmentSettings.inStreamingMode();  
final TableEnvironment tEnv = TableEnvironment.create(environmentSettings);

// 创建Kafka表，具体内容省略
String sql = "create table demo ...";

tEnv.executeSql(sql);  
tEnv.executeSql("select * from demo").print();

需要注意的是，如果在IDE里面运行Flink连接Kerberos环境下Kafka的代码，需要配置Kerberos认证。在代码开头加入：

System.setProperty("java.security.auth.login.config", "/path/to/kafka_jaas.conf");  
System.setProperty("java.security.krb5.conf", "/path/to/krb5.conf");

其中krb5.conf可以从服务器上复制到本地。kafka_jaas.conf服务器上的无法直接使用，需要修改。内容如下所示：

KafkaServer {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
keyTab="/path/to/paul.keytab"
storeKey=true
useTicketCache=false
serviceName="kafka"
principal="paul@PAULTECH.COM";
};
KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
keyTab="/path/to/paul.keytab"
storeKey=true
useTicketCache=true
renewTicket=true
serviceName="kafka"
principal="paul@PAULTECH.COM";
};
Client {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
keyTab="/path/to/paul.keytab"
storeKey=true
useTicketCache=false
serviceName="zookeeper"
principal="paul@PAULTECH.COM";
};

其中keyTab为本地存放keytab的路径，principal为keytab对应的principal。该principal需要具备的权限和上一节中所述相同。

操作方式

使用Flink SQL

执行Kafka查询SQL，如下例子所示：

CREATE TABLE demo (
  `name` string,
  `item_id` BIGINT,
  `ts` TIMESTAMP(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  'topic' = 'demo',
  'properties.bootstrap.servers' = 'kafka1:9092,kafka2:9092,kafka3:9092',
  'properties.security.protocol' = 'SASL_PLAINTEXT',
  'properties.sasl.kerberos.service.name' = 'kafka',
  'properties.sasl.mechanism' = 'GSSAPI',
  'properties.kerberos.domain.name' = 'PAULTECH.COM',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
);

注意：启用Kerberos必须添加如下配置：

'properties.security.protocol' = 'SASL_PLAINTEXT'
'properties.sasl.kerberos.service.name' = 'kafka'
'properties.sasl.mechanism' = 'GSSAPI'
'properties.kerberos.domain.name' = 'PAULTECH.COM'

务必添加上述的配置，否则运行SQL时连接会出现Timed out waiting for a node assignment错误。

上面的SQL中大家可能注意到ts这个METADATA字段。METADATA字段的信息并非包含在topic的数据中，而是保存在Kafka topic以及Kafka Record的元数据中（可以理解为topic的一些属性值，和Kafka Record除了用户数据之外的一些属性）。除了timestamp之外，其他可用的元数据参见官网。

METADATA字段还可以被声明为VIRTUAL，意思是这个字段仅供查询（select可见），但是insert into到其他表的时候，VIRTUAL字段会被忽略。有些METADATA字段是只读的（比如Kafka的topic名），使用这种METADATA字段的时候，必须声明为VIRTUAL。

通过SQL将数据写入Kafka较为简单。同样使用SQL方式建立Kafka topic到Flink的映射表之后，使用insert into sink select * from source语句，可将source表的内容写入到sink表中，最终写入到sink表对应的Kafka topic中。

上面示例代码中已包含常用的配置参数。如果需要更为详细的配置，参见官网Kafka | Apache Flink。

使用Table API

示例Java代码如下所示：

// 如果提交在集群运行，只需在flink-conf.yaml中配置认证，无需在这里添加认证代码
// 仅需在IDE中运行的时候添加
System.setProperty("java.security.auth.login.config", "/path/to/kafka_jaas.conf");  
System.setProperty("java.security.krb5.conf", "/path/to/krb5.conf");
  
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();  
KafkaSource<String> source = KafkaSource.<String>builder()  
        .setBootstrapServers("kafka1:9092,kafka2:9092,kafka3:9092")  
        .setTopics("demo")  
        .setGroupId("my-group")  
        .setStartingOffsets(OffsetsInitializer.earliest())  
        .setValueOnlyDeserializer(new SimpleStringSchema())  
        .setProperty("security.protocol", "SASL_PLAINTEXT")  
        .setProperty("sasl.kerberos.service.name", "kafka")  
        .setProperty("sasl.mechanism", "GSSAPI")  
        .setProperty("kerberos.domain.name", "PAULTECH.COM")  
        .build();  
  
DataStreamSource<String> kafkaSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");  
  
kafkaSource.print();  
env.execute();

注意：这里setProperty配置的4个属性用于连接Kerberos环境下的Kafka集群。如果没有启用Kerberos，可忽略这些配置。其他Kafka参数配置都可通过setProperty方法。

写入数据到Kafka可通过KafkaSink。示例代码如下：

KafkaSink.<String>builder()  
    .setBootstrapServers("kafka1:9092,kafka2:9092,kafka3:9092")  
    .setRecordSerializer(KafkaRecordSerializationSchema.builder()  
        .setTopic(outputTopic)  
        .setValueSerializationSchema(new SimpleStringSchema())  
        .build()  
    )  
    .setDeliverGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)  
    .build();

dataStream.sinkTo(sink);