Idempotent Writes

An idempotent operation can be performed several times but will only result in a single change. For example, repeatedly inserting the same key-value pair into a hashmap is an idempotent operation because the first insert operation adds the value for the key into the map and all following insertions will not change the map since it already contains the key-value pair. On the other hand, an append operation is not an idempotent operation, because appending an element multiple times results in multiple appends. Idempotent write operations are interesting for streaming applications because they can be performed multiple times without changing the results. Hence, they can to some extent mitigate the effect of replayed results as caused by Flink’s checkpointing mechanism.

It should be noted an application that relies on idempotent sinks to achieve exactly-once results must guarantee that it overrides previously written results while it replays. For example, an application with a sink that upserts into a key-value store must ensure that it deterministically computes the keys that are used to upsert. Moreover, applications that read from the sink system might observe unexpected results during the time when an application recovers. When the replay starts, previously emitted results might be overridden by earlier results. Hence, an application that consumes the output of the recovering application might witness a jump back in time, e.g., read a smaller count than before. Also, the overall result of the streaming application will be in an inconsistent state while the replay is in progress because some results will be overridden while others are not. Once the replay completes and the application is past the point at which it previously failed, the result will be consistent again.

Transactional Writes

The second approach to achieve end-to-end exactly-once consistency is based on transactional writes. The idea here is to only write those results to an external sink system that have been computed before the last successful checkpoint. This behavior guarantees end-to-end exactly-once because in case of a failure, the application is reset to the last checkpoint and no results have been emitted to the sink system after that checkpoint. By only writing data once a checkpoint is completed, the transactional approach does not suffer from the replay inconsistency of the idempotent writes. However, it adds latency because results only become visible when a checkpoint completes.

Flink provides two building blocks to implement transactional sink connectors—a generic write-ahead-log (WAL) sink and a two-phase-commit (2PC) sink. The WAL sink writes all result records into application state and emits them to the sink system once it receives the notification that a checkpoint was completed. Since the sink buffers records in the state backend, the WAL sink can be used with any kind of sink system. However, it cannot provide bulletproof exactly-once guarantees,² adds to the state size of an application, and the sink system has to deal with a spiky writing pattern.

In contrast, the 2PC sink requires a sink system that offers transactional support or exposes building blocks to emulate transactions. For each checkpoint, the sink starts a transaction and appends all received records to the transaction, writing them to the sink system without committing them. When it receives the notification that a checkpoint completed, it commits the transaction and materializes the written results. The mechanism relies on the ability of a sink to commit a transaction after recovering from a failure that was opened before a completed checkpoint.

The 2PC protocol piggybacks on Flink’s existing checkpointing mechanism. The checkpoint barriers are notifications to start a new transaction, the notifications of all operators about the success of their individual checkpoint are their commit votes, and the messages of the JobManager that notify about the success of a checkpoint are the instructions to commit the transactions. In contrast to WAL sinks, 2PC sinks can achieve exactly-once output depending on the sink system and the sink’s implementation. Moreover, a 2PC sink continuously writes records to the sink system compared to the spiky writing pattern of a WAL sink.

Table 8-1 shows the end-to-end consistency guarantees for different types of source and sink connectors that can be achieved in the best case; depending on the implementation of the sink, the actual consistency might be worse.

Apache Kafka Source Connector

Apache Kafka is a distributed streaming platform. Its core is a distributed publish-subscribe messaging system that is widely adopted to ingest and distribute event streams. We briefly explain the main concepts of Kafka before we dive into the details of Flink’s Kafka connector.

Kafka organizes event streams as so-called topics. A topic is an event log that guarantees that events are read in the same order in which they were written. In order to scale writing to and reading from a topic, it can be split into partitions that are distributed across a cluster. The ordering guarantee is limited to a partition—Kafka does not provide ordering guarantees when reading from different partitions. The reading position in a Kafka partition is called an offset.

Flink provides source connectors for all common Kafka versions. Through Kafka 0.11, the API of the client library evolved and new features were added. For instance, Kafka 0.10 added support for record timestamps. Since release 1.0, the API has remained stable. Flink provides a universal Kafka connector that works for all Kafka versions since 0.11. Flink also features version-specific connectors for the Kafka versions 0.8, 0.9, 0.10, and 0.11. For the remainder of this section, we focus on the universal connector and for the version-specific connectors, we refer you to Flink’s documentation.

The dependency for the universal Flink Kafka connector is added to a Maven project as shown in the following:

<dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-connector-kafka_2.12</artifactId>
   <version>1.7.1</version>
</dependency>

The Flink Kafka connector ingests event streams in parallel. Each parallel source task can read from one or more partitions. A task tracks for each partition its current reading offset and includes it into its checkpoint data. When recovering from a failure, the offsets are restored and the source instance continues reading from the checkpointed offset. The Flink Kafka connector does not rely on Kafka’s own offset-tracking mechanism, which is based on so-called consumer groups. Figure 8-1 shows the assignment of partitions to source instances.

Figure 8-1. Read offsets of Kafka topic partitions

A Kafka source connector is created as shown in Example 8-1.

