Kafka Streams开发入门(9)

1. 背景

上一篇介绍了如何利用Kafka Streams实时统计某年最卖座和最不卖座的电影票房。主要的方法是通过Streams提供的aggregate方法实现了max/min算子。今天我为大家带来时间窗口函数的使用方法。在Kafka Streams中，时间窗口有三类：固定时间窗口（Tumbling Window）、滑动时间窗口（Sliding Window）和会话窗口（Session Window）。我们不详细讨论这三类窗口的定义与区别，而是直接使用一个项目来说明如何利用Tumbling Window定期统计每部电影的打分人数。

2. 功能演示说明

这篇文章中我们会创建一个Kafka topic来表示电影打分事件。然后我们编写一个程序统计每个时间窗口下每部电影接收到的打分人数。我们依然使用ProtocolBuffer对消息事件进行序列化。事件的JSON格式如下所示：

{"title": "Die Hard", "release_year": 1998, "rating": 8.2, "timestamp": "2019-04-25T18:00:00-0700"}

字段含义一目了然，不再赘述了。

整个程序实时统计不同时间窗口下的电影打分人数，比如输出是这样的：

[Die Hard@1556186400000/1556187000000]	1
[Die Hard@1556186400000/1556187000000]	2
[Die Hard@1556186400000/1556187000000]	3
[Die Hard@1556186400000/1556187000000]	4
[Die Hard@1556188200000/1556188800000]	1

上面输出表示在第一个窗口下Die Hard这部电影依次接收到3次打分，在第二个窗口下Die Hard接收了1次打分。

3. 配置项目

第1步是创建项目功能所在路径，命令如下：

$ mkdir tumbling-windows && cd tumbling-windows

然后在新创建的tumbling-windows路径下新建Gradle配置文件build.gradle，内容如下：

buildscript {
   
    repositories {
        jcenter()
    }
    dependencies {
        classpath 'com.github.jengelman.gradle.plugins:shadow:4.0.2'
    }
}
   
plugins {
    id 'java'
    id "com.google.protobuf" version "0.8.10"
}
apply plugin: 'com.github.johnrengelman.shadow'
   
   
repositories {
    mavenCentral()
    jcenter()
   
    maven {
        url 'http://packages.confluent.io/maven'
    }
}
   
group 'huxihx.kafkastreams'
   
sourceCompatibility = 1.8
targetCompatibility = '1.8'
version = '0.0.1'
   
dependencies {
    implementation 'com.google.protobuf:protobuf-java:3.11.4'
    implementation 'org.slf4j:slf4j-simple:1.7.26'
    implementation 'org.apache.kafka:kafka-streams:2.4.0'
   
    testCompile group: 'junit', name: 'junit', version: '4.12'
}
   
protobuf {
    generatedFilesBaseDir = "$projectDir/src/"
    protoc {
        artifact = 'com.google.protobuf:protoc:3.11.4'
    }
}
   
jar {
    manifest {
        attributes(
                'Class-Path': configurations.compile.collect { it.getName() }.join(' '),
                'Main-Class': 'huxihx.kafkastreams.TumblingWindow'
        )
    }
}
   
shadowJar {
    archiveName = "kstreams-tumbling-windows-standalone-${version}.${extension}"
}　　

项目工程中的主类是huxihx.kafkastreams.TumblingWindow。

保存上面的文件，然后执行下列命令下载Gradle的wrapper套件：

$ gradle wrapper

做完这些之后，我们在tumbling-windows目录下创建名为configuration的子目录，用于保存我们的参数配置文件dev.properties：

$ mkdir configuration
$ cd configuration
$ vi dev.properties

dev.properties内容如下：

application.id=tumbling-window-app

bootstrap.servers=localhost:9092

rating.topic.name=ratings
rating.topic.partitions=1
rating.topic.replication.factor=1

rating.count.topic.name=rating-counts
rating.count.topic.partitions=1
rating.count.topic.replication.factor=1

这里我们创建了一个输入topic：ratings和一个输出topic：rating-counts。前者表示电影打分事件，后者保存每部电影在时间窗口下的打分次数。

