亚马逊托管的 Apache Flink 服务现已支持 Apache Flink 版本 119 大数

Amazon托管的Apache Flink服务现已支持Apache Flink版本119

作者：Francisco Morillo 和 Lorenzo Nicora 日期：2024年7月8日

关键要点

Apache Flink是一个开源的分布式处理引擎，支持强大的编程接口用于流处理和批处理，并且对有状态处理和事件时间语义提供一流支持。Apache Flink支持多种编程语言，如Java、Python、Scala和SQL，同时提供不同抽象级别的多种API，可以在同一个应用程序中交替使用。

Amazon托管的Apache Flink服务 提供一种完全托管的无服务器体验，可以运行Apache Flink应用程序，现在支持Apache Flink 1191，这是撰写时的最新稳定版本。AWS主导了此版本1191的社区发布，相较于2024年3月发布的版本1190，该版本引入了多项错误修复。

在这篇文章中，我们将讨论此新版本中引入的一些有趣的新功能和配置变化。在每个Apache Flink版本发布中，都会有一些激动人心的新实验性功能。然而，在本篇文章中，我们将重点关注此次发布中最易于用户访问的功能。

连接器

随着1191版本的发布，Apache Flink社区还为119运行时发布了新的连接器版本。从116版起，Apache Flink引入了一种新的连接器版本编号，遵循ltconnectorversiongtltflinkversiongt的模式。建议使用与您所使用的运行时版本相匹配的连接器。请参考使用Apache Flink连接器，以便随时了解关于连接器版本和兼容性的未来变化。

SQL

Apache Flink 119引入了新功能和改进，特别是在SQL API方面。这些增强旨在为开发人员提供更大的灵活性、更好的性能以及更易于使用的工具。以下是我们将深入探讨的此版本引入的一些显著的SQL增强。

操作符级的状态TTL

在Apache Flink 118中引入了操作符级的状态TTL配置，但对最终用户来说并不容易访问。修改操作符TTL时，需要在开发时导出计划，手动修改，并强制Apache Flink使用编辑后的计划，而不是在应用程序启动时生成新的计划。119中新增的功能通过允许直接使用SQL提示进行TTL配置，使这一过程变得简单，省去了JSON计划操作的需要。

以下代码示例展示了如何使用SQL提示设置状态TTL：

sql Joins的状态TTLSELECT / STATETTL(Orders = 1d Customers = 20d) / FROM Orders LEFT OUTER JOIN Customers ON Ordersocustkey = Customersccustkey

Aggregations的状态TTLSELECT / STATETTL(o = 1d) / oorderkey SUM(ototalprice) AS revenue FROM Orders AS o GROUP BY oorderkey

会话窗口的表值函数

窗口是处理无限流的核心概念，将流划分为有限的计算桶。在119之前，Apache Flink提供了以下类型的窗口表值函数(TVF)：

在Apache Flink 119发布时，它增强了其SQL能力，通过支持在流模式下使用会话窗口TVFs，允许在SQL查询中进行更复杂和灵活的窗口操作。应用程序现在可以创建基于会话间隙的动态窗口。在这里有一个示例代码：

sql 具有分区键的会话窗口SELECT FROM TABLE( SESSION(TABLE Bid PARTITION BY item DESCRIPTOR(bidtime) INTERVAL 5 MINUTES))

对具有分区键的会话窗口表应用聚合SELECT windowstart windowend item SUM(price) AS totalpriceFROM TABLE( SESSION(TABLE Bid PARTITION BY item DESCRIPTOR(bidtime) INTERVAL 5 MINUTES))GROUP BY item windowstart windowend

常规连接的迷你批处理优化

在使用表API或SQL时，常规连接像表SQL中的标准等值连接，时间不作为因素可能会对状态后端施加相当大的开销，特别是在使用RocksDB时。

通常，Apache Flink一次处理一条记录的标准连接，查找与连接另一侧匹配记录的状态，更新状态并发出结果记录。这可能会给RocksDB带来相当大的压力，因为每条记录需要进行多次读取和写入。

Apache Flink 119引入了使用迷你批处理进行等值连接的功能FLIP415。启用时，Apache Flink将以小批量而不是一条记录的方式处理常规连接，从而大大减少对RocksDB状态后端的压力。迷你批处理会增加一些延迟，用户可以控制其大小。例如，以下SQL代码嵌入于Java中展示了如何实现：

javaTableConfig tableConfig = tableEnvgetConfig()tableConfigset(tableexecminibatchenabled true)tableConfigset(tableexecminibatchallowlatency 5s)tableConfigset(tableexecminibatchsize 5000)

tableEnvexecuteSql(CREATE TEMPORARY VIEW ab AS SELECT aid as aid aacontent bid as bid bbcontent FROM a LEFT JOIN b ON aid = bid)

在此配置下，Apache Flink将在处理整个迷你批之前缓冲最多5000条记录或最多5秒，取两者中较早者。

在Apache Flink 119中，迷你批仅适用于常规连接，而不适用于窗口连接或时间连接。迷你批默认关闭，需要用户显式启用并设置批量大小和延迟。此外，迷你批设置为全局配置，适用于应用程序中的所有常规连接。撰写时，尚无法为单个连接语句设置迷你批处理。

AsyncScalarFunction

在119版本之前，SQL和表API存在一个显著的限制，与Java数据流API相比，由于缺乏异步I/O支持。任何对外部系统的请求，例如数据库或REST API，甚至使用AWS SDK的任何AWS API调用，都是同步阻塞的。Apache Flink的子任务在处理记录的同时一直等待响应，然后才能继续处理下一条记录。因此，每个请求的往返延迟被加入到处理每条记录的延迟中。Apache Flink的异步I/O API消除了这一限制，但仅适用于数据流API和Java。直到119版本，SQL、表API或Python中没有简单而高效的变通方法。