val properties = new Properties()
properties.setProperty("bootstrap.servers", "localhost:9092")
properties.setProperty("group.id", "test")

val stream: DataStream[String] = env.addSource(
  new FlinkKafkaConsumer[String](
    "topic",
    new SimpleStringSchema(),
    properties))

The constructor takes three arguments. The first argument defines the topics to read from. This can be a single topic, a list of topics, or a regular expression that matches all topics to read from. When reading from multiple topics, the Kafka connector treats all partitions of all topics the same and multiplexes their events into a single stream.

The second argument is a DeserializationSchema or KeyedDeserializationSchema. Kafka messages are stored as raw byte messages and need to be deserialized into Java or Scala objects. The SimpleStringSchema, which is used in Example 8-1, is a built-in DeserializationSchema that simply deserializes a byte array into a String. In addition, Flink provides implementations for Apache Avro and text-based JSON encodings. DeserializationSchema and KeyedDeserializationSchema are public interfaces so you can always implement custom deserialization logic.

The third parameter is a Properties object that configures the Kafka client that is internally used to connect to and read from Kafka. A minimum Properties configuration consists of two entries, "bootstrap.servers" and "group.id". Consult the Kafka documentation for additional configuration properties.

In order to extract event-time timestamps and generate watermarks, you can provide an AssignerWithPeriodicWatermark or an AssignerWithPunctuatedWatermark to a Kafka consumer by calling a FlinkKafkaConsumer.assignTimestampsAndWatermark().³ An assigner is applied to each partition to leverage the per partition ordering guarantees, and the source instance merges the partition watermarks according to the watermark propagation protocol (see “Watermark Propagation and Event Time”).

As of version 0.10.0, Kafka supports message timestamps. When reading from Kafka version 0.10 or later, the consumer will automatically extract the message timestamp as an event-time timestamp if the application runs in event-time mode. In this case, you still need to generate watermarks and should apply an AssignerWithPeriodicWatermark or an AssignerWithPunctuatedWatermark that forwards the previously assigned Kafka timestamp.

There are a few more notable configuration options. You can configure the starting position from which the partitions of a topic are initially read. Valid options are:

• The last reading position known by Kafka for the consumer group that was configured via the group.id parameter. This is the default behavior:
  FlinkKafkaConsumer.setStartFromGroupOffsets()

• The earliest offset of each individual partition:
  FlinkKafkaConsumer.setStartFromEarliest()

• The latest offset of each individual partition:
  FlinkKafkaConsumer.setStartFromLatest()

• All records with a timestamp greater than a given timestamp (requires Kafka 0.10.x or later):
  FlinkKafkaConsumer.setStartFromTimestamp(long)

• Specific reading positions for all partitions as provided by a Map object:
  FlinkKafkaConsumer.setStartFromSpecificOffsets(Map)

A Flink Kafka consumer can be configured to automatically discover new topics that match the regular expression or new partitions that were added to a topic. These features are disabled by default and can be enabled by adding the parameter flink.partition-discovery.interval-millis with a nonnegative value to the Properties object.

Apache Kafka Sink Connector

Flink provides sink connectors for all Kafka versions since 0.8. Through Kafka 0.11, the API of the client library evolved and new features were added, such as record timestamp support with Kafka 0.10 and transactional writes with Kafka 0.11. Since release 1.0, the API has remained stable. Flink provides a universal Kafka connector that works for all Kafka versions since 0.11. Flink also features version-specific connectors for Kafka versions 0.8, 0.9, 0.10, and 0.11. For the remainder of this section, we focus on the universal connector and refer you to Flink’s documentation for the version-specific connectors. The dependency for Flink’s universal Kafka connector is added to a Maven project as shown in the following:

<dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-connector-kafka_2.12</artifactId>
   <version>1.7.1</version>
</dependency>

A Kafka sink is added to a DataStream application as shown in Example 8-2.

val stream: DataStream[String] = ...

val myProducer = new FlinkKafkaProducer[String](
  "localhost:9092",         // broker list
  "topic",                  // target topic
  new SimpleStringSchema)   // serialization schema

stream.addSink(myProducer)

The constructor used in Example 8-2 receives three parameters. The first parameter is a comma-separated string of Kafka broker addresses. The second is the name of the topic to which the data is written, and the last is a SerializationSchema that converts the input types of the sink (String in Example 8-2) into a byte array. SerializationSchema is the counterpart of the DeserializationSchema that we discussed in the Kafka source section.

FlinkKafkaProducer provides more constructors with different combinations of arguments as follows:

• Similar to the Kafka source connector, you can pass a Properties object to provide custom options to the internal Kafka client. When using Properties, the list of brokers has to be provided as a "bootstrap.servers" property. Have a look at the Kafka documentation for a comprehensive list of parameters.

• You can specify a FlinkKafkaPartitioner to control how records are mapped to Kafka partitions. We will discuss this feature in more depth later in this section.