4. 创建消息Schema

由于我们使用ProtocolBuffer进行序列化，因此我们要提前生成好Java类来建模实体消息。我们在tumbling-windows路径下执行以下命令创建保存schema的文件夹：

$ mkdir -p src/main/proto && cd src/main/proto

之后在proto文件夹下创建名为rating.proto文件，内容如下：

syntax = "proto3";
   
package huxihx.kafkastreams.proto;
   
message Rating {
    string title = 1;
    int32 release_year = 2;
    double rating = 3;
    string timestamp = 4;
}　

由于输出格式很简单，因此这次我们不为输出topic做单独的schema类生成了。保存上面的文件之后在tumbling-windows目录下运行gradlew命令：

./gradlew build

此时，你应该可以在tumbling-windows的src/main/java/huxihx/kafkastreams/proto下看到生成的Java类：RatingOuterClass。

5. 创建Serdes

这一步我们为所需的topic消息创建Serdes。首先在tumbling-windows目录下执行下面的命令创建对应的文件夹目录：

$ mkdir -p src/main/java/huxihx/kafkastreams/serdes　　

在新创建的serdes文件夹下创建ProtobufSerializer.java，内容如下：

package huxihx.kafkastreams.serdes;
    
import com.google.protobuf.MessageLite;
import org.apache.kafka.common.serialization.Serializer;
    
public class ProtobufSerializer<T extends MessageLite> implements Serializer<T> {
    @Override
    public byte[] serialize(String topic, T data) {
        return data == null ? new byte[0] : data.toByteArray();
    }
}　　

接下来是创建ProtobufDeserializer.java：

package huxihx.kafkastreams.serdes;
    
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.MessageLite;
import com.google.protobuf.Parser;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Deserializer;
    
import java.util.Map;
    
public class ProtobufDeserializer<T extends MessageLite> implements Deserializer<T> {
    
    private Parser<T> parser;
    
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        parser = (Parser<T>) configs.get("parser");
    }
    
    @Override
    public T deserialize(String topic, byte[] data) {
        try {
            return parser.parseFrom(data);
        } catch (InvalidProtocolBufferException e) {
            throw new SerializationException("Failed to deserialize from a protobuf byte array.", e);
        }
    }
}

最后是ProtobufSerdes.java：

package huxihx.kafkastreams.serdes;
    
import com.google.protobuf.MessageLite;
import com.google.protobuf.Parser;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serializer;
    
import java.util.HashMap;
import java.util.Map;
    
public class ProtobufSerdes<T extends MessageLite> implements Serde<T> {
    
    private final Serializer<T> serializer;
    private final Deserializer<T> deserializer;
    
    public ProtobufSerdes(Parser<T> parser) {
        serializer = new ProtobufSerializer<>();
        deserializer = new ProtobufDeserializer<>();
        Map<String, Parser<T>> config = new HashMap<>();
        config.put("parser", parser);
        deserializer.configure(config, false);
    }
    
    @Override
    public Serializer<T> serializer() {
        return serializer;
    }
    
    @Override
    public Deserializer<T> deserializer() {
        return deserializer;
    }
}

6. 开发主流程

对于时间窗口而言，你必须要定义一个TimestampExtractor告诉Kafka Streams如何确定时间维度。本例中我们需要创建一个TimestampExtractor来提取事件中的时间信息。我们在huxihx.kafkastreams下创建一个名为RatingTimestampExtractor.java的文件：

package huxihx.kafkastreams;

import huxihx.kafkastreams.proto.RatingOuterClass;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.streams.processor.TimestampExtractor;

import java.text.ParseException;
import java.text.SimpleDateFormat;

public class RatingTimestampExtractor implements TimestampExtractor {

    @Override
    public long extract(ConsumerRecord<Object, Object> record, long partitionTime) {
        final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss");
        String eventTime = ((RatingOuterClass.Rating)record.value()).getTimestamp();
        try {
            return sdf.parse(eventTime).getTime();
        } catch (ParseException e) {
            return 0;
        }
    }
}　　

