TLS 日志服务提供 Kafka 协议消费功能，可以将一个日志主题当作一个 Kafka Topic 来消费，每条日志对应一条 Kafka 消息。您可以使用 Flink kafka 连接器连接日志服务，通过 Flink 任务将日志服务中采集的日志数据消费到下游的大数据组件或者数据仓库。

前置条件

1. 已经控制台开通 TLS 日志服务和流式计算 Flink 版服务。
2. 利用 Kafka 协议消费日志需要确保 TLS 控制台已经开通 Kafka 协议消费功能。并且需要保证 TLS 和 Flink 属于同一个 Region 使用私网进行连接。
3. Flink 当前版本建议使用 Flink-1.16-volcano 及以上版本。

DDL 定义

用作数据源（Source）

CREATE TABLE kafka_source (
   user_id BIGINT,
   item_id BIGINT,
   behavior STRING
) WITH (
     'connector' = 'kafka',
     -- 消费 topic 需要加前缀 out-
     'topic' = 'out-${TLS_TOPIC_ID}',
     -- 参考 TLS 文档站通过私网域名进行访问，9093 端口
     'properties.bootstrap.servers' = 'tls-cn-beijing.ivolces.com:9093',
     -- 配置安全协议为 SASL_SSL，验证机制为 PLAIN
     'properties.security.protocol' = 'SASL_SSL',
     'properties.sasl.mechanism' = 'PLAIN',
     -- 配置 JAAS。用户名密码分别填写 TLS 项目 ID 和 AK 、SK 内容
     'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="${TLS_PROJECT_ID}" password="${ACCESS_KEY}#${SECRET_KEY}";',
     -- 指定消费者组，和消费模式以及解析格式
     'properties.group.id' = 'test_topic_01',
     'scan.startup.mode' = 'latest-offset',
     'format' = 'json'
 );

用作数据目的（Sink）

CREATE TABLE kafka_sink (
   user_id BIGINT,
   item_id BIGINT,
   behavior STRING
 ) WITH (
   'connector' = 'kafka',
   -- 结合具体地域选择相应的私网域名 9094 端口
   'properties.bootstrap.servers' = 'tls-cn-beijing.ivolces.com:9094',
   --关闭幂等。
   'properties.enable.idempotence' = 'false',   
   -- 配置安全协议为 SASL_SSL，机制为 PLAIN
   'properties.security.protocol' = 'SASL_SSL',
   'properties.sasl.mechanism' = 'PLAIN',
   -- 配置 JAAS。用户名密码分别填写项目id AK 、SK 内容
   'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="${TLS_PROJECT_ID}" password="${ACCESS_KEY}#${SECRET_KEY}";',
   -- 注意写入的时候 topic 为 TLS 的 Topic id
   'topic' = '${TLS_TOPIC_ID}',
   'format' = 'json'
 );

WITH 参数

通用和消费者参数

生产者参数

另外作为生产者参数，TLS Kafka 支持如下参数类型：

SQL 示例代码

源表示例

以下是 TLS 作为源表的消费方式：

CREATE TABLE tls_source (
   user_id BIGINT,
   item_id BIGINT,
   behavior STRING
) WITH (
   'connector' = 'kafka',
   'topic' = 'out-${TLS_TOPIC_ID}',
   'properties.bootstrap.servers' = 'tls-cn-beijing.ivolces.com:9093',
   'properties.security.protocol' = 'SASL_SSL',
   'properties.sasl.mechanism' = 'PLAIN',
   'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="${TLS_PROJECT_ID}" password="${ACCESS_KEY}#${SECRET_KEY}";',
   'properties.group.id' = 'test_topic_01',
   'scan.startup.mode' = 'latest-offset',
   'format' = 'json'
 );

CREATE TABLE print_sink (
   user_id BIGINT,
   item_id BIGINT,
   behavior STRING
) WITH (
   'connector' = 'print'           
);
insert into print_sink
select * from tls_source;

生产者示例

以下是 TLS 作为生产者的结果表示例代码：

CREATE TABLE datagen_source (
   user_id BIGINT,
   item_id BIGINT,
   behavior STRING
) WITH (
   'connector' = 'datagen',
   'rows-per-second' = '5'          
);

CREATE TABLE tls_sink (
   user_id BIGINT,
   item_id BIGINT,
   behavior STRING
 ) WITH (
   'connector' = 'kafka',
   'properties.bootstrap.servers' = 'tls-cn-beijing.ivolces.com:9094',
   'properties.enable.idempotence' = 'false',   
   'properties.security.protocol' = 'SASL_SSL',
   'properties.sasl.mechanism' = 'PLAIN',
   'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="${TLS_PROJECT_ID}" password="${ACCESS_KEY}#${SECRET_KEY}";',
   'topic' = '${TLS_TOPIC_ID}',
   'format' = 'json'
 );

insert into tls_sink
select * from datagen_source;

JAVA 示例代码

POM 文件依赖，其中 Flink 版本当前支持 1.11、1.16 和 1.17 等选项。

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

消费者示例

TLS 消费代码示例：

package com.example;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


public class KafkaConsumerJob {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        String jaas = "org.apache.kafka.common.security.plain.PlainLoginModule" +
                " required username=\"${TLS_PROJECT_ID}\"" +
                " password=\"${ACCESS_KEY}#${SECRET_KEY}\";";

