如何提取 Kafka Streams 中消息中嵌入的时间戳

Question

我想提取每条消息中嵌入的时间戳并将它们作为 json 有效负载发送到我的数据库中。

我想获取以下三个时间戳。

活动时间：

The point in time when an event or data record occurred, i.e. was originally created “by the source”.

处理时间：

The point in time when the event or data record happens to be processed by the stream processing application, i.e. when the record is being consumed.

摄入时间：

The point in time when an event or data record is stored in a topic partition by a Kafka broker.

这是我的流应用程序代码：

Properties props = new Properties();

props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-pipe");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_URL + ":9092"); // pass from env localhost:9092 ,BROKER_URL + ":9092"
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

final StreamsBuilder builder = new StreamsBuilder();

KStream<String, String> source_o365_user_activity = builder.stream("o365_user_activity");

source_o365_user_activity.flatMapValues(new ValueMapper<String, Iterable<String>>() {
    @Override
    public Iterable<String> apply(String value) {
        System.out.println("========> o365_user_activity_by_date Log:     " + value);
        ArrayList<String> keywords = new ArrayList<String>();
        try {
            JSONObject send = new JSONObject();
            JSONObject received = new JSONObject(value);

            send.put("current_date", getCurrentDate().toString()); // UTC TIME
            send.put("activity_time", received.get("CreationTime")); // CONSTANTS FINAL STATIC(Topic Names, Cassandra keys)
            send.put("user_id", received.get("UserId"));
            send.put("operation", received.get("Operation"));
            send.put("workload", received.get("Workload"));
            keywords.add(send.toString());

        } catch (Exception e) {
            // TODO: handle exception
            System.err.println("Unable to convert to json");
            e.printStackTrace();
        }

        return keywords;
    }
}).to("o365_user_activity_by_date");

在代码中，我只是获取每条记录，对其进行一些流处理并将其发送到不同的主题。

现在，我想发送嵌入有效负载中的每条记录

Event-time

、

Processing-time

和

Ingestion-time

。

我看过

FailOnInvalidTimestamp

和

WallclockTimestampExtractor

，但我对如何使用它们感到困惑。

请指导我如何实现这一目标。

Answer 1

Timestamp

提取器只能为您提供一个时间戳，该时间戳用于基于时间的操作，例如窗口聚合或连接。看来你没有做任何基于时间的计算思想，因此，从计算的角度来看，这并不重要。

注意，一条记录只有一个元数据时间戳字段。该时间戳字段可用于存储可由生产者设置的事件时间戳（即应用程序代码可以设置它，否则生产者将默认为

System.currentTimeMillis()

）。作为替代方案，您可以让代理使用代理摄取时间覆盖生产者提供的时间戳。这是一个主题配置。

要访问记录元数据时间戳（无论是事件时间还是摄取时间，都是独立的），请使用默认时间戳提取器来为您提供此时间戳。

新版本的 Kafka（2.7+）

如果您想在应用程序中访问它，则需要使用处理器 API，即在您的情况下使用

.process()

而不是

.flatMap

运算符。在

Processor

内，您可以获得

Record

对象，这些对象允许您访问记录的时间戳。

旧版本的 Kafka (<= 2.6)

如果您想在应用程序中访问它，则需要使用处理器 API，即在您的情况下使用

.transform()

而不是

.flatMap

运算符。您的

Transformer

将使用

context

对象进行初始化，该对象允许您访问提取的时间戳。

因为一条记录只能存储一个元数据时间戳，并且您想将其用于代理摄取时间，所以上游生产者必须将事件时间戳直接放入有效负载中。

新版本的 Kafka（3.0+）

对于处理时间，您可以使用传递到

context

的 init 方法中的

Processor

对象，使用

#currentSystemTimeMs()

。

新版本的 Kafka (<= 2.8)

对于处理时间，只需执行代码片段中所示的系统调用即可。