• Instead of using a SerializationSchema to convert records into byte arrays, you can also specify a KeyedSerializationSchema, which serializes a record into two byte arrays—one for the key and one for the value of a Kafka message. Moreover, KeyedSerializationSchema also exposes more Kafka-specific functionality, such as overriding the target topic to write to multiple topics.

At-least-once guarantees for the Kafka sink

The consistency guarantees that Flink’s Kafka sink provides depend on its configuration. The Kafka sink provides at-least-once guarantees under the following conditions:

• Flink’s checkpointing is enabled and all sources of the application are resettable.

• The sink connector throws an exception if a write does not succeed, causing the application to fail and recover. This is the default behavior. The internal Kafka client can be configured to retry writes before declaring them failed by setting the retries property to a value larger than zero (the default). You can also configure the sink to log write only failures by calling setLogFailuresOnly(true) on the sink object. Note that this will void any output guarantees of the application.

• The sink connector waits for Kafka to acknowledge in-flight records before completing its checkpoint. This is the default behavior. By calling setFlushOnCheck point(false) on the sink object, you can disable this waiting. However, this will also disable any output guarantees.

Exactly-once guarantees for the Kafka sink

Kafka 0.11 introduced support for transactional writes. Due to this feature, Flink’s Kafka sink is also able to provide exactly-once output guarantees given that the sink and Kafka are properly configured. Again, a Flink application must enable checkpointing and consume from resettable sources. Moreover, FlinkKafkaProducer provides a constructor with a Semantic parameter that controls the consistency guarantees provided by the sink. Possible consistency values are:

• Semantic.NONE, which provides no guarantees—records might be lost or written multiple times.

• Semantic.AT_LEAST_ONCE, which guarantees that no write is lost but might be duplicated. This is the default setting.

• Semantic.EXACTLY_ONCE, which builds on Kafka’s transactions to write each record exactly once.

There are a few things to consider when running a Flink application with a Kafka sink that operates in exactly-once mode, and it helps to roughly understand how Kafka processes transactions. In a nutshell, Kafka’s transactions work by appending all messages to the log of a partition and marking messages of open transactions as uncommitted. Once a transaction is committed, the markers are changed to committed. A consumer that reads from a topic can be configured with an isolation level (via the isolation.level property) to declare whether it can read uncommitted messages (read_uncommitted, the default) or not (read_committed). If the consumer is configured to read_committed, it stops consuming from a partition once it encounters an uncommitted message and resumes when the message is committed. Hence, open transactions can block consumers from reading a partition and introduce significant delays. Kafka guards against this by rejecting and closing transactions after a timeout interval, which is configured with the transaction.timeout.ms property.

In the context of Flink’s Kafka sink, this is important because transactions that time out—due to long recovery cycles, for example—lead to data loss. So, it is crucial to configure the transaction timeout property appropriately. By default, the Flink Kafka sink sets transaction.timeout.ms to one hour, which means you probably need to adjust the transaction.max.timeout.ms property of your Kafka setup, which is set to 15 minutes by default. Moreover, the visibility of committed messages depends on the checkpoint interval of a Flink application. Refer to the Flink documentation to learn about a few other corner cases when enabling exactly-once consistency.

Custom Partitioning and Writing Message Timestamps

When writing messages to a Kafka topic, a Flink Kafka sink task can choose to which partition of the topic to write. FlinkKafkaPartitioner can be defined in some constructors of the Flink Kafka sink. If not specified, the default partitioner maps each sink task to a single Kafka partition—all records emitted by the same sink task are written to the same partition and a single partition may contain the records of multiple sink tasks if there are more tasks than partitions. If the number of partitions is larger than the number of subtasks, the default configuration results in empty partitions, which can cause problems for Flink applications consuming the topic in event-time mode.

By providing a custom FlinkKafkaPartitioner, you can control how records are routed to topic partitions. For example, you can create a partitioner based on a key attribute of the records or a round-robin partitioner for even distribution. There is also the option to delegate the partitioning to Kafka based on the message key. This requires a KeyedSerializationSchema in order to extract the message keys and configure the FlinkKafkaPartitioner parameter with null to disable the default partitioner.

Finally, Flink’s Kafka sink can be configured to write message timestamps as supported since Kafka 0.10. Writing the event-time timestamp of a record to Kafka is enabled by calling setWriteTimestampToKafka(true) on the sink object.

Filesystem Source Connector

Filesystems are commonly used to store large amounts of data in a cost-efficient way. In big data architectures they often serve as data source and data sink for batch processing applications. In combination with advanced file formats, such as Apache Parquet or Apache ORC, filesystems can efficiently serve analytical query engines such as Apache Hive, Apache Impala, or Presto. Therefore, filesystems are commonly used to “connect” streaming and batch applications.

Apache Flink features a resettable source connector to ingest data in files as streams. The filesystem source is part of the flink-streaming-java module. Hence, you do not need to add any other dependency to use this feature. Flink supports different types of filesystems, such as local filesystems (including locally mounted NFS or SAN shares, Hadoop HDFS, Amazon S3, and OpenStack Swift FS). Refer to “Filesystem Configuration” to learn how to configure filesystems in Flink. Example 8-3 shows how to ingest a stream by reading text files line-wise.

