Kafka Stream

什么是Kafka Stream

Kafka Streams是一个用于处理和分析数据的客户端库。它先把存储在Kafka中的数据进行处理和分析，然后将最终所得的数据结果回写到Kafka或发送到外部系统去。它建立在一些非常重要的流式处理概念之上，例如：适当区分事件时间和处理时间、窗口支持，以及应用程序状态的简单（高效）管理。同时，它也基于Kafka中的许多概念，例如通过划分主题进行扩展。由于这个原因，它作为一个轻量级的库可以集成到应用程序中去。这个应用程序可以根据需要独立运行、在应用程序服务器中运行、作为Docker容器，或通过资源管理器（如Mesos）进行操作。通过利用Kafka的并行模型，Kafka Streams可以透明地处理同一应用程序的多个实例的负载平衡。
Kafka Streams的一些特点：

• 设计为简单轻量级的客户端库，可以轻松地将其嵌入任何Java应用程序中，并与用户为其流式应用程序所拥有的任何现有打包，部署和操作工具集成
• 除了Apache Kafka本身作为内部消息传递层之外，对系统没有任何外部依赖性
• 使用Kafka的分区模型来水平扩展处理
• 支持容错本地状态，该状态可以进行非常快速有效的状态操作，例如：窗口连接和聚合
• 支持一次精确的处理语义，以确保在处理过程中Streams客户端或Broker发生故障时，每条记录也只处理一次
• 采用一次一个记录的处理以实现毫秒级的处理延迟
• 支持基于事件时间的窗口操作以及记录的无序到达
• 提供必要的流处理原语，以及高级Streams DSL和低级Processor API

流处理示例

public class KafkaWordCount {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-application");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.56.105:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();
        // 从输入主题TextLinesTopic构造一个KStream
        // 其中消息值代表文字行
        KStream<String, String> textLines = builder.stream("TextLinesTopic");
        KTable<String, Long> wordCounts = textLines
                // 用空格将每行文本分割成多个单词
                .flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split("\\W+")))
                // 将文本单词分组为消息键
                .groupBy((key, word) -> word)
                // 计算每个单词（消息键）出现次数
                .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("counts-store"));
        // 将运行结果作为流存储到WordsWithCountsTopic主题
        wordCounts.toStream().to("WordCountTopic", Produced.with(Serdes.String(), Serdes.Long()));

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

示例程序将从TextLinesTopic主题中读取内容，对每个读取的消息执行WordCount算法的计算，并将其当前结果连续写入输出主题流WordCountTopic中。因此，除了日志条目外，将没有任何STDOUT输出，因为结果将回写到Kafka中。

核心概念

流

流是Kafka Streams提供的最重要的概念。它表示unbounded持续更新的数据集。流是有序的，可重放的，容错的，不可变的数据记录，其中的数据记录定义为键值对。

流处理程序

流处理程序是使用Kafka Streams库的任何程序。它通过一个或多个处理器拓扑定义计算逻辑，其中处理器拓扑是由流连接起来的流处理器图。

流处理器

流处理器是处理器拓扑结构中的一个节点。它表示一个处理步骤，通过一次从拓扑中的上游处理器接收输入记录并对其应用操作，来转换流中的数据，然后向下游处理器生成一个或多个输出记录。

特殊处理器

源处理器（Source）

源处理器是一种特殊类型的流处理器，没有任何上游处理器。它通过使用来自主题的记录并将它们转发到下游处理器，从而从一个或多个Kafka主题向拓扑生成输入流。

接收处理器（Sink）

接收器处理器也是一种特殊的流处理器，没有下游处理器。它将所有从上游处理器接收到的记录发送到指定的Kafka主题。

Kafka Streams提供了两种定义流处理拓扑的方法：

• Kafka Streams DSL使用Streams DSL构建一个处理器拓扑，开发者可以使用KStreamBuilder类，它是TopologyBuilder的扩展。DSL提供了最通用的数据转换操作。如开箱即用的map，filter，join和aggregation。
• Processor API低级处理器允许开发人员定义和连接自定义处理器以及与状态存储进行交互。

时间