        // 配置 Kafka source
        KafkaSource<String> source = KafkaSource.<String>builder()
                // 设置消费
                .setBootstrapServers("tls-cn-beijing.ivolces.com:9093")
                .setTopics("out-${TLS_TOPIC_ID}")
                .setGroupId("flink-consumer-group")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setProperty("security.protocol", "SASL_SSL")
                .setProperty("sasl.mechanism", "PLAIN")
                .setProperty("sasl.jaas.config", jaas)
                .build();

        // 添加 source 并处理数据
        DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
        // 添加数据处理逻辑
        stream.map(value -> {
            System.out.println("Received message: " + value);
            return value;
        });
        // 执行任务
        env.execute("Kafka Consumer Job");
    }
}

生产者示例

TLS 写入代码示例：

package com.example;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;

import java.time.LocalDateTime;
import java.util.Properties;

public class KafkaProducerJob {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 创建数据源
        DataStream<String> sourceStream = env.addSource(new CustomDataSource());
        // 认证信息中注意填写 project-id、AK、SK
        String jaas = "org.apache.kafka.common.security.plain.PlainLoginModule" +
                " required username=\"${project-id}\"" +
                " password=\"${ACCESS_KEY}#${SECRET_KEY}\";";

        // 配置 Kafka sink
        KafkaSink<String> sink = KafkaSink.<String>builder()
                // 结合具体地域修改私网访问地址
                .setBootstrapServers("tls-cn-beijing.ivolces.com:9094")
                // 注意配置 topic-id 
                .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                        .setTopic("${topic-d}")
                        .setValueSerializationSchema(new SimpleStringSchema())
                        .build()
                )
                // 配置认证、关闭幂等特性等配置
                .setProperty("security.protocol", "SASL_SSL")
                .setProperty("enable.idempotence", "false")
                .setProperty("sasl.mechanism", "PLAIN")
                .setProperty("sasl.jaas.config", jaas)
                .build();
        // 将数据写入 Kafka
        sourceStream.sinkTo(sink);
        // 执行任务
        env.execute("Kafka Producer Job");
    }

    // 自定义数据源
    public static class CustomDataSource implements SourceFunction<String> {
        private volatile boolean isRunning = true;

        @Override
        public void run(SourceContext<String> ctx) throws Exception {
            while (isRunning) {
                // 模拟 json 数据写入
                String message = "{\"a\": 1, \"b\": 2}";
                ctx.collect(message);
                Thread.sleep(500); // 每秒生成一条数据
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }
    }
}

FAQ

丢状态重启

如果修改了作业的拓扑结构，新增了一些算子，则不能从历史状态恢复，此时可以丢状态，并从 timestamp 或者 group-offsets 模式启动：

• 从 timestamp 时间戳恢复

适用于之前的作业没开过 checkpoint。需要从数据曲线-业务延迟监控里，找到当前消费延迟，反推出当前消费的时间点，从这个时间点恢复即可。
比如业务延迟是 1h，当前时刻是 12 点，则反推出已消费时间点是 11 点，为了避免丢数，则可以根据情况往前多消费一点，比如往前 30min，那么要设置的恢复的时刻就是 10:30。
参数如下：

'properties.scan.startup.mode' = 'timestamp',
'properties.scan.startup.timestamp-millis' = 'xxxx' -- 毫秒时间戳，可以从 https://tool.chinaz.com/tools/unixtime.aspx 获取

• 从 group offsets 恢复

适用于之前的作业开过 checkpoint，并完成过 checkpoint。由于 checkpoint 时，kafka source 会向 kafka 实例提交 offset，因此 offset 会保存一份到 kafka 实例。
此时需要把 kafka source 里的 startup mode 改为 group-offsets，参数如下：

'properties.scan.startup.mode' = 'group-offsets'

注意，如果这个消费者组之前没有 offset 信息，第一次启动的时候会报错无法获取 partition 的 offset 信息，此时需要设置如下 kafka 参数：

'properties.auto.offset.reset' = 'earliest'

此参数意义为如果没有 group-offset 则从最老的位置进行消费

自动提交 Offsets 机制

目前一般使用以下两种方式自动提交 Kafka Offsets。

• 方式 1：依赖 Flink 任务 Checkpoint。
  Flink 任务开启 Checkpoint 时，Kafka Source 在完成 Checkpoint 时会提交当前的消费位点，以保证 Flink 的 Checkpoint 状态和 Kafka Broker 上的提交位点一致。

  注意

  依赖 Flink 任务 Checkpoint 来管理 Kafka Offsets 时，如果上游数据量很大，很可能会触发上游的 LAG 告警。

• 方式 2：依赖 Kafka Consumer 的位点定时提交逻辑。
  当 Flink 任务没有开启 Checkpoint 时，Kafka Source 将依赖 Kafka Consumer 的位点定时提交逻辑。您可以通过设置enable.auto.commit和auto.commit.interval.ms两个参数来控制位点定时自动提交。

  -- 是否自动提交 Offsets。取值为 true 表示自动提交 Offsets；取值为 false，表示手动同步或异步提交。
    'enable.auto.commit' = 'true',
    -- 自动提交 Offsets 的时间间隔，单位为 ms。
    'auto.commit.interval.ms' = '500',