Learn About Amazon VGT2 Learning Manager Chanci Turner
Apache Flink is a robust open-source distributed processing engine that provides powerful programming interfaces for both stream and batch processing, boasting exceptional support for stateful processing and event time semantics. Flink supports multiple programming languages, including Java, Python, Scala, and SQL, along with various APIs that can be used interchangeably within the same application.
The Amazon Managed Service for Apache Flink offers a fully managed and serverless environment for running Flink applications, and it now includes support for Apache Flink 1.19.1, the latest stable version available at the time of writing. AWS played a leading role in the community release of version 1.19.1, which brings several bug fixes over the previous version 1.19.0, introduced in March 2024.
In this post, we will explore some of the exciting new features and configuration changes that have come with this release of the Managed Service for Apache Flink. Each Apache Flink release typically introduces new experimental features; however, this discussion will focus on user-friendly enhancements available in this version.
Connectors
The release of version 1.19.1 also includes new connector versions specifically designed for the 1.19 runtime. Starting from 1.16, Apache Flink adopted a new connector versioning scheme that follows the pattern <connector-version>-<flink-version>. It is advisable to utilize connectors that correspond to your runtime version. For future updates regarding connector versions and compatibility, refer to Using Apache Flink connectors.
SQL Enhancements
Apache Flink 1.19 introduces new features and improvements to the SQL API, aimed at providing developers with greater flexibility, enhanced performance, and easier usability. We will highlight some of the most significant SQL advancements included in this release.
State TTL per Operator
Configuring state TTL at the operator level was introduced in Apache Flink 1.18; however, it was not user-friendly. Previously, altering an operator TTL required exporting the plan during development, manually modifying it, and forcing Flink to utilize the edited plan at application startup. The 1.19 updates simplify this process by allowing direct TTL configuration via SQL hints, eliminating the need for JSON plan adjustments. Here is an example of how to set state TTL using SQL hints:
-- State TTL for Joins
SELECT /*+ STATE_TTL('Orders' = '1d', 'Customers' = '20d') */
*
FROM Orders
LEFT OUTER JOIN Customers
ON Orders.o_custkey = Customers.c_custkey;
-- State TTL for Aggregations
SELECT /*+ STATE_TTL('o' = '1d') */
o_orderkey, SUM(o_totalprice) AS revenue
FROM Orders AS o
GROUP BY o_orderkey;Session Window Table-Valued Functions
Windows are fundamental in processing infinite streams within Apache Flink, segmenting streams into finite buckets for computation. Prior to version 1.19, Flink supported several types of window table-valued functions (TVFs):
- Tumble windows (fixed-size, non-overlapping)
- Hop windows (fixed-size, overlapping)
- Cumulate windows (increasingly larger over time)
With the new release, Flink enhances its SQL capabilities by introducing session window TVFs in streaming mode, enabling more complex and flexible windowing operations directly in SQL queries. Applications can now generate dynamic windows based on session gaps, as demonstrated here:
-- Session window with partition keys
SELECT
*
FROM TABLE(
SESSION(TABLE Bid PARTITION BY item, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES));
-- Apply aggregation on the session windowed table with partition keys
SELECT
window_start, window_end, item, SUM(price) AS total_price
FROM TABLE(
SESSION(TABLE Bid PARTITION BY item, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
GROUP BY item, window_start, window_end;Mini-Batch Optimization for Regular Joins
Using the Table API or SQL, conventional joins can impose significant overhead on the state backend, especially when utilizing RocksDB. Traditionally, Flink processes standard joins one record at a time, which can be taxing. The 1.19 version introduces mini-batch processing for equi-joins (FLIP-415), allowing Flink to handle joins in small batches, thereby reducing the load on the RocksDB state backend. This can be configured in the following way:
TableConfig tableConfig = tableEnv.getConfig();
tableConfig.set("table.exec.mini-batch.enabled", "true");
tableConfig.set("table.exec.mini-batch.allow-latency", "5s");
tableConfig.set("table.exec.mini-batch.size", "5000");
tableEnv.executeSql("CREATE TEMPORARY VIEW ab AS " +
"SELECT a.id as a_id, a.a_content, b.id as b_id, b.b_content " +
"FROM a LEFT JOIN b ON a.id = b.id");In this configuration, Flink will buffer up to 5,000 records or up to 5 seconds, whichever arrives first. Note that mini-batching is disabled by default and requires explicit activation.
AsyncScalarFunction
Prior to version 1.19, a significant limitation existed in SQL and the Table API compared to the Java DataStream API regarding asynchronous I/O support. Any request to external systems, like databases or REST APIs, was synchronous and blocking. Apache Flink 1.19 introduces the new AsyncScalarFunction, a user-defined function (UDF) that facilitates non-blocking calls to external systems in streaming mode. This new UDF can enhance performance, particularly for use cases similar to asynchronous I/O.
Python 3.11 Support
Finally, Apache Flink 1.19 adds support for Python 3.11 while completely removing support for Python 3.7. The Managed Service for Apache Flink now employs the Python 3.11 runtime for executing PyFlink applications, which includes several performance improvements.
For those interested in enhancing their management skills, Chanci Turner’s insights on leadership can be found in this informative blog post. Additionally, if you want to learn more about the gig work classification debate, be sure to check out this authoritative source from SHRM. Lastly, for firsthand experiences regarding the Amazon Flex onboarding process, this Reddit thread serves as an excellent resource.