Apache Flink 119引入了新的AsyncScalarFunction，一种用户定义函数UDF，可以使用对外部系统的非阻塞调用来实现，以支持在SQL和表API中类似于异步I/O的用例。

这种新类型的UDF仅在流模式下可用。目前，仅支持有序输出。数据流异步I/O还支持无序输出，当严格的顺序不是必需时，这可以进一步降低延迟。

支持Python 311

Python 311现在得到了支持，同时Python 37的支持也已完全移除FLINK33029。Apache Flink托管服务当前使用Python 311运行时来运行PyFlink应用程序。Python 311是一个仅用于修复bug的版本。Python 311引入了多项性能改进和bug修复，并没有破坏API的变更。

更多故障排除信息：作业初始化和检查点跟踪

随着FLIP384的引入，Apache Flink 119引入了跟踪报告，它显示了检查点和作业初始化的跟踪信息。从119开始，这些跟踪信息可以通过Slf4j发送到日志。在Apache Flink托管服务中，这一功能现在是默认启用的。您可以在Amazon CloudWatch日志中找到检查点和作业初始化的详细信息，还有应用程序的其他日志。

检查点跟踪包含有关每个检查点的宝贵信息。您可以在Apache Flink仪表板上找到类似的信息，但仅限于最新检查点，并且仅在应用程序运行期间。相反，在日志中，您可以查看检查点的完整历史。以下是检查点跟踪的示例：

plaintextSimpleSpan{ scope=orgapacheflinkruntimecheckpointCheckpointStatsTracker name=Checkpoint startTsMillis=1718779769305 endTsMillis=1718779769542 attributes={ jobId=1b418a2404cbcf47ef89071f83f2dff9 checkpointId=9774 checkpointStatus=COMPLETED fullSize=9585 checkpointedSize=9585 }}

作业初始化跟踪在作业启动并从检查点或保存点恢复状态时生成。您可以找到宝贵的统计信息，通常在其他地方无法找到，包括Apache Flink仪表板。以下是作业初始化跟踪的示例：

plaintextSimpleSpan{ scope=orgapacheflinkruntimecheckpointCheckpointStatsTracker name=JobInitialization startTsMillis=1718781201463 endTsMillis=1718781409657 attributes={ maxReadOutputDataDurationMs=89 initializationStatus=COMPLETED fullSize=26167879378 sumMailboxStartDurationMs=621 sumGateRestoreDurationMs=29 sumDownloadStateDurationMs=199482 sumRestoredStateSizeBytesLOCALMEMORY=46764 checkpointId=270 sumRestoredStateSizeBytesREMOTE=26167832614 maxDownloadStateDurationMs=199482 sumReadOutputDataDurationMs=90 maxRestoredStateSizeBytesREMOTE=26167832614 maxInitializeStateDurationMs=201122 sumInitializeStateDurationMs=201241 jobId=8edb291c9f1c91c088db51b48de42308 maxGateRestoreDurationMs=22 maxMailboxStartDurationMs=391 maxRestoredStateSizeBytesLOCALMEMORY=46764 }}

检查点和作业初始化跟踪以INFO级别记录。只有在将Apache Flink托管服务应用程序的日志级别配置为INFO或DEBUG时，您才能在CloudWatch日志中找到它们。

Apache Flink托管服务行为变更

作为一个完全托管的服务，Apache Flink托管服务控制一些运行时配置参数，以确保应用程序的稳定性。有关可以修改的Apache Flink设置的详细信息，请参阅Apache Flink设置。

在119运行时中，如果您以编程方式修改了由Apache Flink托管服务直接控制的配置参数，则在应用程序启动时会收到显式的ProgramInvocationException，说明是什么参数造成了问题，并阻止应用程序启动。在118及更早版本中，由托管服务控制的参数的更改被悄然忽略。

要了解更多关于Apache Flink托管服务如何处理119或更高版本中的配置更改的信息，可以参考FlinkRuntimeException：“检测到不允许的配置更改”。

结论

在本文中，我们探讨了Apache Flink 119引入的一些新增功能和配置变化，现已得到Apache Flink托管服务的支持。该最新版本带来了众多增强功能，旨在提升开发人员在使用Apache Flink时的性能、灵活性和可用性。

随着对Apache Flink 119的支持，Apache Flink托管服务现在支持最新发布的Apache Flink版本。我们已经看到了一些可用于Flink SQL和PyFlink的新功能。

您可以在Apache Flink博客和版本说明中找到有关最近发布的更多详细信息：

如果您是Apache Flink的新用户，建议阅读我们选择正确API和语言的指南，并遵循入门指南以开始使用Apache Flink托管服务。

如果您已经在Apache Flink托管服务中运行应用程序，可以安全地就地升级为新的119运行时。

作者简介

Francisco Morillo 是AWS的流解决方案架构师，专注于实时分析架构。在流数据领域拥有超过五年的工作经验，Francisco曾在初创公司担任数据分析师，并担任咨询公司的大数据工程师，构建流数据管道。他在Amazon托管的Apache Kafka(Amazon MSK)和Amazon托管的Apache Flink服务方面拥有深厚的专业知识。Francisco与AWS客户密切合作，构建可扩展的流数据解决方案和先进的流数据湖，确保无缝的数据处理和实时洞见。

Lorenzo Nicora 是AWS的高级流解决方案架构师，帮助EMEA地区的客户。过去25年中，他一直在构建以云为中心、数据密集型的系统，在金融行业通过咨询和金融科技产品公司工作。他广泛利用开源技术，并为多个项目做出了贡献，包括Apache Flink。

评论加载中