Log Fused Stage Name Instead of computation id in more places - #39560
Open
rwiggles wants to merge 76 commits into
Open
Log Fused Stage Name Instead of computation id in more places#39560rwiggles wants to merge 76 commits into
rwiggles wants to merge 76 commits into
Conversation
Contributor
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
…ging is turned on. Also switch to logging the dfe name instead of the computation name and fix an overflow in logging the sharding key
…ache/beam/runners/dataflow/worker/streaming/KeyCommitTooLargeException.java Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
…ache/beam/runners/dataflow/worker/windmill/work/processing/StreamingWorkScheduler.java Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
…ache/beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
Bumps [actions/checkout](https://github.com/actions/checkout) from 6 to 7. - [Release notes](https://github.com/actions/checkout/releases) - [Changelog](https://github.com/actions/checkout/blob/main/CHANGELOG.md) - [Commits](actions/checkout@v6...v7) --- updated-dependencies: - dependency-name: actions/checkout dependency-version: '7' dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
…ache#39333) Bumps [cloud.google.com/go/datastore](https://github.com/googleapis/google-cloud-go) from 1.24.0 to 1.25.0. - [Release notes](https://github.com/googleapis/google-cloud-go/releases) - [Changelog](https://github.com/googleapis/google-cloud-go/blob/main/documentai/CHANGES.md) - [Commits](googleapis/google-cloud-go@kms/v1.24.0...kms/v1.25.0) --- updated-dependencies: - dependency-name: cloud.google.com/go/datastore dependency-version: 1.25.0 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
…#39334) Bumps [github.com/aws/aws-sdk-go-v2/feature/s3/manager](https://github.com/aws/aws-sdk-go-v2) from 1.22.32 to 1.22.33. - [Release notes](https://github.com/aws/aws-sdk-go-v2/releases) - [Commits](aws/aws-sdk-go-v2@feature/s3/manager/v1.22.32...feature/s3/manager/v1.22.33) --- updated-dependencies: - dependency-name: github.com/aws/aws-sdk-go-v2/feature/s3/manager dependency-version: 1.22.33 dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
…ache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java Co-authored-by: Arun Pandian <arunpandianp@gmail.com>
…erations by removing backend details
…he#39550) Bumps [cloud.google.com/go/spanner](https://github.com/googleapis/google-cloud-go) from 1.93.0 to 1.94.0. - [Release notes](https://github.com/googleapis/google-cloud-go/releases) - [Changelog](https://github.com/googleapis/google-cloud-go/blob/main/CHANGES.md) - [Commits](googleapis/google-cloud-go@spanner/v1.93.0...spanner/v1.94.0) --- updated-dependencies: - dependency-name: cloud.google.com/go/spanner dependency-version: 1.94.0 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Bumps [github.com/aws/aws-sdk-go-v2/service/s3](https://github.com/aws/aws-sdk-go-v2) from 1.106.2 to 1.106.3. - [Release notes](https://github.com/aws/aws-sdk-go-v2/releases) - [Commits](aws/aws-sdk-go-v2@service/s3/v1.106.2...service/s3/v1.106.3) --- updated-dependencies: - dependency-name: github.com/aws/aws-sdk-go-v2/service/s3 dependency-version: 1.106.3 dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Derrick Williams <derrickaw@google.com>
Bumps com.gradle.common-custom-user-data-gradle-plugin from 2.7.0 to 2.8.0. --- updated-dependencies: - dependency-name: com.gradle.common-custom-user-data-gradle-plugin dependency-version: 2.8.0 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
* [Go SDK] Add Coder.IsDeterministic and ShardedKey standard coder This introduces the supporting infrastructure required by the upcoming GroupIntoBatches transform (apache#19868): - (*coder.Coder).IsDeterministic() reports whether a coder produces byte-stable output. Primitives (bytes, bool, varint, double, string) are deterministic; composite coders (KV, CoGBK, Nullable, Iterable, LP, ShardedKey) are deterministic iff every component is. Custom user-registered coders are non-deterministic by default and opt in via the new RegisterDeterministicCoder registration helper. - beam.Coder.IsDeterministic() forwards to the inner coder's method so transform authors can gate on determinism without reaching into internals. - beam.PCollection.WindowingStrategy() exposes the input's windowing strategy publicly so transforms honoring allowed lateness (e.g. GroupIntoBatches) can read it without package-private access. - typex.ShardedKey[K] is a concrete Go generic struct representing a sharded user key. The accompanying Kind (coder.ShardedKey) and beam:coder:sharded_key:v1 URN wiring (graphx marshal/unmarshal, exec encode/decode) produce the exact wire format documented in standard_coders.yaml:501-521 — verified byte-identical against the four published fixtures. Cross-SDK byte compatibility is required for Dataflow/Flink interoperability; a single divergent byte would silently corrupt pipelines. Roundtrip tests cover all four yaml fixtures. * [Go SDK] Add GroupIntoBatches transform and ShardedKey composite (apache#19868) Builds on top of the Coder.IsDeterministic / DeterministicCoder foundation and introduces the full user-facing surface for batching PCollection<KV<K,V>> elements by key. * typex.ShardedKey is added as a new Composite marker type (alongside KV, CoGBK, WindowedValue, Timers). Its runtime representation is a two-part FullValue (Elm=key, Elm2=[]byte shardID). * coder.NewSK builds the associated coder; graphx/coder and exec/coder wire the beam:coder:sharded_key:v1 URN in both directions. The wire format is byte-identical to the Java util.ShardedKey.Coder and the Python sharded_key coder — verified against all four standard_coders.yaml fixtures (lines 501-521). * beam.PCollection.WindowingStrategy and beam.Coder.IsDeterministic are exposed publicly, matching the access pattern already used inside the beam package (pardo.go, gbk.go). * transforms/batch introduces GroupIntoBatches, a stateful DoFn that buffers per-key values in a state.Bag and flushes when BatchSize / BatchSizeBytes / MaxBufferingDuration / end-of-window + allowed lateness triggers fire. The transform honors the input's allowed lateness (Java parity; Python currently ignores it) and panics at pipeline-build time on invalid params, non-KV inputs, or non-deterministic key coders. * CHANGES.md is updated under [2.74.0] - Unreleased. Scope note: this release ships GroupIntoBatches with string keys and string values. The underlying ShardedKey infrastructure is fully in place (type, coder, URN, tests); GroupIntoBatchesWithShardedKey and arbitrary K/V generics are follow-up work once the Go SDK binds universal types through state.Bag element coders. End-to-end Prism integration testing of the stateful DoFn path remains a follow-up — the pipeline hangs on job completion in the bounded case, pending investigation of Prism's watermark signalling for event-time timers set on the GlobalWindow maxTimestamp. All unit tests (coder roundtrip, Params validation, primitive sizer) pass. * [Go SDK] Support generic K,V and WithShardedKey in GroupIntoBatches (apache#19868) Extends GroupIntoBatches to arbitrary key/value types and adds GroupIntoBatchesWithShardedKey, completing the Apache Beam GroupIntoBatches feature parity with Java/Python (apache#19868). Generic K, V support: - Replaces the string-only DoFn with a typex.T / typex.V universal pair, resolved by beam.ParDo's type-binding engine at graph construction. Values flow through a state.Bag[[]byte] encoded via a cached beam.ElementEncoder/Decoder lazily initialised from beam.EncodedType{T: valueType} — a single reflect.Type captured at graph time and serialised across the SDK-worker boundary. - Separates into two concrete DoFn shapes: the plain groupIntoBatchesFn (event-time timer only) and groupIntoBatchesBufferedFn (event-time + processing-time). A single DoFn with an unused processing-time timer family stalls Prism waiting for the family's completion signal — splitting the shape by params.MaxBufferingDuration avoids the stall. - ProcessingTime timer is only wired when the user requests buffering, eliminating the Prism stall we hit on the initial implementation. WithShardedKey: - Adds GroupIntoBatchesWithShardedKey(s, params, col) that round-trips KV<K, V> → KV<[]byte-shardKey, V> → batched → KV<K, []V>. ShardIDs are 24-byte worker-UUID + atomic-counter tuples matching Java/Python layouts; downstream workers see independent state per shard, so a single hot logical key's processing spreads across workers on distributed runners. - Output shape: PCollection<KV<K, []V>>, identical to GroupIntoBatches. The Go SDK's type-binding engine does not accept custom generic structs as DoFn output types, so we do not surface ShardedKey<K> to the user. Cross-SDK bytes-compat ShardedKey coder infrastructure is still wired at the core/typex + core/graph/coder level for future bidirectional pipelines. Testing: - End-to-end Prism tests for GroupIntoBatches across count, byte and per-key-isolation triggers, including a non-string value type (int). - GroupIntoBatchesWithShardedKey pipeline construction test (Prism panics on the 3-stage round-trip pipeline with "assignment to nil map" in aggregateStageKind.buildEventTimeBundle — a runner-side regression we verify does NOT reproduce on non-Prism runners). Follow-up items documented in the package godoc. * [Go SDK] Fix ShardedKey coder serialization for generic closures (apache#19868) Go generic functions produce closures with identical compiler-assigned symbol names across type instantiations — all RegisterShardedKeyType[K] instantiations generated closures named "RegisterShardedKeyType[...].func1", causing cross-worker deserialization to resolve the wrong enc/dec function (last-registered wins). Root cause: reflectx.FunctionName calls runtime.FuncForPC which returns the compiler name; Go does not qualify closure names by type parameter. Fix: three surgical additions to core SDK infrastructure: 1. reflectx.MakeFuncWithName wraps a Func with a caller-supplied Name() so the serializer (encodeUserFn → u.Fn.Name()) emits a type-qualified name like "batch.encShardedKey[string]". 2. runtime.RegisterFunctionWithName registers a function under a custom name in the resolution cache so the deserializer (decodeUserFn → ResolveFunction) finds it. 3. coder.RegisterDeterministicCoderWithFuncs accepts pre-wrapped funcx.Fn values carrying the qualified names, bypassing the automatic name derivation in NewCustomCoder. RegisterShardedKeyType[K] now uses these three mechanisms to produce stable, collision-free names per type parameter. Additionally completes GroupIntoBatchesWithShardedKey as a fully generic function that wraps each key with ShardedKey{Key, ShardID} and routes through GroupIntoBatches. End-to-end Prism test passes. * Update batch.go Fix staticcheck finding * Update doc.go * Update registry.go * Update batch.go package comment --------- Co-authored-by: Jack McCluskey <34928439+jrmccluskey@users.noreply.github.com>
…#39605) Bumps [github.com/aws/aws-sdk-go-v2/feature/s3/manager](https://github.com/aws/aws-sdk-go-v2) from 1.22.37 to 1.22.38. - [Release notes](https://github.com/aws/aws-sdk-go-v2/releases) - [Commits](aws/aws-sdk-go-v2@feature/s3/manager/v1.22.37...feature/s3/manager/v1.22.38) --- updated-dependencies: - dependency-name: github.com/aws/aws-sdk-go-v2/feature/s3/manager dependency-version: 1.22.38 dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Derrick Williams <derrickaw@google.com>
Bumps [zizmorcore/zizmor-action](https://github.com/zizmorcore/zizmor-action) from 0.6.1 to 0.6.2. - [Release notes](https://github.com/zizmorcore/zizmor-action/releases) - [Commits](zizmorcore/zizmor-action@6fc4b00...3dc1ecc) --- updated-dependencies: - dependency-name: zizmorcore/zizmor-action dependency-version: 0.6.2 dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
* Split unit tests not requiring a broker out of JmsIO * Run messaging xlang Python postcommit on single Python version * Update website about io support status
…uster (apache#39580) * [Docs] Add a contributor guide for running Python on a local Flink cluster Covers getting a Flink distribution, cluster configuration and startup, running a pipeline with FlinkRunner and LOOPBACK, troubleshooting, and teardown. Linked from the contributor docs README. * [Docs] Describe the local Flink setup as three components
…y CSV reads (apache#39581) Signed-off-by: ManvithPanyam <250704031+ManvithPanyam@users.noreply.github.com>
…pache#39600) * wrap up * trigger ITs and add to CHANGES.md * fix late snapshot edge case * update resolution optimization --------- Co-authored-by: Ahmed Abualsaud <ahmedabualsaud@MacBook-Pro-2.local>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
20 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Switch logging to fused stage name in more places. This simplifies operations by removing backend details that required reaching out to Dataflow support for translation.
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes.See the Contributor Guide for more tips on how to make review process smoother.
To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md
GitHub Actions Tests Status (on master branch)
See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.