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

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提取器只能为您提供一个时间戳，此时间戳用于基于时间的操作，如窗口聚合或联接。似乎你没有做任何基于时间的计算思维，因此，从计算的角度来看，这并不重要。

请注意，一条记录只有一个元数据时间戳字段。此时间戳字段可用于存储可由生成者设置的事件时间戳。或者，您可以让代理使用代理引入时间覆盖创建者提供的时间戳。这是一个主题配置。

若要访问记录元数据时间戳(如果是事件时间或引入时间，则无关)，默认时间戳提取程序会提供此时间戳。如果要在应用程序中访问它，则需要使用处理器API，即，在您的情况下，使用.transform()而不是.flatMap运算符。您的Transformer将使用允许您访问提取的时间戳的context对象进行初始化。

由于记录只能存储一个元数据时间戳，并且您希望将其用于代理引入时间，因此上游创建者必须将事件时间戳直接放入有效负载中。

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