上面代码利用SimpleDateFormat做时间格式转换，将字符串形式的时间转换成一个时间戳返回。

接下来，我们编写主程序：TumblingWindow.java：

package huxihx.kafkastreams;

import huxihx.kafkastreams.proto.RatingOuterClass;
import huxihx.kafkastreams.serdes.ProtobufSerdes;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.admin.TopicListing;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.Grouped;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.Windowed;

import java.io.FileInputStream;
import java.io.IOException;
import java.nio.file.Files;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.TimeZone;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

public class TumblingWindow {

    public static void main(String[] args) throws Exception {
        if (args.length < 1) {
            throw new IllegalArgumentException("This program takes one argument: the path to an environment configuration file.");
        }

        new TumblingWindow().runRecipe(args[0]);
    }

    private Properties loadEnvProperties(String fileName) throws IOException {
        Properties envProps = new Properties();
        try (FileInputStream input = new FileInputStream(fileName)) {
            envProps.load(input);
        }
        return envProps;
    }

    private void runRecipe(final String configPath) throws Exception {
        Properties envProps = this.loadEnvProperties(configPath);
        Properties streamProps = this.createStreamsProperties(envProps);

        Topology topology = this.buildTopology(envProps);
        this.preCreateTopics(envProps);

        final KafkaStreams streams = new KafkaStreams(topology, streamProps);
        final CountDownLatch latch = new CountDownLatch(1);

        // Attach shutdown handler to catch Control-C.
        Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") {
            @Override
            public void run() {
                streams.close();
                latch.countDown();
            }
        });

        try {
            streams.start();
            latch.await();
        } catch (Throwable e) {
            System.exit(1);
        }
        System.exit(0);

    }


    private Topology buildTopology(final Properties envProps) {
        final StreamsBuilder builder = new StreamsBuilder();
        final String ratingTopic = envProps.getProperty("rating.topic.name");
        final String ratingCountTopic = envProps.getProperty("rating.count.topic.name");

        builder.stream(ratingTopic, Consumed.with(Serdes.String(), ratingProtobufSerdes()))
                .map((key, rating) -> new KeyValue<>(rating.getTitle(), rating))
                .groupByKey(Grouped.with(Serdes.String(), ratingProtobufSerdes()))
                .windowedBy(TimeWindows.of(Duration.ofMinutes(10)))
                .count()
                .toStream()
                .<String, String>map((Windowed<String> key, Long count) -> new KeyValue(windowedKeyToString(key), count.toString()))
                .to(ratingCountTopic);

        return builder.build();
    }


    private static void preCreateTopics(Properties envProps) throws Exception {
        Map<String, Object> config = new HashMap<>();
        config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, envProps.getProperty("bootstrap.servers"));
        String inputTopic = envProps.getProperty("rating.topic.name");
        String outputTopic = envProps.getProperty("rating.count.topic.name");
        Map<String, String> topicConfigs = new HashMap<>();
        topicConfigs.put("retention.ms", Long.toString(Long.MAX_VALUE));


        try (AdminClient client = AdminClient.create(config)) {
            Collection<TopicListing> existingTopics = client.listTopics().listings().get();

            List<NewTopic> topics = new ArrayList<>();
            List<String> topicNames = existingTopics.stream().map(TopicListing::name).collect(Collectors.toList());
            if (!topicNames.contains(inputTopic))
                topics.add(new NewTopic(
                        inputTopic,
                        Integer.parseInt(envProps.getProperty("rating.topic.partitions")),
                        Short.parseShort(envProps.getProperty("rating.topic.replication.factor"))).configs(topicConfigs));

            if (!topicNames.contains(outputTopic))
                topics.add(new NewTopic(
                        outputTopic,
                        Integer.parseInt(envProps.getProperty("rating.count.topic.partitions")),
                        Short.parseShort(envProps.getProperty("rating.count.topic.replication.factor"))).configs(topicConfigs));

            if (!topics.isEmpty())
                client.createTopics(topics).all().get();
        }
    }

    private Properties createStreamsProperties(Properties envProps) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, envProps.getProperty("application.id"));
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, envProps.getProperty("bootstrap.servers"));
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, RatingTimestampExtractor.class.getName());
        props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
        try {
            props.put(StreamsConfig.STATE_DIR_CONFIG,
                    Files.createTempDirectory("tumbling-windows").toAbsolutePath().toString());
        } catch (IOException ignored) {
        }
        return props;
    }

    private String windowedKeyToString(Windowed<String> key) {
        return String.format("[%s@%s/%s]", key.key(), key.window().start(), key.window().end());
    }

    private static ProtobufSerdes<RatingOuterClass.Rating> ratingProtobufSerdes() {
        return new ProtobufSerdes<>(RatingOuterClass.Rating.parser());
    }
}