在流处理方面有一个重要的时间概念，以及它是如何建模和集成。例如：一些操作，如基于时间界限定义的窗口。
时间在流中的常见概念如下：

• 事件时间：一个事件或数据记录发生的时间点
• 处理时间：事件或数据消息由流处理应用程序处理的时间点，即记录被消费的时间点
• 摄取时间：将事件或数据记录存储在Topic分区中的时间点

Kafka Streams通过TimestampExtractor接口为每条数据记录分配一个时间戳。这些每条记录的时间戳描述了流相对于时间的进度，并被和时间有关的操作（如：窗口操作）所使用。我们将此数据驱动时间称为应用程序的“流时间”，以与该应用程序实际执行时的时间（wall-time）区分开。在这之后，TimestampExtractor的具体实现会为流时间定义提供不同的语义。因此，开发人员可以根据其业务需求实施不同的时间操作。如：每条记录的时间戳描述了流的时间增长（尽管记录在stream中是无序的）并利用时间依赖性来进行操作。
最后，每当Kafka Streams应用程序将记录写入Kafka时，还将为这些新记录分配时间戳。时间戳的分配方式取决于上下文：

• 当通过处理某些输入记录生成新的输出记录时，输出记录会直接从输入记录中继承时间戳
• 当通过周期性函数生成新的输出记录时，输出记录的时间戳为流任务的当前内部时间
• 对于聚合，结果记录的时间戳将是对结果有贡献的所有输入记录的最大时间戳

聚合

聚合操作采用一个输入流或表，并通过将多个输入记录合并为一个输出记录来产生一个新表。

窗口

流式数据是在时间上无界的数据。而聚合操作只能作用在特定的数据集，也即有界的数据集上。因此需要通过某种方式从无界的数据集上按特定的语义选取出有界的数据。窗口是一种非常常用的设定计算边界的方式。
Streams DSL中提供了窗口操作。使用窗口时，可以为窗口指定期限。该期限控制Kafka Streams将等待给定窗口多长时间的无序数据记录。如果在窗口期限之后到达记录，则该记录将被丢弃，并且不会在该窗口中进行处理。

Kafka Stream API编程

核心数据抽象

KStream

只有DSL中有KStream的概念。KStream是一个流式记录的抽象，是一个无界的数据集。用一个表来类比，数据记录在一个流中可以理解为一直在进行“INSERT”的动作。只进行追加。

KTable

只有DSL中有KTable的概念。KTable是一个changelog stream的抽象，每个数据记录都被表示为一个update。如果key在KTable中已经存在，则表示为一个“UPDATE”；如果不存在，则表示为一个“INSERT”。

GlobalKTable

只有DSL中有GlobalKTable的概念。和KTable一样，GlobalKTable也是一个changelog stream的抽象，每个数据记录都被表示为一个update。
KTable存储的数据根据key进行分区的，GlobalKTable是不分区的，且存储的数据足够小，能完全装入内存，因此保证每个流任务都有所有数据的完整副本，而不关心传入record的key是什么。基于以上的特性，GlobalKTable有如下优点：

• 更高效的join操作：当链接多个join操作时，使用GlobalKTable效率更高，它不需要co-partitioned（类似shuffle的一种操作）的发生
• 可用于将信息“广播”到应用程序的所有实例

程序配置

在使用Streams之前，必须先配置Kafka和Kafka Streams配置选项。可以通过在java.util.Properties实例中指定参数来配置Kafka流。

前两项为程序配置必须项

基本函数试用示例

Table to Stream

将表转换为流

StreamsBuilder builder = new StreamsBuilder();
KTable<String, String> table = builder.table("topic-02");;
KStream<String, String> stream = table.toStream();
1
2
3

Stream to Table

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> kstream = builder.stream("topic-02");
KTable<String, String> table = stream.toTable();
1
2
3

无状态转换

无状态转换不需要进行状态处理，也不需要与流处理器有关的状态存储。

Branch

根据提供的谓词将KStream分支（或拆分）为一个或多个KStream实例