val lineReader = new TextInputFormat(null) 

val lineStream: DataStream[String] = env.readFile[String](
  lineReader,                 // The FileInputFormat
  "hdfs:///path/to/my/data",  // The path to read
  FileProcessingMode
    .PROCESS_CONTINUOUSLY,    // The processing mode
  30000L)                     // The monitoring interval in ms

The arguments of the StreamExecutionEnvironment.readFile() method are:

• A FileInputFormat, which is responsible for reading the content of the files. We discuss the details of this interface later in this section. The null parameter of TextInputFormat in Example 8-3 defines the path that is separately set.

• The path that should be read. If the path refers to a file, the single file is read. If it refers to a directory, FileInputFormat scans the directory for files to read.

• The mode in which the path should be read. The mode can either be PROCESS_ONCE or PROCESS_CONTINUOUSLY. In PROCESS_ONCE mode, the read path is scanned once when the job is started and all matching files are read. In PROCESS_CONTINUOUSLY, the path is periodically scanned (after an initial scan) and new and modified files are continuously read.

• The interval in milliseconds in which the path is periodically scanned. The parameter is ignored in PROCESS_ONCE mode.

FileInputFormat is a specialized InputFormat to read files from a filesystem.⁴ A FileInputFormat reads files in two steps. First it scans a filesystem path and creates so-called input splits for all matching files. An input split defines a range on a file, typically via a start offset and a length. After dividing a large file into multiple splits, the splits can be distributed to multiple reader tasks to read the file in parallel. Depending on the encoding of a file, it can be necessary to only generate a single split to read the file as a whole. The second step of a FileInputFormat is to receive an input split, read the file range that is defined by the split, and return all corresponding records.

A FileInputFormat used in a DataStream application should also implement the CheckpointableInputFormat interface, which defines methods to checkpoint and reset the the current reading position of an InputFormat within a file split. The filesystem source connector provides only at-least-once guarantees when checkpointing is enabled if the FileInputFormat does not implement the CheckpointableInputFormat interface because the input format will start reading from the beginning of the split that was processed when the last complete checkpoint was taken.

In version 1.7, Flink provides a few classes that extend FileInputFormat and implement CheckpointableInputFormat. TextInputFormat reads text files line-wise (split by newline characters), subclasses of CsvInputFormat read files with comma-separated values, and AvroInputFormat reads files with Avro-encoded records.

In PROCESS_CONTINUOUSLY mode, the filesystem source connector identifies new files based on their modification timestamp. This means a file is completely reprocessed if it is modified because its modification timestamp changes. This includes modifications due to appending writes. Therefore, a common technique to continuously ingest files is to write them in a temporary directory and atomically move them to the monitored directory once they are finalized. When a file is completely ingested and a checkpoint completed, it can be removed from the directory. Monitoring ingested files by tracking the modification timestamp also has implications if you read from file stores with eventually consistent list operations, such as S3. Since files might not appear in order of their modification timestamps, they may be ignored by the filesystem source connector.

If you want to use a filesystem source connector in an event-time application, you should be aware that it can be challenging to generate watermarks since input splits are generated in a single process and round-robin distributed to all parallel readers that process them in order of the modification timestamp of the file. In order to generate satisfying watermarks you need to reason about the smallest timestamp of a record that is included in a split later processed by the task.

Filesystem Sink Connector

Writing a stream into files is a common requirement, for example, to prepare data with low latency for offline ad-hoc analysis. Since most applications can only read files once they are finalized and streaming applications run for long periods of time, streaming sink connectors typically chunk their output into multiple files. Moreover, it is common to organize records into so-called buckets, so that consuming applications have more control over which data to read.

Like the filesystem source connector, Flink’s StreamingFileSink connector is contained in the flink-streaming-java module. Hence, you do not need to add a dependency to your build file to use it.

StreamingFileSink provides end-to-end exactly-once guarantees for an application given that the application is configured with exactly-once checkpoints and all its sources are reset in the case of a failure. We will discuss the recovery mechanism in more detail later in this section. Example 8-4 shows how to create a StreamingFileSink with minimal configuration and append it to a stream.

val input: DataStream[String] = …
val sink: StreamingFileSink[String] = StreamingFileSink
  .forRowFormat(
    new Path("/base/path"), 
    new SimpleStringEncoder[String]("UTF-8"))
  .build()

input.addSink(sink)

When a StreamingFileSink receives a record, the record is assigned to a bucket. A bucket is a subdirectory of the base path that is configured with the builder of StreamingFileSink—"/base/path" in Example 8-4.

The bucket is chosen by a BucketAssigner, which is a public interface and returns for every record a BucketId that determines the directory to which the record will be written. The BucketAssigner can be configured on the builder using the withBucketAssigner() method. If no BucketAssigner is explicitly specified, it uses a DateTimeBucketAssigner that assigns records to hourly buckets based on the processing time when they are written.

Each bucket directory contains multiple part files that are concurrently written by multiple parallel instances of the StreamingFileSink. Moreover, each parallel instance chunks its output into multiple part files. The path of a part file has the following format:

