Is there an equivalent to Kafka's KTable in Apache Flink?

user_2620716 0
Viewed 1111

Apache Kafka has a concept of a KTable, where

where each data record represents an update

Essentially, I can consume a kafka topic, and only keep the latest message for per key.

Is there a similar concept available in Apache Flink? I have read about Flink's Table API, but does not seem to be solving the same problem.

Some help comparing and contrasting the 2 frameworks would be helpful. I am not looking for which is better or worse. But rather just how they differ. The answer for which is right would then depend on my requirements.

apache-kafka-streams apache-flink
1 Answers

You are right. Flink's Table API and its Table class do not correspond to Kafka's KTable. The Table API is a relational language-embedded API (think of SQL integrated in Java and Scala).

Flink's DataStream API does not have a built-in concept that corresponds to a KTable. Instead, Flink offers sophisticated state management and a KTable would be a regular operator with keyed state.

For example, a stateful operator with two inputs that stores the latest value observed from the first input and joins it with values from the second input, can be implemented with a CoFlatMapFunction as follows:

DataStream<Tuple2<Long, String>> first = ...
DataStream<Tuple2<Long, String>> second = ...

DataStream<Tuple2<String, String>> result = first
  // connect first and second stream
  .connect(second)
  // key both streams on the first (Long) attribute
  .keyBy(0, 0)
  // join them
  .flatMap(new TableLookup());

// ------

public static class TableLookup 
  extends RichCoFlatMapFunction<Tuple2<Long,String>, Tuple2<Long,String>, Tuple2<String,String>> {

  // keyed state
  private ValueState<String> lastVal;

  @Override
  public void open(Configuration conf) {
    ValueStateDescriptor<String> valueDesc = 
      new ValueStateDescriptor<String>("table", Types.STRING);
    lastVal = getRuntimeContext().getState(valueDesc);
  }

  @Override
  public void flatMap1(Tuple2<Long, String> value, Collector<Tuple2<String, String>> out) throws Exception {
    // update the value for the current Long key with the String value.
    lastVal.update(value.f1);
  }

  @Override
  public void flatMap2(Tuple2<Long, String> value, Collector<Tuple2<String, String>> out) throws Exception {
    // look up latest String for current Long key.
    String lookup = lastVal.value();
    // emit current String and looked-up String
    out.collect(Tuple2.of(value.f1, lookup));
  }
}

In general, state can be used very flexibly with Flink and let's you implement a wide range of use cases. There are also more state types, such as ListState and MapState and with a ProcessFunction you have fine-grained control over time, for example to remove the state of a key if it has not been updated for a certain amount of time (KTables have a configuration for that as far as I know).

user_36095710
Related