Amazon Kinesis Data Streams SQL Connector #

Scan Source: Unbounded Sink: Batch Sink: Streaming Append Mode

The Kinesis connector allows for reading data from and writing data into Amazon Kinesis Data Streams (KDS).

Dependencies #

<dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-connector-kinesis</artifactId>
      <version>5.0.0-1.20</version>
    </dependency>

The Kinesis connector is not part of the binary distribution. See how to link with it for cluster execution here.

Versioning #

There are two available Table API and SQL distributions for the Kinesis connector. This has resulted from an ongoing migration from the deprecated SourceFunction and SinkFunction interfaces to the new Source and Sink interfaces.

The Table API and SQL interfaces in Flink only allow one TableFactory for each connector identifier. Only one TableFactory with identifier kinesis can be included in your application’s dependencies.

The following table clarifies the underlying interface that is used depending on the distribution selected:

Only include one artifact, either flink-sql-connector-aws-kinesis-streams or flink-sql-connector-kinesis. Including both will result in clashing TableFactory names.

These docs are targeted for versions 5.x onwards. The main configuration section targets kinesis identifier. For legacy configuration, please see Configuration (kinesis-legacy)

Migrating from v4.x to v5.x #

There is no state compatibility between Table API and SQL API between 4.x and 5.x. This is due to the underlying implementation being changed.

Consider starting the job with v5.x kinesis table with source.init.position of AT_TIMESTAMP slightly before the time when the job with v4.x kinesis table was stopped. Note that this may result in some re-processed some records.

How to create a Kinesis data stream table #

Follow the instructions from the Amazon KDS Developer Guide to set up a Kinesis stream. The following example shows how to create a table backed by a Kinesis data stream:

CREATE TABLE KinesisTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `category_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP(3)
)
PARTITIONED BY (user_id, item_id)
WITH (
  'connector' = 'kinesis',
  'stream.arn' = 'arn:aws:kinesis:us-east-1:012345678901:stream/my-stream-name',
  'aws.region' = 'us-east-1',
  'source.init.position' = 'LATEST',
  'format' = 'csv'
);

Available Metadata #

The kinesis table Source has a known bug that means VIRTUAL columns are not supported. Please use kinesis-legacy until the fix is completed.

The following metadata can be exposed as read-only (VIRTUAL) columns in a table definition. This is only available in the kinesis-legacy connector only.

The extended CREATE TABLE example demonstrates the syntax for exposing these metadata fields:

CREATE TABLE KinesisTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `category_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP(3),
  `arrival_time` TIMESTAMP(3) METADATA FROM 'timestamp' VIRTUAL,
  `shard_id` VARCHAR(128) NOT NULL METADATA FROM 'shard-id' VIRTUAL,
  `sequence_number` VARCHAR(128) NOT NULL METADATA FROM 'sequence-number' VIRTUAL
)
PARTITIONED BY (user_id, item_id)
WITH (
  'connector' = 'kinesis-legacy',
  'stream' = 'user_behavior',
  'aws.region' = 'us-east-2',
  'scan.stream.initpos' = 'LATEST',
  'format' = 'csv'
);

Connector Options #

Features #

Refer to the Kinesis Datastream API documentation for more detailed description of features.

Sink Partitioning #

Kinesis data streams consist of one or more shards, and the sink.partitioner option allows you to control how records written into a multi-shard Kinesis-backed table will be partitioned between its shards. Valid values are:

• fixed: Kinesis PartitionKey values derived from the Flink subtask index, so each Flink partition ends up in at most one Kinesis partition (assuming that no re-sharding takes place at runtime).
• random: Kinesis PartitionKey values are assigned randomly. This is the default value for tables not defined with a PARTITION BY clause.
• Custom FixedKinesisPartitioner subclass: e.g. 'org.mycompany.MyPartitioner'.

Records written into tables defining a PARTITION BY clause will always be partitioned based on a concatenated projection of the PARTITION BY fields. In this case, the sink.partitioner field cannot be used to modify this behavior (attempting to do this results in a configuration error). You can, however, use the sink.partitioner-field-delimiter option to set the delimiter of field values in the concatenated PartitionKey string (an empty string is also a valid delimiter).

Data Type Mapping #

Kinesis stores records as Base64-encoded binary data objects, so it doesn’t have a notion of internal record structure. Instead, Kinesis records are deserialized and serialized by formats, e.g. ‘avro’, ‘csv’, or ‘json’. To determine the data type of the messages in your Kinesis-backed tables, pick a suitable Flink format with the format keyword. Please refer to the Formats pages for more details.

Connector Options (kinesis-legacy) #

Back to top