Chronon on Flink

Important: The Flink connector is an experimental feature that is still in the process of being open-sourced.

Chronon on Flink is an alternative to Chronon on Spark Streaming. It's intended for organizations that either don't have access to Spark Streaming or want to use The Tiled Architecture.

How to use the Flink connector

The process of integrating Flink will differ among organizations. The overall idea is simple: you need to integrate the FlinkJob so that it reads an event stream (e.g., Kafka) and writes out to a KV store.

There are two versions of the Flink app that you can choose from, tiled and untiled. See The Tiled Architecture for an overview of the differences. Briefly, the untiled version writes out events to the KV store, whereas the tiled version writes out pre-aggregates. In FlinkJob.scala you will find both options.

You will also likely need to modify your KVStore implementation while integrating Flink.

Configuration Properties

When running Flink jobs, you can configure various properties. These properties control the behavior of Flink execution.

kv_concurrency

Type: Integer
Required: No
Default: Default AsyncKVStoreWriter concurrency
Description: Controls the concurrency level for KV store writes, affecting throughput and resource usage.

trigger

Type: String
Required: No
Default: always_fire
Valid values: always_fire, buffered
Description: Controls when the Flink window triggers output:
- always_fire: Triggers on every incoming event (lower latency, higher write volume)
- buffered: Batches writes for efficiency (higher latency, reduced write volume)

start_offset

Type: Long (timestamp in milliseconds)
Required: No
Default: Uses committed offsets with LATEST strategy
Description: Specifies the timestamp from which to start consuming Kafka messages. Useful for catching up with historical data or reprocessing events from a specific point in time.

Overview of the Flink operators

The operators for the tiled and untiled Flink jobs differ slightly. The main difference is that the tiled job is stateful and contains a window operator. This section goes over the tiled version. See FlinkJob.scala for details on the untiled version.

The tiled Flink job

The Flink job contains five main operators

Source - Reads events of type T from a source, generally a Kafka topic. The generic type T could be a POJO, Scala case class, Thrift, Proto, etc.
- Note: currently, the Source and Job do not adhere to the mutation interface of chronon.
Spark expression evaluation - Evaluates the Spark SQL expression in the GroupBy and projects and filters the input data. This operator runs Spark inside the Flink app using CatalystUtil.
Window/tiling - This is the main tiling operator. It uses a window to aggregate incoming events and keep track of the IRs. It outputs the pre-aggregates on every event so they are written out to the KV store and the Fetcher has access to fresh values.
Avro conversion - Finishes Avro-converting the output of the window (the IRs) to a form that can be written out to the KV store (PutRequest object).
KV store sink - Writes the PutRequest objects to the KV store using the AsyncDataStream API.

End-to-end example

This example shows a simplified version of what happens to events as they move through the Flink operators.

Say we have a high-tech Ice Cream shop and we want to create an ML model. We want to define features for:

Counting the number of ice cream cones a person has bought in the last 6 hours.
Keeping track of the last ice cream flavor a person had in the last 6 hours.

A GroupBy might look like this:

ice_cream_group_by = GroupBy(
    sources=Source(
        events=ttypes.EventSource(
            query=Query(
                selects=select(
                    customer_id="customer_id",
                    flavor="ice_cream_flavor",
                ),
                time_column="created",
            )
        )
    ),
    keys=["customer_id"],
    aggregations=[
        Aggregation(
            input_column="customer_id",
            operation=Operation.COUNT,
            windows=[Window(length=6, timeUnit=TimeUnit.HOURS)],
        ),
        Aggregation(
            input_column="flavor",
            operation=Operation.LAST,
            windows=[Window(length=6, timeUnit=TimeUnit.HOURS)],
        ),
    ],
    accuracy=Accuracy.TEMPORAL,
    online=True
)

1. Source

The Source operator consumes events from Kafka and deserializes them into typed objects. For our Ice Cream shop example, we have this Proto:

IceCreamEventProto(
  customer_id: String,
  created: Long,
  ice_cream_flavor: String,
  ice_cream_cone_size: String
)

2. Spark expression evaluation

This operator transforms the object of type T based on the GroupBy's defined selects and filters, outputting a Map[String, Any].

// Input
IceCreamEventProto(
  customer_id = "Alice",
  created = 1000L,
  ice_cream_flavor = "chocolate",
  ice_cream_cone_size = "large" // Not used in the GroupBy definition
)
// Output
Map(
  "customer_id" -> "Alice",
  "created" -> 1000L,
  "flavor" -> "chocolate"
)

3. Window operator

This window operator pre-aggregates incoming Map(String -> Any) and produces an array of IRs. Example:

Event 1 Map("Alice", 1000L, "chocolate").

Pre-aggregates for key "Alice": [count: 1, last_flavor: "chocolate"]

Event 2 for "Bob": Map("Bob", 1200L, "strawberry")

Pre-aggregates for key "Bob": [count: 1, last_flavor: "strawberry"]

Event 3 for "Alice": Map("Alice", 1500L, "olive oil")

Pre-aggregates for key "Alice": [count: 2, last_flavor: "olive oil"]

4. Avro conversion

This operator uses Avro to finish encoding the array of IRs into bytes and creates a PutRequest.

Input: [2, "olive oil"]

Output: PutRequest(keyBytes: a927dcc=, valueBytes: d823eaa82==, ...) (varies depending on your specific KVStore implementation).

5. KVStore Sink

The final operator asynchronously writes the PutRequests to the KV store. These tiles are later decoded by the Fetcher and merged to calculate the final feature values.

Edit this page

Orchestrating Chronon

Project Setup