7. 编写测试Producer

现在创建src/main/java/huxihx/kafkastreams/tests/TestProducer.java和TestConsumer.java用于测试，内容分别如下：　　

package huxihx.kafkastreams.tests;

import huxihx.kafkastreams.proto.RatingOuterClass;
import huxihx.kafkastreams.serdes.ProtobufSerializer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Arrays;
import java.util.List;
import java.util.Properties;

public class TestProducer {
    private static final List<RatingOuterClass.Rating> TEST_EVENTS = Arrays.asList(
            RatingOuterClass.Rating.newBuilder().setTitle("Die Hard").setReleaseYear(1998).setRating(8.2)
                    .setTimestamp("2019-04-25T18:00:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("Die Hard").setReleaseYear(1998).setRating(4.5)
                    .setTimestamp("2019-04-25T18:03:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("Die Hard").setReleaseYear(1998).setRating(5.1)
                    .setTimestamp("2019-04-25T18:04:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("Die Hard").setReleaseYear(1998).setRating(2.0)
                    .setTimestamp("2019-04-25T18:07:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("Die Hard").setReleaseYear(1998).setRating(8.3)
                    .setTimestamp("2019-04-25T18:32:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("Die Hard").setReleaseYear(1998).setRating(3.4)
                    .setTimestamp("2019-04-25T18:36:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("Die Hard").setReleaseYear(1998).setRating(4.2)
                    .setTimestamp("2019-04-25T18:43:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("Die Hard").setReleaseYear(1998).setRating(7.6)
                    .setTimestamp("2019-04-25T18:44:00-0700").build(),

            RatingOuterClass.Rating.newBuilder().setTitle("Tree of Life").setReleaseYear(2011).setRating(4.9)
                    .setTimestamp("2019-04-25T20:01:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("Tree of Life").setReleaseYear(2011).setRating(5.6)
                    .setTimestamp("2019-04-25T20:02:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("Tree of Life").setReleaseYear(2011).setRating(9.0)
                    .setTimestamp("2019-04-25T20:03:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("Tree of Life").setReleaseYear(2011).setRating(6.5)
                    .setTimestamp("2019-04-25T20:12:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("Tree of Life").setReleaseYear(2011).setRating(2.1)
                    .setTimestamp("2019-04-25T20:13:00-0700").build(),


            RatingOuterClass.Rating.newBuilder().setTitle("A Walk in the Clouds").setReleaseYear(1995).setRating(3.6)
                    .setTimestamp("2019-04-25T22:20:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("A Walk in the Clouds").setReleaseYear(1995).setRating(6.0)
                    .setTimestamp("2019-04-25T22:21:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("A Walk in the Clouds").setReleaseYear(1995).setRating(7.0)
                    .setTimestamp("2019-04-25T22:22:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("A Walk in the Clouds").setReleaseYear(1995).setRating(4.6)
                    .setTimestamp("2019-04-25T22:23:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("A Walk in the Clouds").setReleaseYear(1995).setRating(7.1)
                    .setTimestamp("2019-04-25T22:24:00-0700").build(),


            RatingOuterClass.Rating.newBuilder().setTitle("A Walk in the Clouds").setReleaseYear(1998).setRating(9.9)
                    .setTimestamp("2019-04-25T21:15:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("A Walk in the Clouds").setReleaseYear(1998).setRating(8.9)
                    .setTimestamp("2019-04-25T21:16:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("A Walk in the Clouds").setReleaseYear(1998).setRating(7.9)
                    .setTimestamp("2019-04-25T21:17:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("A Walk in the Clouds").setReleaseYear(1998).setRating(8.9)
                    .setTimestamp("2019-04-25T21:18:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("A Walk in the Clouds").setReleaseYear(1998).setRating(9.9)
                    .setTimestamp("2019-04-25T21:19:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("A Walk in the Clouds").setReleaseYear(1998).setRating(9.9)
                    .setTimestamp("2019-04-25T21:20:00-0700").build(),

            RatingOuterClass.Rating.newBuilder().setTitle("Super Mario Bros.").setReleaseYear(1993).setRating(3.5)
                    .setTimestamp("2019-04-25T13:00:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("Super Mario Bros.").setReleaseYear(1993).setRating(4.5)
                    .setTimestamp("2019-04-25T13:07:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("Super Mario Bros.").setReleaseYear(1993).setRating(5.5)
                    .setTimestamp("2019-04-25T13:30:00-0700").build(),
            RatingOuterClass.Rating.newBuilder().setTitle("Super Mario Bros.").setReleaseYear(1993).setRating(6.5)
                    .setTimestamp("2019-04-25T13:34:00-0700").build());

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, new ProtobufSerializer<RatingOuterClass.Rating>().getClass());

        try (final Producer<String, RatingOuterClass.Rating> producer = new KafkaProducer<>(props)) {
            TEST_EVENTS.stream().map(event ->
                    new ProducerRecord<String, RatingOuterClass.Rating>("ratings", event)).forEach(producer::send);
        }
    }
}

8. 测试

首先我们运行下列命令构建项目：

$ ./gradlew shadowJar　　

然后启动Kafka集群，之后运行Kafka Streams应用：

$ java -jar build/libs/kstreams-transform-standalone-0.0.1.jar configuration/dev.properties

现在启动一个终端测试Producer：

$ java -cp build/libs/kstreams-transform-standalone-0.0.1.jar huxihx.kafkastreams.tests.TestProducer

之后打开另一个终端，运行一个ConsoleConsumer命令验证输出结果：

$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic rating-counts --from-beginning --property print.key=true

 如果一切正常的话，你应该可以看见如下输出：

[Die Hard@1556186400000/1556187000000]	1
[Die Hard@1556186400000/1556187000000]	2
[Die Hard@1556186400000/1556187000000]	3
[Die Hard@1556186400000/1556187000000]	4
[Die Hard@1556188200000/1556188800000]	1
[Die Hard@1556188200000/1556188800000]	2
[Die Hard@1556188800000/1556189400000]	1
[Die Hard@1556188800000/1556189400000]	2
[Tree of Life@1556193600000/1556194200000]	1
[Tree of Life@1556193600000/1556194200000]	2
[Tree of Life@1556193600000/1556194200000]	3
[Tree of Life@1556194200000/1556194800000]	1
[Tree of Life@1556194200000/1556194800000]	2
[A Walk in the Clouds@1556202000000/1556202600000]	1
[A Walk in the Clouds@1556202000000/1556202600000]	2
[A Walk in the Clouds@1556202000000/1556202600000]	3
[A Walk in the Clouds@1556202000000/1556202600000]	4
[A Walk in the Clouds@1556202000000/1556202600000]	5
[A Walk in the Clouds@1556197800000/1556198400000]	1
[A Walk in the Clouds@1556197800000/1556198400000]	2
[A Walk in the Clouds@1556197800000/1556198400000]	3
[A Walk in the Clouds@1556197800000/1556198400000]	4
[A Walk in the Clouds@1556197800000/1556198400000]	5
[A Walk in the Clouds@1556198400000/1556199000000]	1
[Super Mario Bros.@1556168400000/1556169000000]	1
[Super Mario Bros.@1556168400000/1556169000000]	2
[Super Mario Bros.@1556170200000/1556170800000]	1
[Super Mario Bros.@1556170200000/1556170800000]	2

中括号中的文字是电影名称，第一个时间戳是该窗口起始时间，第二个时间戳是窗口结束时间。