[base-path]/[bucket-path]/part-[task-idx]-[id]

For example, given a base path of "/johndoe/demo" and a part prefix of "part", the path "/johndoe/demo/2018-07-22--17/part-4-8" points to the eight file that was written by the fifth (0-indexed) sink task to bucket "2018-07-22--17"—the 5 p.m. bucket of July 22, 2018.

A RollingPolicy determines when a task creates a new part file. You can configure the RollingPolicy with the withRollingPolicy() method on the builder. By default, StreamingFileSink uses a DefaultRollingPolicy, which is configured to roll part files when they exceed 128 MB or are older than 60 seconds. You can also configure an inactivity interval after which a part file is rolled.

StreamingFileSink supports two modes of writing records to part files: row encoding and bulk encoding. In row encoding mode, every record is individually encoded and appended to a part file. In bulk encoding, records are collected and written in batches. Apache Parquet, which organizes and compresses records in a columnar format, is a file format that requires bulk encoding.

Example 8-4 creates a StreamingFileSink with row encoding by providing an Encoder that writes single records to a part file. In Example 8-4, we use a SimpleStringEncoder, which calls the toString() method of the record and writes the String representation of the record to the file. Encoder is a simple interface with a single method that can be easily implemented.

A bulk-encoding StreamingFileSink is created as shown in Example 8-5.

val input: DataStream[String] = …
val sink: StreamingFileSink[String] = StreamingFileSink
  .forBulkFormat(
    new Path("/base/path"), 
    ParquetAvroWriters.forSpecificRecord(classOf[AvroPojo]))
  .build()

input.addSink(sink)

A StreamingFileSink in bulk-encoding mode requires a BulkWriter.Factory. In Example 8-5 we use a Parquet writer for Avro files. Note that the Parquet writer is contained in the flink-parquet module, which needs to be added as a dependency. As usual, BulkWriter.Factory is an interface that can be implemented for custom file formats, such as Apache Orc.

StreamingFileSink provides exactly-once output guarantees. The sink achieves this by a commit protocol that moves files through different stages, in progress, pending, and finished, and that is based on Flink’s checkpointing mechanism. While a sink writes to a file, the file is in the in-progress state. When the RollingPolicy decides to roll a file, it is closed and moved into the pending state by renaming it. Pending files are moved into the finished state (again by renaming) when the next checkpoint completes.

In the case of a failure, a sink task needs to reset its current in-progress file to its writing offset at the last successful checkpoint. This is done by closing the current in-progress file and discarding the invalid part at the file’s end, for example, by using the filesystem’s truncate operation.

Apache Cassandra Sink Connector

Apache Cassandra is a popular, scalable, and highly available column store database system. Cassandra models datasets as tables of rows that consist of multiple typed columns. One or more columns have to be defined as (composite) primary keys. Each row can be uniquely identified by its primary key. Among other APIs, Cassandra features the Cassandra Query Language (CQL), a SQL-like language to read and write records and create, modify, and delete database objects, such as keyspaces and tables.

Flink provides a sink connector to write data streams to Cassandra. Cassandra’s data model is based on primary keys, and all writes to Cassandra happen with upsert semantics. In combination with exactly-once checkpointing, resettable sources, and deterministic application logic, upsert writes yield eventually exactly-once output consistency. The output is only eventually consistent, because results are reset to a previous version during recovery, meaning consumers might read older results than read previously. Also, the versions of values for multiple keys might be out of sync.

In order to prevent temporal inconsistencies during recovery and provide exactly-once output guarantees for applications with nondeterministic application logic, Flink’s Cassandra connector can be configured to leverage a WAL. We will discuss the WAL mode in more detail later in this section. The following code shows the dependency you need to add to the build file of your application in order to use the Cassandra sink connector:

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-cassandra_2.12</artifactId>
  <version>1.7.1</version>
</dependency>

To illustrate the use of the Cassandra sink connector, we use the simple example of a Cassandra table that holds data about sensor readings and consists of two columns, sensorId and temperature. The CQL statements in Example 8-6 create a keyspace “example” and a table “sensors” in that keyspace.

CREATE KEYSPACE IF NOT EXISTS example
  WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'};

CREATE TABLE IF NOT EXISTS example.sensors (
  sensorId VARCHAR,
  temperature FLOAT,
  PRIMARY KEY(sensorId)
);

Flink provides different sink implementations to write data streams of different data types to Cassandra. Flink’s Java tuples and Row type and Scala’s built-in tuples and case classes are handled differently than user-defined POJO types. We discuss both cases separately. Example 8-7 shows how to create a sink that writes a DataStream of tuples, case classes, or rows into a Cassandra table. In this example, a DataStream[(String, Float)] is written into the “sensors” table.

val readings: DataStream[(String, Float)] = ???

val sinkBuilder: CassandraSinkBuilder[(String, Float)] =
  CassandraSink.addSink(readings)
sinkBuilder
  .setHost("localhost")
  .setQuery(
    "INSERT INTO example.sensors(sensorId, temperature) VALUES (?, ?);")
  .build()

