Apache Beam Enhances Runner Parity with Drain Support and MQTT Portability

Apache Beam activity this week focused on runner parity and expanded IO connector capabilities for ClickHouse and MQTT. The project introduced Pipeline Drain support for Dataflow and Flink along with sub second precision for ClickHouse timestamps.

The project has introduced a significant enhancement for streaming operators with the addition of drain support for the Dataflow and Flink runners. This change, implemented in commit d032f0e3, provides a standard way to gracefully shut down streaming pipelines without data loss.

In the context of streaming data engineering, a drain operation allows a pipeline to stop ingestion while continuing to process all data already present in the system. Once the existing records reach their final sinks and all watermarks progress to infinity, the pipeline terminates. This is a critical requirement for production environments where scheduled maintenance or configuration updates must happen without dropping messages.

The update includes a new PipelineResult drain API. This allows developers to trigger the drain process programmatically across different runners. By unifying this behavior between Dataflow and Flink, the project continues to reduce the friction of moving workloads between managed cloud environments and self hosted clusters.

The Java SDK received notable updates to its IO connectors, specifically for ClickHouse and MQTT. These changes improve data fidelity and portability across different language SDKs.

The ClickHouseIO connector now supports DateTime64 with precision below one second. Previously, the connector only recognized second precision for timestamps, which caused issues for pipelines emitting high frequency data. As detailed in commit 378e10de, the new implementation allows for precision up to nine decimal places.

The mapping logic translates Beam field types to Joda DATETIME for precision up to three digits, SqlTypes.TIMESTAMP for four to six digits, and NanosInstant for seven and above. The writer encodes values as a little endian int64, ensuring compatibility with the underlying ClickHouse storage format. This level of detail is essential for engineers working with financial or sensor data where millisecond or microsecond accuracy is mandatory.

The MQTT connector also saw a major improvement with the introduction of portable MqttIO transforms. This work, found in commit 001093e5, adds MqttReadSchemaTransformProvider and MqttWriteSchemaTransformProvider. These providers allow MQTT to be used through the portable SchemaTransform API, making the connector available to other SDKs like Python through cross language transforms. The project also introduced a dedicated messaging-expansion-service to host these transforms, keeping the core expansion service lean.

Reliability in data ingestion is a constant focus for the project. Several fixes and features landed this week to address specific edge cases in Google Cloud Platform integrations.

A fix for WriteToPubSub now ensures that the ordering_key is correctly passed to the publish method. This change, visible in the sdks/python/apache_beam/yaml/yaml_io.py logic, is vital for applications that rely on strict message ordering. Without this fix, messages intended for the same key could arrive out of order if the client failed to propagate the key during the internal publish call.

The YAML DSL also received an update for PubSub ingestion. The ReadFromPubSub transform now supports a publish_time_field parameter. This allows users to map the metadata timestamp from PubSub directly into a field within the Beam record schema during the ingestion phase. This simplifies downstream processing where the event time might be used for windowing or auditing.

Furthermore, Google Cloud Storage integration was improved with a fix for glob matching. The project addressed an issue where double asterisks and forward slashes in object names were not correctly handled during file discovery. This ensures that complex directory structures in GCS are properly traversed when using file based sources.

Beam SQL continues to evolve as a primary interface for data analysis. This week saw several usability improvements to the SQL shell and DDL documentation.

The SQL shell now automatically includes Iceberg dependencies when using the icebergio connector. This reduces the boilerplate required to start querying Iceberg tables from the CLI. Additionally, the project updated the documentation for SQL DDL to reflect recent improvements in database connectivity and schema discovery.

These documentation updates are more than just text changes. They reflect a maturing SQL layer that aims to provide a familiar environment for engineers used to traditional data warehouses. By improving the DDL experience, the project makes it easier to define external tables and manage schema mappings without writing extensive Java or Python code.

Looking ahead, there are several internal changes that signal future directions for the project. A new finishKey() method was added to DoFnRunner in commit a88686c1. While currently an empty operation, this method is a precursor to a major change in how streaming work items are handled. The goal is to allow multiple streaming work items within a single Beam bundle, which should improve throughput for high volume streaming jobs.

Python developers should also note the addition of memory profiling instrumentation in the SDK. This will provide better visibility into memory usage patterns, which is often a bottleneck in large scale data processing. Finally, the project has officially added support for Apache Flink 2.1.3, ensuring that users can deploy the latest features and security fixes from the Flink ecosystem.