public class StreamBranch {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "branch-application");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.56.105:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> kstream = builder.stream("topic-02");
        KStream<String, String>[] branches = kstream.branch(
                // 第一个谓词
                (key, value) -> key.startsWith("keyA"),
                // 第二个谓词
                (key, value) -> key.startsWith("keyB")
        );

        for (KStream<String, String> stream : branches) {
            stream.foreach(new ForeachAction<String, String>() {
                @Override
                public void apply(String key, String value) {
                    System.out.println(key + " => " + value);
                }
            });
        }

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

Filter

过滤流中的元素

public class StreamFilter {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "filter-application0");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.56.105:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> kstream = builder.stream("topic-02");
        kstream.filter(new Predicate<String, String>() {
            @Override
            public boolean test(String key, String value) {
                return key.startsWith("keyB");
            }
        }).foreach(new ForeachAction<String, String>() {
            @Override
            public void apply(String key, String value) {
                System.out.println(key + " => " + value);
            }
        });

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

FlatMap

取得一条记录并由此产生零个，一个或多个记录，同时还可以修改记录键和值，包括类型。

public class StreamFlatmap {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "flatmap-application");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.56.105:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> kstream = builder.stream("topic-02");
        KStream<String, Integer> transformed = kstream.flatMap(
                (key, value) -> {
                    List<KeyValue<String, Integer>> result = new LinkedList<>();
                    result.add(KeyValue.pair(value.toUpperCase(), 1000));
                    result.add(KeyValue.pair(value.toLowerCase(), 9000));
                    return result;
                }
        );

        transformed.foreach((key, value) -> System.out.println(key + " => " + value));
        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

GroupByKey

按键对记录进行分组

public class StreamGroupByKey {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "groupbykey-application");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.56.105:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> kstream = builder.stream("topic-02");
        KGroupedStream<String, String> groupedStream = kstream.groupByKey(
                Grouped.with(Serdes.String(), Serdes.String())
        );

        groupedStream.count().toStream().foreach((key, value) -> System.out.println(key + " => " + value));
        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

GroupBy

通过新的Key对记录进行分组，新Key可能是不同的Key类型。进行分组时，还可以指定新的Value值和Value类型

public class StreamGroupby {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "groupby-application");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.56.105:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> kstream = builder.stream("topic-02");
        KTable<String, String> table = kstream.toTable();

        // 按新Key和其类型对流进行分组
        KGroupedStream<String, String> groupedStream = kstream.groupBy(
                new KeyValueMapper<String, String, String>() {
                    @Override
                    public String apply(String key, String value) {
                        return value;
                    }
                },
                Grouped.with(
                        // key 可以被修改
                        Serdes.String(),
                        Serdes.String())
        );

        // 按新的Key和其类型对表进行分组，同时修改值和值类型
        KGroupedTable<String, String> groupedTable = table.groupBy(
                new KeyValueMapper<String, String, KeyValue<String, String>>() {
                    @Override
                    public KeyValue<String, String> apply(String key, String value) {
                        return KeyValue.pair(key, key + "," + value);
                    }
                },
                Grouped.with(
                        // 类型可以被修改
                        Serdes.String(),
                        Serdes.String())
        );

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

有状态转换

有状态转换依赖处理输入和产生输出的状态，并且需要与流处理器有关的状态存储。如：在聚合操作中，窗口状态存储用于收集每个窗口的最新聚合结果。在联接操作中，窗口状态存储用来收集截至目前在定义的窗口边界内接收到的所有记录。

Left-Join

public class StreamLeftJoin {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "leftjoin-application");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.56.105:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> left = builder.stream("topic-03");
        KStream<String, String> right = builder.stream("topic-04");

        KStream<String, String> joined = left.selectKey(((key, value) -> value.split(",")[0]))
                .leftJoin(right.selectKey((key, value) -> value.split(",")[0]),
                (leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue, /* ValueJoiner */
                JoinWindows.of(Duration.ofMinutes(5)),
                Joined.with(
                        // key
                        Serdes.String(),
                        // left value
                        Serdes.String(),
                        // right value
                        Serdes.String())
        );

        joined.foreach((key, value) -> System.out.println(key + " => " + value));
        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30