Cassandra sinks are created and configured using a builder that is obtained by calling the CassandraSink.addSink() method with the DataStream object that should be emitted. The method returns the right builder for the data type of the DataStream. In Example 8-7, it returns a builder for a Cassandra sink that handles Scala tuples.

The Cassandra sink builders for tuples, case classes, and rows require the specification of a CQL INSERT query.⁵ The query is configured using the CassandraSinkBuilder.setQuery() method. During execution, the sink registers the query as a prepared statement and converts the fields of tuples, case classes, or rows into parameters for the prepared statement. The fields are mapped to the parameters based on their position; the first value is converted to the first parameter and so on.

Since POJO fields do not have a natural order, they need to be treated differently. Example 8-8 shows how to configure a Cassandra sink for a POJO of type SensorReading.

val readings: DataStream[SensorReading] = ???

CassandraSink.addSink(readings)
  .setHost("localhost")
  .build()

As you can see in Example 8-8, we do not specify an INSERT query. Instead, POJOs are handed to Cassandra’s Object Mapper, which automatically maps POJO fields to fields of a Cassandra table. In order for this to work, the POJO class and its fields need to be annotated with Cassandra annotations and provide setters and getters for all fields as shown in Example 8-9. The default constructor is required by Flink as mentioned in “Supported Data Types” when discussing supported data types.

@Table(keyspace = "example", name = "sensors")
class SensorReadings(
  @Column(name = "sensorId") var id: String,
  @Column(name = "temperature") var temp: Float) {

  def this() = {
      this("", 0.0)
 }

  def setId(id: String): Unit = this.id = id
  def getId: String = id
  def setTemp(temp: Float): Unit = this.temp = temp
  def getTemp: Float = temp
}

In addition to the configuration options in Figures 8-7 and 8-8, a Cassandra sink builder provides a few more methods to configure the sink connector:

• setClusterBuilder(ClusterBuilder): The ClusterBuilder builds a Cassandra Cluster that manages the connection to Cassandra. Among other options, it can configure the hostnames and ports of one or more contact points; define load balancing, retry, and reconnection policies; and provide access credentials.

• setHost(String, [Int]): This method is a shortcut for a simple ClusterBuilder configured with the hostname and port of a single contact point. If no port is configured, Cassandra’s default port 9042 is used.

• setQuery(String): This specifies the CQL INSERT query to write tuples, case classes, or rows to Cassandra. A query must not be configured to emit POJOs.

• setMapperOptions(MapperOptions): This provides options for Cassandra’s Object Mapper, such as configurations for consistency, time-to-live (TTL), and null field handling. The options are ignored if the sink emits tuples, case classes, or rows.

• enableWriteAheadLog([CheckpointCommitter]): This enables the WAL to provide exactly-once output guarantees in the case of nondeterministic application logic. CheckpointCommitter is used to store information about completed checkpoints in an external datastore. If no CheckpointCommitter is configured, the information is written into a specific Cassandra table.

The Cassandra sink connector with WAL is implemented based on Flink’s GenericWriteAheadSink operator. How this operator works, including the role of the CheckpointCommitter, and which consistency guarantees it provides, is described in more detail in “Transactional Sink Connectors”.

Resettable Source Functions

Earlier in this chapter we explained that Flink can only provide satisfying consistency guarantees for applications that use source connectors that can replay their output data. A source function can replay its output if the external system that provides the data exposes an API to retrieve and reset a reading offset. Examples of such systems are filesystems that provide the offset of a file stream and a seek method to move a file stream to a specific position or Apache Kafka, which provides offsets for each partition of a topic and can set the reading position of a partition. A counterexample is a source connector that reads data from a network socket, which immediately discards delivered the data.

A source function that supports output replay needs to be integrated with Flink’s checkpointing mechanism and must persist all current reading positions when a checkpoint is taken. When the application is started from a savepoint or recovers from a failure, the reading offsets are retrieved from the latest checkpoint or savepoint. If the application is started without existing state, the reading offsets must be set to a default value. A resettable source function needs to implement the CheckpointedFunction interface and should store the reading offsets and all related meta information, such as file paths or partition ID, in operator list state or operator union list state depending on how the offsets should be distributed to parallel task instances in the case of a rescaled application. See “Scaling Stateful Operators” for details on the distribution behavior of operator list state and union list state.

In addition, it is important to ensure that the SourceFunction.run() method, which runs in a separate thread, does not advance the reading offset and emit data while a checkpoint is taken; in other words, while the CheckpointedFunction.snapshotState() method is called. This is done by guarding the code in run() that advances the reading position and emits records in a block that synchronizes on a lock object, which is obtained from the SourceContext.getCheckpointLock() method. Example 8-11 makes the CountSource of Example 8-10 resettable.

