hlteoh37 commented on code in PR #765: URL: https://github.com/apache/flink-web/pull/765#discussion_r1856348322
########## docs/content/posts/2024-11-25-whats-new-aws-connectors-5.0.0.md: ########## @@ -0,0 +1,137 @@ +--- +title: "Introducing the new Amazon Kinesis Data Stream and Amazon DynamoDB Stream sources" +date: "2024-11-25T18:00:00.000Z" +authors: +- hong: + name: "Hong Liang Teoh" +--- + +We are pleased to introduce newer (and better!) versions of the Amazon Kinesis Data Stream and Amazon DynamoDB Stream sources, based on the [FLIP-27 source interface](https://cwiki.apache.org/confluence/display/FLINK/FLIP-27%3A+Refactor+Source+Interface). + +The new [`KinesisStreamsSource`](https://github.com/apache/flink-connector-aws/blob/v5.0/flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/KinesisStreamsSource.java) supercedes the older [`FlinkKinesisConsumer`](https://github.com/apache/flink-connector-aws/blob/v5.0/flink-connector-aws/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumer.java); and the new [`DynamoDbStreamsSource`](https://github.com/apache/flink-connector-aws/blob/v5.0/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/DynamoDbStreamsSource.java) supercedes the older [`FlinkDynamoDBStreamsConsumer`](https://github.com/apache/flink-connector-aws/blob/v5.0/flink-connector-aws/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/FlinkDynamoDBStreamsConsumer.java). The new connectors are available for Flink 1.19 onwards, and are available from AWS Connector version 5.0.0 onwards. For more information, see the section on [Dependencies](#dependencies). + +In this blogpost, we will dive into the motivation for the new source connectors, the improvements introduced in the new source connectors, and provide some guidance on migrating from old to new source connectors. + +## Dependencies + +<table> + <tr> + <th>Connector</th> + <th>API</th> + <th>Dependency</th> + <th>Usage</th> + </tr> + <tr> + <td>Amazon Kinesis Data Streams source</td> + <td>DataStream<br>Table API</td> + <td> Use the <code>flink-connector-aws-kinesis-streams</code> artifact. See <a href="https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/datastream/kinesis/";> + Flink Kinesis connector documentation</a> for details. + </td> + <td> + Use the fluent + <a href="https://github.com/apache/flink-connector-aws/blob/v5.0/flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/KinesisStreamsSourceBuilder.java";> + KinesisStreamsSourceBuilder</a> to create the source. See + <a href="https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/datastream/kinesis/";> + Flink Kinesis connector documentation</a> for more details. + </td> + </tr> + <tr> + <td>Amazon Kinesis Data Streams source</td> + <td>SQL</td> + <td> Use the <code>flink-sql-connector-aws-kinesis-streams</code> artifact. See <a href="https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/table/kinesis/";> + Flink SQL Kinesis connector documentation</a> for details. + </td> + <td> + Use the table identifier <code>kinesis</code>. See the + <a href="https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/table/kinesis/";> + Flink SQL Kinesis connector documentation</a> for configuration and usage details. + </td> + </tr> + <tr> + <td>Amazon DynamoDB Streams source</td> + <td>DataStream</td> + <td> Use the <code>flink-connector-dynamodb</code> artifact. See <a href="https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/datastream/dynamodb/";> + Flink DynamoDB connector documentation</a> for details. + </td> + <td> + Use the fluent + <a href="https://github.com/apache/flink-connector-aws/blob/v5.0/flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/DynamoDbStreamsSourceBuilder.java";> + DynamoDbStreamsSourceBuilder</a> to create the source. See + <a href="https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/datastream/dynamodb/";> + Flink DynamoDB connector documentation</a> for more details. + </td> + </tr> +</table> + + +## Why did we need new source connectors? + +We implemented new source connectors because the legacy `FlinkKinesisConsumer` and `FlinkDynamoDBStreamsConsumer` use the legacy `SourceFunction` interface, which is removed in Flink 2.x. As such, from Flink 2.x onwards, only sources using the new `Source` interface will be supported. + +## New features + +Moving over to the new `Source` interface introduced the following advantages: +- **Native integration with Flink watermarks.** On the new `Source` interface, watermark generation is abstracted away to the Flink framework, and no longer a responsibility of the source. This means the new source has support for watermark alignment, and idle watermark handling out-of-the-box. +- **Standardised Flink Source metrics.** The new `Source` framework also introduces [standardised Source metrics](https://cwiki.apache.org/confluence/display/FLINK/FLIP-33%3A+Standardize+Connector+Metrics). This means that tracking record throughput and lag can be easily tracked. +- **Records are read in-order even after a resharding operation on the stream.** The new sources ensure that parent shards are read completely before reading children shards. This allows record ordering to be maintained even after a resharding operation. See [Detailed explanation of record ordering](#appendix-detailed-explanation-of-record-ordering) for more information. + +We also take this opportunity to perform the following improvements to the connector: +- **Migrate away from AWS SDK v1 to AWS SDK v2.** This SDK update aligns with best practices. +- **Migrate away from custom retry strategies to use the AWS SDK native retry strategies instead.** This allows us to benefit from AWS error classification as well as retry algorithm. +- **Reduce jar size by >99%, from ~60MB to ~100KB.** In the new source, we no longer shade the AWS SDK and no longer need to package the Kinesis Producer Library (needed for legacy sink). This should lead to smaller Flink application jars. +- **Improve defaults.** For the `FlinkKinesisConsumer`, we implemented an improved `UniformShardAssigner` which provides even shard distribution across subtasks even when there is a resharding event, by using information around the Shard hash key range. This was not made default as it would have been a breaking change. In the new `KinesisStreamsSource`, this is now the default shard assigner. + +During the implementation of the source Table API, we had to make some breaking changes around the table identifier of `kinesis`. As such, we had to do a major version bump from `4.x` to `5.x`. In Table API / SQL, for version `4.x` and below, `kinesis` refers to the `FlinkKinesisConsumer`. However, from `5.x` onwards, `kinesis` now refers to the `KinesisStreamSource`. To use the legacy `FlinkKinesisConsumer`, you can use the table identifier of `kinesis-legacy`. See [`Kinesis Table API documentation`](https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/table/kinesis/) for more details. Review Comment: done -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
