beam command. beam run FILE.yaml runs a Beam YAML pipeline like python -m apache_beam.yaml.main --yaml_pipeline_file=FILE.yaml, with documented exit codes, and Ctrl-C cancels the pipeline (#40891).millis_instant and decimal logical types for cross-language rows (#39684).PRIMARY KEY (...) NOT ENFORCED to CREATE EXTERNAL TABLE, and ALTER TABLE ... ADD/DROP PRIMARY KEY (#40826).WITH (...) clause to set per-statement connector options in FROM and INSERT INTO (#40848).CHANGES(table [, start [, end]]) to read a table's changes in FROM, starting with Iceberg tables (#40982).BatchElements can group elements by key across workers (key_fn, num_shards), and RunInference model handlers can use it through batch_key_fn / batch_key_num_shards. apache_beam.ml.inference.prefix_batching adds a PrefixBatchingConfig and prompt/chat prefix extractors so LLM handlers can batch requests that share a prompt prefix and hit the server's prefix cache (#40905).--lazyBatchGroupingInputs option is enabled by default (disable it with --lazyBatchGroupingInputs=false). Records sent to GroupByKey and Combine operators keep their values encoded until they are accessed, so Flink can move them from the network into its sorter without decoding and re-encoding them (#40607).--lengthPrefixBatchRecords option is enabled by default (disable it with --lengthPrefixBatchRecords=false). Each record is prefixed with its encoded length, so Flink's sorter can spill records to disk without decoding and re-encoding them (#40607).Reshuffle in batch distributes elements round-robin from deterministic start channels instead of random ones, so inputs with few elements per sender stay balanced (#40607).--sortedBatchCombine option is enabled by default (disable it with --sortedBatchCombine=false). GroupByKey and Combine in the global window with the default trigger then finish with a dedicated operator that relies on Flink sorting its input by key, instead of keyed state and timers (#40607).ReadFromKafka(redistribute=True) creating one key per record when redistribute_num_keys is not set. It now uses the KafkaIO default of 32768 keys, as Java withRedistribute() does. On Flink the per-record keys left timer state that grew until the job ran out of memory (#41103).generic filters now accept >=, <=, == and !=; previously only < and > were recognized (#41043).pyarrow.Table with more than one column, which failed with ArrowInvalid: Column 1 ... expected length N but got length 0. This affected, for example, Beam YAML ReadFromCsv or ReadFromJson followed directly by WriteToParquet (#41042).model_byte_size metric) when torch is not installed. It was always 0, because a failed ModelManager import also disabled the resource module (#40854).WindowInto 'd' (day) duration unit parsing (24h instead of 12h) and fixed ExtractWindowingInfo duplicate field name validation (#40902).GeminiModelHandler returning only one result per batch: each prompt is now sent as its own request (up to 32 at a time), so every input gets its own result. A custom request_fn must now return one response per input (#40913).SqlTransformSchemaTransformProvider configuration schema (schematransform:org.apache.beam:sql_transform:v1) now uses only portable types so it can be discovered from Python: dialect is a string, tableproviders is an array of strings, and the unused parameters field was removed. Invalid dialects and unknown table providers are rejected with a clear error (#34613).WriteToSplunk (Java/YAML) (#40470).RpcQosOptions.Builder.withRampupThrottlingDisabled(), mirroring the existing DatastoreIO option, to opt out of the 500/50/5 write ramp-up (#27170).quota_project_id parameter in ReadFromBigQuery or --bigquery_quota_project_id pipeline option, honored by the Python BigQuery IO transforms (WriteToBigQuery with FILE_LOADS or STREAMING_INSERTS; not yet STORAGE_WRITE_API, which is Java-backed), --bigQueryQuotaProjectId pipeline option (Java), and bigqueryio.WithQuotaProject read/query option (Go) (#37431).readAsJson() can now include op and the remaining Debezium source fields (e.g. MySQL file, pos, row, ts_ms; PostgreSQL lsn, txId, ts_ms) in metadata via withFullMetadata(true) / full_metadata=True / debeziumio.FullMetadata(). Off by default (#40701).Managed.write(Managed.ICEBERG_ADD_FILES) (Java/Python) (#40827).AddFiles can register files in several tables in one transform, naming each file‘s table from its path. Set path_pattern (withPathPattern in Java) to a glob with {name} placeholders, such as gs://lake/export/{table}/**, and make table a template over what it captures, such as default.{table}. The schema pre-pass then runs per table, so a table whose files conflict is refused and its files reported while every other table is still registered. Requires schema_evolution_options and error_handling; not supported with dry_run. Error rows gain a table field only when path_pattern is set, so an existing pipeline’s error output is unchanged (#40849).--experiments=enable_heap_dump) with process RSS, CPython allocator/GC stats, and glibc mallinfo2 native-heap/fragmentation stats to help distinguish native-heap from Python-object memory growth (#39244).--resource_hints are now forwarded to expansion services, so they apply to cross-language transforms as they do in the Java and Python SDKs. The max_active_bundles_per_worker hint name is now also accepted by --resource_hints (#23893).GcsUtil now uses the java-storage client (GcsUtil V2) for GCS operations by default instead of gcsio (GcsUtil V1). To go back to V1, pass --experiments=disable_gcsutil_v2. Pipelines with the use_grpc_for_gcs experiment keep using V1. The use_gcsutil_v2 experiment is no longer needed and is now ignored (#40712).wait.On, which delays each input window until the corresponding windows in its signal PCollections have closed (#39909).EvaluateDecisions transform, with an optional Jev model adapter (#40220).QdrantWriteConfig can now optionally store chunk content in the point payload via content_payload_key, so retrieved points include the chunk text (Python) (#40476).method=DIRECT_READ, using 4-part project.catalog.namespace.table identifiers. Previously project:catalog.namespace.table was silently mis-parsed and tables that report no numBytes failed to split (Python) (#39597).AwsOptions with --aws_* pipeline options (e.g. --aws_access_key_id, --aws_region_name, --aws_endpoint_url) that set defaults for all AWS service clients. The existing --s3_* options in S3Options override these defaults for the S3 client only (#40528).--aws_credentials_secret_manager so AWS credentials (--aws_access_key_id, --aws_secret_access_key, --aws_session_token and their --s3_* overrides) can be secret specs resolved through a secret manager such as GoogleCloudSecretManager, instead of plain-text values (#40528).AwsSecret (AwsSecretsManager / type:awssecret) to retrieve secrets from AWS Secrets Manager, configured from AwsOptions (--aws_*) with optional per-secret connection overrides (#40528).schema.RegisterLogicalTypeConversion and schema.RegisterLogicalTypeCoder to register portable logical types in one call, and Go types for the standard beam:logical_type:date:v1 and beam:logical_type:micros_instant:v1 logical types, so schemas from other SDKs with these types can be used in cross-language pipelines (#39684).--dataflow_container_base_repository (Python, Go) and --dataflowContainerBaseRepository (Java) to set the repository (e.g. docker.example.com/my-prefix/cloud-dataflow/v1beta3) from which the Dataflow runner pulls its default SDK container images, including the default images of cross-language Apache Beam environments. Defaults to gcr.io/cloud-dataflow/v1beta3 (#40569).GOOGLE_CLOUD_UNIVERSE_DOMAIN or the default credentials. Added a --gcs_endpoint pipeline option to override the GCS endpoint (#40569).disableCounterMetrics, disableStringSetMetrics and disableBoundedTrieMetrics experiments are now honored by the Python SDK, as they already were in Java (Python) (#38746).jakarta.jms (JMS 3.1) and ActiveMQ 6.2.5. User code implementing JmsIO.MessageMapper, valueMapper, topicNameMapper, or providing a ConnectionFactory must update imports from javax.jms.* to jakarta.jms.*. The module now requires Java 17 at runtime (#38729).xgboost extras now install xgboost>=2.0.0,<4 and datatable>=1.1.0,<2, moving both the lower and upper bounds for both. The previous minimum xgboost and datatable versions (1.6.0 and 1.0.0 respectively) are no longer allowed (#40694).beam-sdks-java-extensions-sql-datacatalog) because GCP Data Catalog has been decommissioned (#40663).--apiRootUrl no longer defaults to https://dataflow.googleapis.com/. When neither --apiRootUrl nor an absolute --dataflowEndpoint is set, the Google API client library resolves the Dataflow endpoint and honors GOOGLE_CLOUD_UNIVERSE_DOMAIN, so Dataflow jobs can be submitted in non-default universes. Code that reads DataflowPipelineDebugOptions.getApiRootUrl() should handle a null value (#40569).Regex.find_all with group=Regex.ALL now returns every group, as (match, group1, group2, ...), instead of only group 1, and with outputEmpty=False it drops a match only when the whole match is empty, matching the group=0 behavior (#40399).Regex.all_matches now returns every group of the regex, instead of failing with a TypeError when the regex has no groups and dropping nested or unmatched trailing groups (#40562).--dataflow_endpoint now defaults to '' instead of https://dataflow.googleapis.com. When unset, the Google Cloud client library resolves the Dataflow endpoint and honors GOOGLE_CLOUD_UNIVERSE_DOMAIN, so Dataflow jobs can be submitted in non-default universes. Code that reads GoogleCloudOptions.dataflow_endpoint should handle an empty value (#40569).manifest_file_size into smaller manifests and more snapshots; its outputs are now globally windowed in batch. Streaming pipelines keep the windowing they had (#40849).int16 and uint16 struct fields as 2 byte big endian INT16 values, matching the Java and Python SDKs. This is an update incompatible change for streaming pipelines that use rows with int16 or uint16 fields (#40151).IcebergAddFiles in YAML) now carries the destination table alongside each file through the registration chain, so that one transform can register files into many tables. The coders of its internal PCollections change, which makes this an update incompatible change for streaming pipelines that use it (#40817).IcebergAddFiles will be automatically replaced with the compatible managed transform. For older Beam versions, streaming update compatibility can be maintained by specifying the pipeline option update_compatibility_version (#40827).XGBoostModelHandlerDatatable is deprecated and will be removed in Beam 2.79.0, along with the datatable dependency of the xgboost extra. datatable has not had a release since 2023 and has no wheels for Python 3.13+. Use XGBoostModelHandlerPandas (or the NumPy/SciPy handlers) instead (#40705).DoFn.process or DoFn.process_batch method gives its element or batch parameter a default value (#19631).FILE_LOADS silently omitting files when dynamic destinations use different spellings of the same table (#40001).DoFn.finish_bundle() instead of silently dropping them (#40519).row[1:] in beam.Map(lambda row: row[1:]). The slice used to keep the type of the whole tuple, which could make a following CombinePerKey fail with a TypeCheckError or encode elements with the wrong coder (#20899).KeyedModelHandler not enforcing max_models_per_worker_hint when several copies of the model handler run in the same SDK process (#40468).RunInference was serving the first batch on each worker with the original models instead of the paths from a KeyModelPathMapping side input (#40665).RunInference where adaptive throttling in remote inference codepath was accidentally disabled (#40598).RunInference that caused unnecessary model reloading in some cases when large_model=True was not set (#40441).google.golang.org/genproto package removed in recent releases, which broke builds of Go modules depending on a current genproto version (#40018).VertexAIFeatureStoreEnrichmentHandler now logs a warning instead of failing when the caller lacks permission to read the feature store metadata (#40442).BigQueryIO.Write.withMaxRetryJobs is now honored for bounded (batch) pipelines using FILE_LOADS, which previously always retried failed load jobs 3 times. Pipelines that never call it keep their existing defaults (#28281).EnumerationType (proto enum) fields as strings, instead of throwing Unsupported Beam logical type Enum (#40299).Unknown type tag (#40374).AsyncWrapper finalizing bundles before generator processing, which could drop buffered writes, and nesting iterable outputs (#40000).Duration built from float seconds now rounds to the nearest microsecond instead of truncating, which could lose a microsecond (#40263).ReadFromText with escapechar no longer skips a delimiter that directly follows an escaped delimiter, which merged two records into one (#40459).agg with a list that includes 'count' or 'size' now labels those rows count/size instead of sum (#40504).ConcatSource no longer raises ZeroDivisionError when every source bundle reports a weight of 0, e.g. bundles smaller than 1 MB (#21316).Series.value_counts(bins=...) no longer always fails with a TypeError (#40503).ReadBigQueryChangeHistory no longer fails at pipeline construction with TypeCheckError: The transform 'ExecuteQueries' requires PCollections of type ... _QueryRange when pipeline type checking is enabled (the default); a regression in 2.76.0 (#40618).ReadBigQueryChangeHistory with change_function='CHANGES' no longer fails reading its temp table with a Storage Read API 500 InternalServerError: the _CHANGE_IS_FOR_UPDATE pseudo-column is now excluded from the query instead of being written to the temp table (#40618).ReadBigQueryChangeHistory now backtick-quotes the column names it puts in the generated CHANGES/APPENDS query (columns, change_type_column, change_timestamp_column), so reserved words and names with spaces work, and columns=[] selects only the pseudo-columns instead of generating invalid SQL (#40618).max_batch_weight on a RunInference ModelHandler no longer fails with a TypeError: BatchElements now supports max_batch_weight, which limits the total element_size_fn of a batch in addition to its number of elements (#40644).BatchElements now honors element_size_fn when max_batch_duration_secs is set, and no longer emits an empty batch before an element that exceeds the batch size on its own (#40644).KeyedModelHandler and MaybeKeyedModelHandler now apply the element_size_fn and batch_length_fn of the wrapped ModelHandler to examples, instead of to (key, example) pairs (#40644).SplunkWriteError had no fields, so it now has a hand-written schema with statusCode, statusMessage and payload (#40470).VarIntCoder no longer fail (#40470).ProcessBundleResponse (#40720).drop_duplicates, duplicated and nunique no longer fail on rows with NaN values or on categorical columns with unused categories, and nunique(dropna=False) now counts NaN (#40570).TableRow conversion now honors allowMissingRequiredFields when building protocol-buffer messages, so callers that pass true with required descriptors get partial messages instead of a conversion-time UninitializedMessageException (#40189).register.Function1x1 or register.DoFn1x1) panicking on functions and methods that take or return an anonymous struct. Unnamed struct types are now only registered with the schema registry, since they have no type key for the runtime type registry (#21789).schema_update_options to WriteToBigQuery file loads, allowing BigQuery load jobs to add nullable fields or relax required fields when appending data (Python) (#21141).project.catalog.namespace.table identifiers (or a TableReference with a composite catalog.namespace dataset id). Previously such references were silently mis-parsed (Java) (#39597) .Decimal(P, S) / Decimal32/64/128/256 columns (Java) (#39840).IcebergAddFiles in YAML) can evolve the table schema before registering files, with schema_evolution_options, required_columns, incompatible_schema_handling and unverifiable_file_handling (Java/YAML, batch only) (#40144).IcebergIO.writeCdcRows (Java) or by setting mode: merge-on-read on the Managed ICEBERG write (Java, Python, YAML) (#39979).IcebergIO.writeRows(...).withSideInputTableCache() (Java) or by setting use_side_input_table_cache: true on the Managed ICEBERG write (Java, Python) (#39723).Watch can bound its deduplication state by event time, retiring an output key once the greatest emitted timestamp has moved more than the allowed lateness past it. Java adds Watch.growthOf(...).withTimestampCursor(). Python adds allowed_lateness for the existing timestamp_cursor option (#18459).@RequiresTimeSortedInput and tagged outputs is now supported in batch mode (#39779).sourceStaticSplitThresholdMb pipeline option. The default of 0 keeps the existing lazy pull-based assignment (#39873).Sample.Any, the Python equivalent of Java's Sample.any, which returns up to n arbitrary elements from a PCollection (#18552).TensorRTEngineHandlerNumPy now requires TensorRT 10 or later. TensorRT 8.x is no longer supported, since TensorRT 10 removed the engine binding API the handler was written against (#36306).AsyncWrapper mixing results between keys or windows whose values or custom identifiers match, and cancelling work belonging to other windows (#39996).beam_fn_api experiment (#29998)._id types (e.g. string ids) instead of failing to parse the generated range filters (#39900).TensorRTEngineHandlerNumPy failing with CUDA_ERROR_INVALID_VALUE on models with a single-element input or output tensor (#36306).PickleCoder/_MemoizingPickleCoder.as_deterministic_coder() raising TypeError instead of returning a working deterministic coder (#28558).--experiments=enable_otel_defaults,element_metadata_supported,disable_portable_worker. Cloud Trace incurs additional cost. (#33176)GroupIntoBatches transform and the standard beam:coder:sharded_key:v1 coder to the Go SDK, along with beam.Coder.IsDeterministic, beam.PCollection.WindowingStrategy, and coder.RegisterDeterministicCoder for opt-in deterministic custom coders (Go) (#19868).envoy-data-plane (and transitive betterproto) dependency; EnvoyRateLimiter now uses a small vendored protobuf definition instead, resolving dependency conflicts for downstream projects (#37854).sys.path on the Python SDK worker at startup. This makes Python files provided via the ‘--files_to_stage’ pipeline option importable in the pipeline code and makes it easier to initialize Python SDK harness at startup via the --beam_plugins pipeline option. For more information, see the Staging Individual Files section of the dependency management docs. This behavior can be disabled by passing the ‘--experiments=no_staged_dir_in_sys_path’ pipeline option (#39431).equal_to_approx, an assert_that matcher that compares numeric pipeline outputs with a configurable tolerance (#18028).Timestamp now supports variable subsecond precision, up to nanoseconds. The portable beam:logical_type:timestamp:v1 logical type now maps to Python's Timestamp (#39344).UnboundedSource, an interface for reading an infinite stream of records with checkpointing, watermark reporting, and bundle finalization. Read one with beam.io.Read (#19137).Watch, a transform that polls a growing set of outputs for each input element, deduplicates outputs across poll rounds, and stops per a user-supplied termination condition (#21521).pystack (or gdb if installed) using the --profiler_agent=coredump pipeline option. (#39484).(Python) Removed google-perftools from the SDK container images. Users who wish to use --profiler_agent=tcmalloc should install google-perftools APT package in their custom container images separately (#39323).
[IcebergIO] Reading a timestamptz column will now return a Timestamp.MICROS Beam logical type to preserve microseconds (the old Beam Schema.FieldType#DATETIME primitive type truncates past milliseconds). This may break the following use cases when a timestamptz column is present:
Use pipeline option --updateCompatibilityVersion=2.75.0 (or any older version) to keep the old behavior (#39344).
DoFn.process returning a str, bytes, or dict (instead of an iterable wrapping one) now raises a TypeError rather than silently iterating per-character/byte/key (Python) (#18712).
(Java) Added DRAINING and DRAINED states to PipelineResult, including runner state mappings and Dataflow update handling (#39020).
(Java) IcebergIO and projects that use it must now be built with Java 17 or later as a result of Iceberg 1.11.0 upgrade (#38925).
ValueProvider options being stringified in Python Dataflow Flex Templates (#39499).DoFn type descriptor resolution when creating cached invokers (#39309).DateTime64(precision[, 'timezone']) columns with sub-second precision (Java) (#38466).--experiments=disable_streaming_engine_state_tag_encoding_v2 or --updateCompatibilityVersion=2.74.0 pipeline option. Note that the tag encoding version cannot change during a job update. Jobs using tag encoding v2 (enabled by default for new jobs on 2.75.0+) cannot be downgraded to Beam versions prior to 2.73.0, as only versions 2.73.0 and later support tag encoding v2. (#38705).--exclude_infer_dataclass_field_type (#38797). However fixing forward is recommended./ in object names and support ** for recursive matching (Go) (#38059).0x00 bytes, which broke equality predicate pushdown in some query engines (Java) (#38580).org.lz4:lz4-java (e.g., Kafka clients) may encounter a Gradle capability conflict, because Flink 2.1+ ships at.yawk.lz4:lz4-java which declares the same capability. To resolve, add a capabilitiesResolution rule to your build.gradle that selects at.yawk.lz4:lz4-java (#38947).sort_fields config (#38269).-Pwith-kafka-streams-runner to include it (#18479).--diskProvisionedIops and --diskProvisionedThroughputMibps pipeline options (Java/Go/Python) (#38349).SELECT some_field from PCOLLECTION can now operate a PCollection of Beam Row containing pickable Python user type (#20738).beam.coders.registry.register_row as preferred API to register a named tuple or dataclass with a Beam Row. At pipelne runtime, the original type associated with the registered row are preserved across the serialization boundary (#38108).type_overrides parameter to WriteToBigQuery allowing users to specify custom BigQuery to Python type mappings when using Storage Write API. This enables support for types like DATE, DATETIME, and JSON (Python) (#25946).--disable_beartype pipeline option. (#38275)max_batch_duration_secs passthrough support in Python Enrichment BigQuery and CloudSQL handlers so batching duration can be forwarded to BatchElements (#38243).OffsetRetainer interface and FileSystemOffsetRetainer implementation to persist and restore CDC offsets across pipeline restarts, and exposed withStartOffset / withOffsetRetainer on DebeziumIO.Read and the cross-language ReadBuilder (#28248).ADKAgentModelHandler for running Google Agent Development Kit (ADK) agents (Python) (#37917).--experiments=pip_no_build_isolation to disable build isolation when installing dependencies in the runtime environment (#37331).apache_beam.ml.rag package (#37628).boot.go now passes pipeline options through a file instead of the PIPELINE_OPTIONS environment variable. If a user pairs a new Python SDK container with an older SDK version (which does not support the file-based approach), the pipeline options will not be recognized and the pipeline will fail. Users must ensure their SDK and container versions are synchronized (#37370).--update_compatibility_version=2.72.0. To fix typehints replace any incorrect typehints that were previously ignored (#37590)--experiments=pip_no_build_isolation to disable build isolation when installing dependencies in the runtime environment (#37331).pip install apache-beam[gcp,interactive,yaml,redis,hadoop,tfrecord], though most users will not need all of these extras (#34554).--gbek pipeline option to automatically replace all GroupByKey transforms (Java/Python) (#36214).--element_processing_timeout_minutes option will also interrupt the SDK process if slowness happens during DoFn initialization, for example in DoFn.setup() (#36518).dill is no longer a required, default dependency for Apache Beam (#21298).pickle_library=dill pipeline option.dill==0.3.1.1 is still pre-installed on the official Beam SDK base images, it is no longer a direct dependency of the apache-beam Python package. This means it can be overridden by other dependencies in your environment.pickle_library=dill, you must manually ensure dill==0.3.1.1 is installed in both your submission and runtime environments.pip install apache-beam[gcp,dill].FROM apache/beam_python3.10_sdk:2.69.0dill==0.3.1.1 to your worker's requirements file (e.g., requirements.txt)FROM python:3.9-slimRUN pip install --no-cache-dir apache-beam[gcp,dill]Can't get attribute '_create_code' on <module 'dill._dill' from....ImportError: Pipeline option pickle_library=dill is set, but dill is not installed...pickle_library=dill to the GitHub issue (#21298)pickle_library=dill_unsafe pipeline option. This allows overriding dill==0.3.1.1 using dill as the pickle_library. Use with extreme caution. Other versions of dill has not been tested with Apache Beam (#21298).--update_compatibility_version=2.68.0 (#36345).--transform_name_mapping.pip install apache-beam, use pip install apache-beam[hadoop,gcp,interactive,redis,test,tfrecord].sdks/java/testing/junit, which provides TestPipelineExtension for JUnit 5 while maintaining backward compatibility with existing JUnit 4 TestRule-based tests (Java) (#18733, #35688).org.apache.beam:beam-sdks-java-testing-junit.ReactiveThrottler and ThrottlingSignaler classes to streamline throttling behavior in DoFns, expose throttling mechanisms for users (#35984)--element_processing_timeout_minutes option to reduce the chance of having stalled pipelines due to unexpected cases of slow processing, where slowness might not happen again if processing of the same element is retried.--update_compatibility_version=2.67.0 (35725). Report any pickling related issues to #34903JdbcIO.WriteWithResults. Previously, these settings were ignored (#35669).ParquetTable external table provider has changed its handling of the LOCATION property. To read from a directory, the path must now end with a trailing slash (e.g., LOCATION '/path/to/data/'). Previously, a trailing slash was not required. This change was made to enable support for glob patterns and single-file paths (#35582).--r for --runner), you will now need to specify the whole argument (#34934).updateCompatibilityVersion option to the SDK version used for the original pipeline, example --updateCompatabilityVersion=2.64.0AlloyDBVectorWriterConfig API to align with new PostgresVectorWriter transform. Heres a quick guide to update your code: (#35225)v17()) to current() for better backward compatibility in the future.pickle_library, where previously dill was the default in #34695. For known issues, reporting new issues, and understanding cloudpickle behavior refer to #34903.update_compatibility_version (#34767).TypedDict typehints are now compatible with Mapping and Dict type annotations.pickle_library=dill in pipeline options (#35062)VLLM_CONFIGURE_LOGGING=0 environment variable in your custom container.number_of_worker_harness_threads.Managed.ICEBERG_CDC (#33504)--add-modules JVM option is added through a new pipeline option JdkAddRootModules. This allows extending the module graph with optional modules such as SDK incubator modules. Sample usage: <pipeline invocation> --jdkAddRootModules=jdk.incubator.vector (#30281).collections.abc.Sequence type hints was added, which can lead to pipelines failing type hint checks that were previously passing erroneously. These issues will be most commonly seen trying to consume a PCollection with a Sequence type hint after a GroupByKey or a CoGroupByKey. (#33999).SpannerConfig for SpannerIO, calling withHost with a null or empty host will now result in a Null Pointer Exception (java.lang.NullPointerException: Cannot invoke "java.lang.CharSequence.length()" because "this.text" is null). See https://github.com/GoogleCloudPlatform/DataflowTemplates/issues/34489 for context.--groupFilesFileLoad pipeline option to mitigate side-input related issues in BigQueryIO batch FILE_LOAD on certain runners (including Dataflow Runner V2) (Java) (#33587).--windmillRequestBatchedGetWorkResponse=false--files_to_stage flag (Python) (#34208)date, time, timestamp, and timestamp(tz) (#32939)SolaceIO.Write) added (Java) (#31905)."123" are preserved as strings rather than silently coerced (and possibly truncated) to numeric values. To retain the old behavior, pass dtype=True (or any other value accepted by pandas.read_json).updateCompatibilityVersion option to the SDK version used for the original pipeline, example --updateCompatabilityVersion=2.58.1--runner flag.--runner flag.upload_graph feature when using Dataflow Runner V2 (#32159).Prism is under active development and does not yet support all pipelines. See #29650 for progress.
Java file-based IOs read or write lots (100k+) files could experience slowness and/or broken metrics visualization on Dataflow UI #32649.
BigQuery Enrichment (Python): The following issues are present when using the BigQuery enrichment transform (#32780):
[Managed Iceberg] DataFile metadata is assigned incorrect partition values (#33497).
[FileBasedIO] StringSet metrics can grow unlimitedly large when pipeline involves read/write large number of files, and degrading functionalities such us metrics monitoring and Dataflow job upgrade.
ReadFromKafkaViaSDF are redistributed and may contain duplicates regardless of the configuration. This affects Java pipelines with Dataflow v2 runner and xlang pipelines reading from Kafka, (#32196)upload_graph experiment, will fail at construction time (#32159).upload_graph experiment, will fail at construction time (#32159).ReadFromKafkaViaSDF are redistributed and may contain duplicates regardless of the configuration. This affects Java pipelines with Dataflow v2 runner and xlang pipelines reading from Kafka, (#32196)inMemory() variant of Java List and Map side inputs for more efficient lookups when the entire side input fits into memory.--jinja_variables flag.DoFn.SetupContextParam and DoFn.BundleContextParam which can be used as a python DoFn.process, Map, or FlatMap parameter to invoke a context manager per DoFn setup or bundle (analogous to using setup/teardown or start_bundle/finish_bundle respectively.)upload_graph experiment, will fail at construction time (#32159).--requirements_file (Python).upload_graph experiment, will fail at construction time (#32159).--enableBundling pipeline option for BigQueryIO DIRECT_READ is replaced by --enableStorageReadApiV2. Both were considered experimental and may subject to change (Java) (#26354).--max_cache_memory_usage_mb pipeline option from 100 to 0. This option was first introduced in 2.52.0 SDK. This change restores the behavior of 2.51.0 SDK, which does not use the state cache. If your pipeline uses iterable side inputs views, consider increasing the cache size by setting the option manually. (#30360).--max_cache_memory_usage_mb=0 pipeline option. (#30360).collections.abc.Collections types properly. Some type hints that were erroneously allowed by the SDK may now fail. (#29272)--max_cache_memory_usage_mb=0 pipeline option. (#30360).beam-sdks-java-extensions-avro instead. This will allow to easily update Avro version in user code without potential breaking changes in Beam “core” since the Beam Avro extension already supports the latest Avro versions and should handle this. (#25252).UseDataStreamForBatch pipeline option to the Flink runner. When it is set to true, Flink runner will run batch jobs using the DataStream API. By default the option is set to false, so the batch jobs are still executed using the DataSet API.upload_graph as one of the Experiments options for DataflowRunner is no longer required when the graph is larger than 10MB for Java SDK (PR#28621).--max_cache_memory_usage_mb to configure state and side input cache size. The cache has been enabled to a default of 100 MB. Use --max_cache_memory_usage_mb=X to provide cache size for the user state API and side inputs. (#28770).org.apache.beam.sdk.io.CountingSource.CounterMark uses custom CounterMarkCoder as a default coder since all Avro-dependent classes finally moved to extensions/avro. In case if it's still required to use AvroCoder for CounterMark, then, as a workaround, a copy of “old” CountingSource class should be placed into a project code and used directly (#25252).host to firestoreHost in FirestoreOptions to avoid potential conflict of command line arguments (Java) (#29201).SnappyCoder are update incompatible with previous versions of the same transform (Java) on some runners. This includes PubSubIO's read (#28655).watch_file_pattern arg of the RunInference arg had no effect prior to 2.52.0. To use the behavior of arg watch_file_pattern prior to 2.52.0, follow the documentation at https://beam.apache.org/documentation/ml/side-input-updates/ and use WatchFilePattern PTransform as a SideInput. (#28948)MLTransform doesn't output artifacts such as min, max and quantiles. Instead, MLTransform will add a feature to output these artifacts as human readable format - #29017. For now, to use the artifacts such as min and max that were produced by the eariler MLTransform, use read_artifact_location of MLTransform, which reads artifacts that were produced earlier in a different MLTransform (#29016)mypy on user pipelines (#27906)worker-startup log content is now captured by the worker logger. Users who relied on print() statements for logging might notice that some logs don't flush before pipeline succeeds - we strongly advise to use logging package instead of print() statements for logging. (#28317)t reflect.Type from parquetio.Write. The element type is derived from the input PCollection (Go) (#28490)fastavro dependency to 1.8.3 or earlier on some runners that don't use Beam Docker containers: #28811--staging_location at pipeline submission. Custom container images that are not based on Beam's default image must include Apache Beam installation.(#26996)func TestMain(m *testing.M) { ptest.MainWithDefault(m, "direct") }direct_num_workers!=1.(#27373)orjson dependency might experience segmentation faults or get stuck: #28318.BigtableIO.ReadChangeStream (#27183)--prebuild_sdk_container_engine=cloud_build, like images depending on tensorflow or torch (#27023).pip cache when installing packages on the workers. This reduces the size of prebuilt Python container images (#27035).NullPointerException when a progress check is made before the reader is started (IO) (#23868)--impersonate_service_account option with BigQuery IOs might fail on Dataflow (#32030). This is fixed in 2.59.0 release.df | (Transform1 | Transform2 | ...) to avoid excessive conversions.)key and reverse. (#25888).:sdks:java:extensions:avro (#24748)LZMA compression/decompression of text files added to the Python SDK (#25316)use_bounded_concurrent_output_for_sdf.WatchFilePattern transform, which can be used as a side input to the RunInference PTransfrom to watch for model updates using a file pattern. (#24042)PytorchModelHandler. The TorchScript model path can be passed to PytorchModelHandler using torch_script_model_path=<path_to_model>. (#25321)target_batch_duration_secs_including_fixed_cost=1 to BatchElements.Dockerfile that the Beam container entrypoint and/or Beam base image version match the Beam SDK version used at job submission.beam-sdks-java-core and will be eventually removed. Please, migrate to a new module beam-sdks-java-extensions-avro instead by importing the classes from org.apache.beam.sdk.extensions.avro package. For the sake of migration simplicity, the relative package path and the whole class hierarchy of Avro related classes in new module is preserved the same as it was before. For example, import org.apache.beam.sdk.extensions.avro.coders.AvroCoder class instead oforg.apache.beam.sdk.coders.AvroCoder. (#24749).disable_runner_v2, disable_runner_v2_until_2023, disable_prime_runner_v2 experiments will raise an error during pipeline construction. You can no longer specify the Dataflow worker jar override. Note that non-portable Java jobs and non-portable Python batch jobs are not impacted. (#24515).pyarrow>=3 and pandas>=1.4.3 since older versions are not compatible with numpy==1.24.0.python -m apache_beam.runners.render --help for more details.--extra_package option (Python) (#23684).--resource_hints flag (Go) (#23990).map_windows urn added to Go SDK (#24307).ParquetIO.withSplit was removed since splittable reading has been the default behavior since 2.35.0. The effect of this change is to drop support for non-splittable reading (Java)(#23832).beam-sdks-java-extensions-google-cloud-platform-core is no longer a dependency of the Java SDK Harness. Some users of a portable runner (such as Dataflow Runner v2) may have an undeclared dependency on this package (for example using GCS with TextIO) and will now need to declare the dependency.beam-sdks-java-core is no longer a dependency of the Java SDK Harness. Users of a portable runner (such as Dataflow Runner v2) will need to provide this package and its dependencies.poetry config virtualenvs.create false before installing deps, see an example in: #25085. If you were negatively impacted by this change and cannot find a workaround, feel free to chime in on #16658. To disable this behavior, you could upgrade to Beam 2.48.0 and set an environment variable ENV RUN_PYTHON_SDK_IN_DEFAULT_ENVIRONMENT=1 in your Dockerfile.--spark_version. Spark 2 support is deprecated and will be removed soon (#23728).AfterProcessingTime behavior in Python's DirectRunner to match Java (#23071)--experiments=disable_projection_pushdown.--experiments=disable_projection_pushdown.BoundedSourceAsSDFWrapperFn and UnboundedSourceAsSDFWrapper.javax.jms.Message (Java) (BEAM-16308).topicNameMapper must be set to extract the topic name from the input value.valueMapper must be set to convert the input value to JMS message.DataFrame.unstack(), DataFrame.pivot() and Series.unstack() implemented for DataFrame API (BEAM-13948, BEAM-13966).ShallowCloneParDoPayload(), ShallowCloneSideInput(), and ShallowCloneFunctionSpec() have been removed from the Go SDK's pipelinex package (BEAM-13739).valueMapper to be set (BEAM-16308). You can use the TextMessageMapper to convert String inputs to JMS TestMessages:JmsIO.<String>write() .withConnectionFactory(jmsConnectionFactory) .withValueMapper(new TextMessageMapper());
metadata() added to io.filesystem.FileSystem in the Python SDK. (BEAM-14314)--experiments=disable_projection_pushdown.amazon-web-services2 has reached feature parity and is finally recommended over the earlier amazon-web-services and kinesis modules (Java). These will be deprecated in one of the next releases (BEAM-13174).Kinesis was added (BEAM-13175).AwsOptions (BEAM-13563, BEAM-13663, BEAM-13587).S3 Filesystem (BEAM-13245, BEAM-13246, BEAM-13441, BEAM-13445, BEAM-14011), DynamoDB IO (BEAM-13209, BEAM-13209), SQS IO (BEAM-13631, BEAM-13510) and others.--requirements_file will now be staged to the runner using binary distributions (wheels) of the PyPI packages for linux_x86_64 platform (BEAM-4032). To restore the behavior to use source distributions, set pipeline option --requirements_cache_only_sources. To skip staging the packages at submission time, set pipeline option --requirements_cache=skip (Python).DoFn.infer_output_types was expected to return Iterable[element_type] where element_type is the PCollection elemnt type. It is now expected to return element_type. Take care if you have overriden infer_output_type in a DoFn (this is not common). See BEAM-13860.amazon-web-services2) The types of awsRegion / endpoint in AwsOptions changed from String to Region / URI (BEAM-13563).amazon-web-services2) Client providers (withXYZClientProvider()) as well as IO specific RetryConfigurations are deprecated, instead use withClientConfiguration() or AwsOptions to configure AWS IOs / clients. Custom implementations of client providers shall be replaced with a respective ClientBuilderFactory and configured through AwsOptions (BEAM-13563).pyarrow version parsing (Python)(BEAM-14235)SqsMessage for AWS IOs for SDK v2 was changed from String to long, visibility of all fields was fixed from package private to public BEAM-13638.JdbcIO.readWithPartitions from int to long (BEAM-13149). This is a relatively minor breaking change, which we're implementing to improve the usability of the transform without increasing cruft. This transform is relatively new, so we may implement other breaking changes in the future to improve its usability.with_exception_handling option for easily ignoring bad records and implementing the dead letter pattern.ReadFromBigQuery and ReadAllFromBigQuery now run queries with BATCH priority by default. The query_priority parameter is introduced to the same transforms to allow configuring the query priority (Python) (BEAM-12913).ReadFromBigQuery. The newly introduced method parameter can be set as DIRECT_READ to use the Storage Read API. The default is EXPORT which invokes a BigQuery export request. (Python) (BEAM-10917).use_native_datetime parameter to ReadFromBigQuery to configure the return type of DATETIME fields when using ReadFromBigQuery. This parameter can only be used when method = DIRECT_READ(Python) (BEAM-10917).dataframe extra to the Python SDK that tracks pandas versions we've verified compatibility with. We now recommend installing Beam with pip install apache-beam[dataframe] when you intend to use the DataFrame API (BEAM-12906).go.mod files will need to change to require github.com/apache/beam/sdks/v2..../sdks/go/... to .../sdks/v2/go/...useReflectApi setting to control it (BEAM-12628).--allow_unsafe_triggers. (BEAM-9487).--allow_unsafe_triggers flag starting with Beam 2.34. (BEAM-9487).spark.jackson.version) to at least version 2.9.2, due to Beam updating its dependencies.VARCHAR, NVARCHAR, LONGVARCHAR, LONGNVARCHAR, DATE, TIME (Java)(BEAM-12385).--allow_unsafe_triggers. (BEAM-9487).--allow_unsafe_triggers flag starting with Beam 2.33. (BEAM-9487).CREATE FUNCTION DDL statement added to Calcite SQL syntax. JAR and AGGREGATE are now reserved keywords. (BEAM-12339).TriggerFn has a new may_lose_data method to signal potential data loss. Default behavior assumes safe (necessary for backwards compatibility). See Deprecations for potential impact of overriding this. (BEAM-9487).Row(x=3, y=4) is no longer considered equal to Row(y=4, x=3) (BEAM-11929).TopCombineFn disallow compare as its argument (Python) (BEAM-7372).--allow_unsafe_triggers. (BEAM-9487).--allow_unsafe_triggers flag starting with Beam 2.33. (BEAM-9487).FakeDeterministicFastPrimitivesCoder with beam.coders.registry.register_fallback_coder(beam.coders.coders.FakeDeterministicFastPrimitivesCoder()) or use the allow_non_deterministic_key_coders pipeline option.dependencyManagement in Maven and force in Gradle.ReadAllFromBigQuery that can receive multiple requests to read data from BigQuery at pipeline runtime. See PR 13170, and BEAM-9650.--region flag in amazon-web-services2 was replaced by --awsRegion (BEAM-11331).--experiments=use_deprecated_read. The Apache Beam community is looking for feedback for this change as the community is planning to make this change permanent with no opt-out. If you run into an issue requiring the opt-out, please send an e-mail to user@beam.apache.org specifically referencing BEAM-10670 in the subject line and why you needed to opt-out. (Java) (BEAM-10670)--HTTPWriteTimeout=0 to revert to the old behavior. (BEAM-6103)--experiments=use_runner_v2 before using this feature. (BEAM-3736)CombineFn.from_callable() or CombineFn.maybe_from_callable() can lead to incorrect behavior. (BEAM-11522).--experiments=use_deprecated_read. The Apache Beam community is looking for feedback for this change as the community is planning to make this change permanent with no opt-out. If you run into an issue requiring the opt-out, please send an e-mail to user@beam.apache.org specifically referencing BEAM-10670 in the subject line and why you needed to opt-out. (Java) (BEAM-10670)apache_beam.io.kinesis (BEAM-10138, BEAM-10137).apache_beam.io.snowflake (BEAM-9898).ElasticsearchIO#Write. Now, Java’s ElasticsearchIO can be used to selectively delete documents using withIsDeleteFn function (BEAM-5757).ReadFromBigQuery added. (Python) (BEAM-10524)@ptransform_fn decorators in the Python SDK. (BEAM-4091) This has not enabled by default to preserve backwards compatibility; use the --type_check_additional=ptransform_fn flag to enable. It may be enabled by default in future versions of Beam.apache_beam.io.external.snowflake to apache_beam.io.snowflake. The previous path will be removed in the future versions.--experiments=use_legacy_bq_sink.apache_beam.io.jdbc (BEAM-10135, BEAM-10136).apache_beam.io.external.snowflake (BEAM-9897).typing.FrozenSet type hints, which are not interchangeable with typing.Set. You may need to update your pipelines if type checking fails. (BEAM-10197)apache_beam.io.gcp.bigquery.ReadFromBigQuery. This transform is experimental. It reads data from BigQuery by exporting data to Avro files, and reading those files. It also supports reading data by exporting to JSON files. This has small differences in behavior for Time and Date-related fields. See Pydoc for more information.RowJson.RowJsonDeserializer, JsonToRow, and PubsubJsonTableProvider now accept “implicit nulls” by default when deserializing JSON (Java) (BEAM-10220). Previously nulls could only be represented with explicit null values, as in {"foo": "bar", "baz": null}, whereas an implicit null like {"foo": "bar"} would raise an exception. Now both JSON strings will yield the same result by default. This behavior can be overridden with RowJson.RowJsonDeserializer#withNullBehavior.GroupIntoBatches experimental transform in Python to actually group batches by key. This changes the output type for this transform (BEAM-6696).--workerCacheMB flag is supported in Dataflow streaming pipeline (BEAM-9964)--direct_num_workers=0 is supported for FnApi runner. It will set the number of threads/subprocesses to number of cores of the machine executing the pipeline (BEAM-9443).--job_endpoint to be set when using --runner=PortableRunner (BEAM-9860). Users seeking the old default behavior should set --runner=FlinkRunner instead.apache_beam.io.gcp.datastore.v1 has been removed as the client it uses is out of date and does not support Python 3 (BEAM-9529). Please migrate your code to use apache_beam.io.gcp.datastore.v1new. See the updated datastore_wordcount for example usage.Python SDK will now use Python 3 type annotations as pipeline type hints. (#10717)
If you suspect that this feature is causing your pipeline to fail, calling apache_beam.typehints.disable_type_annotations() before pipeline creation will disable is completely, and decorating specific functions (such as process()) with @apache_beam.typehints.no_annotations will disable it for that function.
More details will be in (Ensuring Python Type Safety)(https://beam.apache.org/documentation/sdks/python-type-safety/) and an upcoming blog post.
Java SDK: Introducing the concept of options in Beam Schemas. These options add extra context to fields and schemas. This replaces the current Beam metadata that is present in a FieldType only, options are available in fields and row schemas. Schema options are fully typed and can contain complex rows. Remark: Schema aware is still experimental. (BEAM-9035)
Java SDK: The protobuf extension is fully schema aware and also includes protobuf option conversion to beam schema options. Remark: Schema aware is still experimental. (BEAM-9044)
Added ability to write to BigQuery via Avro file loads (Python) (BEAM-8841)
By default, file loads will be done using JSON, but it is possible to specify the temp_file_format parameter to perform file exports with AVRO. AVRO-based file loads work by exporting Python types into Avro types, so to switch to Avro-based loads, you will need to change your data types from Json-compatible types (string-type dates and timestamp, long numeric values as strings) into Python native types that are written to Avro (Python's date, datetime types, decimal, etc). For more information see https://cloud.google.com/bigquery/docs/loading-data-cloud-storage-avro#avro_conversions.
Added integration of Java SDK with Google Cloud AI VideoIntelligence service (BEAM-9147)
Added integration of Java SDK with Google Cloud AI natural language processing API (BEAM-9634)
docker-pull-licenses tag was introduced. Licenses/notices of third party dependencies will be added to the docker images when docker-pull-licenses was set. The files are added to /opt/apache/beam/third_party_licenses/. By default, no licenses/notices are added to the docker images. (BEAM-9136)
--region option to be set, unless a default value is set in the environment (BEAM-9199). See here for more details.2.23.0. (BEAM-9704)--zone option in the Dataflow runner is now deprecated. Please use --worker_zone instead. (BEAM-9716)SpannerConfig.connectToSpanner has been moved to SpannerAccessor.create. (BEAM-9310).force_generated_pcollection_output_ids experiment.