class ResettableCountSource
    extends SourceFunction[Long] with CheckpointedFunction {

  var isRunning: Boolean = true
  var cnt: Long = _
  var offsetState: ListState[Long] = _

  override def run(ctx: SourceFunction.SourceContext[Long]) = {
    while (isRunning && cnt < Long.MaxValue) {
      // synchronize data emission and checkpoints
      ctx.getCheckpointLock.synchronized {
        cnt += 1
        ctx.collect(cnt)
      }
    }
  }

  override def cancel() = isRunning = false

  override def snapshotState(snapshotCtx: FunctionSnapshotContext): Unit = {
    // remove previous cnt
    offsetState.clear()
    // add current cnt
    offsetState.add(cnt)
  }

  override def initializeState(
      initCtx: FunctionInitializationContext): Unit = {
 
    val desc = new ListStateDescriptor[Long]("offset", classOf[Long])
    offsetState = initCtx.getOperatorStateStore.getListState(desc)
    // initialize cnt variable
    val it = offsetState.get()
    cnt = if (null == it || !it.iterator().hasNext) {
      -1L
    } else {
      it.iterator().next()
    }
  }
}

Source Functions, Timestamps, and Watermarks

Another important aspect of source functions are timestamps and watermarks. As pointed out in “Event-Time Processing” and “Assigning Timestamps and Generating Watermarks”, the DataStream API provides two options to assign timestamps and generate watermarks. Timestamps and watermarks can be assigned and generate by a dedicated TimestampAssigner (see “Assigning Timestamps and Generating Watermarks” for details) or be assigned and generated by a source function.

A source function assigns timestamps and emits watermarks through its SourceContext object. SourceContext provides the following methods:

• def collectWithTimestamp(T record, long timestamp): Unit
• def emitWatermark(Watermark watermark): Unit

collectWithTimestamp() emits a record with its associated timestamp and emitWatermark() emits the provided watermark.

Besides removing the need for an additional operator, assigning timestamps and generating watermarks in a source function can be beneficial if one parallel instance of a source function consumes records from multiple stream partitions, such as partitions of a Kafka topic. Typically, external systems, such as Kafka, only guarantee message order within a stream partition. Given the case of a source function operator that runs with a parallelism of two and that reads data from a Kafka topic with six partitions, each parallel instance of the source function will read records from three Kafka topic partitions. Consequently, each instance of the source function multiplexes the records of three stream partitions to emit them. Multiplexing records most likely introduces additional out-of-orderness with respect to the event-time timestamps such that a downstream timestamp assigner might produce more late records than expected.

To avoid such behavior, a source function can generate watermarks for each stream partition independently and always emit the smallest watermark of its partitions as its watermark. This way, it can ensure that the order guarantees on each partition are leveraged and no unnecessary late records are emitted.

Another problem source functions have to deal with are instances that become idle and do not emit anymore data. This can be very problematic, because it may prevent the whole application from advancing its watermarks and hence lead to a stalling application. Since watermarks should be data driven, a watermark generator (either integrated in a source function or in a timestamp assigner) will not emit new watermarks if it does not receive input records. If you look at how Flink propagates and updates watermarks (see “Watermark Propagation and Event Time”), you can see that a single operator that does not advance watermarks can grind all watermarks of an application to a halt if the application involves a shuffle operation (keyBy(), rebalance(), etc.).

Flink provides a mechanism to avoid such situations by marking source functions as temporarily idle. While being idle, Flink’s watermark propagation mechanism will ignore the idle stream partition. The source is automatically set as active as soon as it starts to emit records again. A source function can decide when to mark itself as idle and does so by calling the method SourceContext.markAsTemporarilyIdle().

Idempotent Sink Connectors

For many applications, the SinkFunction interface is sufficient to implement an idempotent sink connector. This is possible if the following two properties hold:

1. The result data has a deterministic (composite) key, on which idempotent updates can be performed. For an application that computes the average temperature per sensor and minute, a deterministic key could be the ID of the sensor and the timestamp for each minute. Deterministic keys are important to ensure all writes are correctly overwritten in case of a recovery.

2. The external system supports updates per key, such as a relational database system or a key-value store.

Example 8-13 illustrates how to implement and use an idempotent SinkFunction that writes to a JDBC database, in this case an embedded Apache Derby database.

val readings: DataStream[SensorReading] = ???

// write the sensor readings to a Derby table
readings.addSink(new DerbyUpsertSink)

// -----

class DerbyUpsertSink extends RichSinkFunction[SensorReading] {
  var conn: Connection = _
  var insertStmt: PreparedStatement = _
  var updateStmt: PreparedStatement = _

  override def open(parameters: Configuration): Unit = {
    // connect to embedded in-memory Derby
    conn = DriverManager.getConnection(
       "jdbc:derby:memory:flinkExample",
       new Properties())
    // prepare insert and update statements
    insertStmt = conn.prepareStatement(
      "INSERT INTO Temperatures (sensor, temp) VALUES (?, ?)")
    updateStmt = conn.prepareStatement(
      "UPDATE Temperatures SET temp = ? WHERE sensor = ?")
  }

  override def invoke(r: SensorReading, context: Context[_]): Unit = {
    // set parameters for update statement and execute it
    updateStmt.setDouble(1, r.temperature)
    updateStmt.setString(2, r.id)
    updateStmt.execute()
    // execute insert statement if update statement did not update any row
    if (updateStmt.getUpdateCount == 0) {
      // set parameters for insert statement
      insertStmt.setString(1, r.id)
      insertStmt.setDouble(2, r.temperature)
      // execute insert statement
      insertStmt.execute()
    }
  }

  override def close(): Unit = {
    insertStmt.close()
    updateStmt.close()
    conn.close()
  }
}

Since Apache Derby does not provide a built-in UPSERT statement, the example sink performs UPSERT writes by first trying to update a row and inserting a new row if no row with the given key exists. The Cassandra sink connector follows the same approach when the WAL is not enabled.

Transactional Sink Connectors

Whenever an idempotent sink connector is not suitable, either the characteristics of the application’s output, the properties of the required sink system, or due to stricter consistency requirements, transactional sink connectors can be an alternative. As described before, transactional sink connectors need to be integrated with Flink’s checkpointing mechanism because they may only commit data to the external system when a checkpoint completes successfully.

In order to ease the implementation of transactional sinks, Flink’s DataStream API provides two templates that can be extended to implement custom sink operators. Both templates implement the CheckpointListener interface to receive notifications from the JobManager about completed checkpoints (see “Receiving Notifications About Completed Checkpoints” for details about the interface):

• The GenericWriteAheadSink template collects all outgoing records per checkpoint and stores them in the operator state of the sink task. The state is checkpointed and recovered in the case of a failure. When a task receives a checkpoint completion notification, it writes the records of the completed checkpoints to the external system. The Cassandra sink connector with WAL-enabled implements this interface.

• The TwoPhaseCommitSinkFunction template leverages transactional features of the external sink system. For every checkpoint, it starts a new transaction and writes all following records to the sink system in the context of the current transaction. The sink commits a transaction when it receives the completion notification of the corresponding checkpoint.

In the following, we describe both interfaces and their consistency guarantees.

boolean sendValues(Iterable<IN> values, long chkpntId, long timestamp)

val readings: DataStream[SensorReading] = ???

// write the sensor readings to the standard out via a write-ahead log
readings.transform(
  "WriteAheadSink", new SocketWriteAheadSink)

// -----

class StdOutWriteAheadSink extends GenericWriteAheadSink[SensorReading](
    // CheckpointCommitter that commits checkpoints to the local filesystem
    new FileCheckpointCommitter(System.getProperty("java.io.tmpdir")),
    // Serializer for records
    createTypeInformation[SensorReading]
      .createSerializer(new ExecutionConfig),
    // Random JobID used by the CheckpointCommitter
    UUID.randomUUID.toString) {

  override def sendValues(
      readings: Iterable[SensorReading],
      checkpointId: Long,
      timestamp: Long): Boolean = {

    for (r <- readings.asScala) {
      // write record to standard out
      println(r)
    }
    true
  }
}

class TransactionalFileSink(val targetPath: String, val tempPath: String)
    extends TwoPhaseCommitSinkFunction[(String, Double), String, Void](
      createTypeInformation[String].createSerializer(new ExecutionConfig),
      createTypeInformation[Void].createSerializer(new ExecutionConfig)) {

  var transactionWriter: BufferedWriter = _

  /** Creates a temporary file for a transaction into which the records are
    * written.
    */
  override def beginTransaction(): String = {
    // path of transaction file is built from current time and task index
    val timeNow = LocalDateTime.now(ZoneId.of("UTC"))
      .format(DateTimeFormatter.ISO_LOCAL_DATE_TIME)
    val taskIdx = this.getRuntimeContext.getIndexOfThisSubtask
    val transactionFile = s"$timeNow-$taskIdx"
 
    // create transaction file and writer
    val tFilePath = Paths.get(s"$tempPath/$transactionFile")
    Files.createFile(tFilePath)
    this.transactionWriter = Files.newBufferedWriter(tFilePath)
    println(s"Creating Transaction File: $tFilePath")
    // name of transaction file is returned to later identify the transaction
    transactionFile
  }

  /** Write record into the current transaction file. */
  override def invoke(
      transaction: String,
      value: (String, Double),
      context: Context[_]): Unit = {
    transactionWriter.write(value.toString)
    transactionWriter.write('
')
  }

  /** Flush and close the current transaction file. */
  override def preCommit(transaction: String): Unit = {
    transactionWriter.flush()
    transactionWriter.close()
  }

  /** Commit a transaction by moving the precommitted transaction file
    * to the target directory.
    */
  override def commit(transaction: String): Unit = {
    val tFilePath = Paths.get(s"$tempPath/$transaction")
    // check if the file exists to ensure that the commit is idempotent
    if (Files.exists(tFilePath)) {
      val cFilePath = Paths.get(s"$targetPath/$transaction")
      Files.move(tFilePath, cFilePath)
    }
  }

  /** Aborts a transaction by deleting the transaction file. */
  override def abort(transaction: String): Unit = {
    val tFilePath = Paths.get(s"$tempPath/$transaction")
    if (Files.exists(tFilePath)) {
      Files.delete(tFilePath)
    }
  }
}