This is an automated email from the ASF dual-hosted git repository. udim pushed a commit to branch release-2.33.0 in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/release-2.33.0 by this push: new 1a7271d [BEAM-9487] Revert PR15340 for Beam 2.33 new 3b74c92 Merge pull request #15609 from zhoufek/reversion 1a7271d is described below commit 1a7271d99efb2ca9718d0e894d7db5e8b8c4d486 Author: zhoufek <zhou...@google.com> AuthorDate: Tue Sep 28 14:28:19 2021 -0400 [BEAM-9487] Revert PR15340 for Beam 2.33 --- CHANGES.md | 5 ++-- .../python/apache_beam/options/pipeline_options.py | 3 ++- sdks/python/apache_beam/transforms/core.py | 31 ++++++++++++---------- 3 files changed, 21 insertions(+), 18 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index f2076b8..089ee9a 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -73,8 +73,6 @@ ## Breaking Changes -* Python GBK by defualt will fail on unbounded PCollections that have global windowing and a default trigger. The `--allow_unsafe_triggers` flag can be used to override this. ([BEAM-9487](https://issues.apache.org/jira/browse/BEAM-9487)). -* Python GBK will fail if it detects an unsafe trigger unless the `--allow_unsafe_triggers` flag is set. ([BEAM-9487](https://issues.apache.org/jira/browse/BEAM-9487)). * Go SDK pipelines require new import paths to use this release due to migration to Go Modules. * `go.mod` files will need to change to require `github.com/apache/beam/sdks/v2`. * Code depending on beam imports need to include v2 on the module path. @@ -83,7 +81,8 @@ ## Deprecations -* X behavior is deprecated and will be removed in X versions ([BEAM-X](https://issues.apache.org/jira/browse/BEAM-X)). +* Python GBK will stop supporting unbounded PCollections that have global windowing and a default trigger in Beam 2.34. This can be overriden with `--allow_unsafe_triggers`. ([BEAM-9487](https://issues.apache.org/jira/browse/BEAM-9487)). +* Python GBK will start requiring safe triggers or the `--allow_unsafe_triggers` flag starting with Beam 2.34. ([BEAM-9487](https://issues.apache.org/jira/browse/BEAM-9487)). ## Known Issues diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py index bba56ef..622c9f0 100644 --- a/sdks/python/apache_beam/options/pipeline_options.py +++ b/sdks/python/apache_beam/options/pipeline_options.py @@ -534,7 +534,8 @@ class TypeOptions(PipelineOptions): 'compatibility. See BEAM-11719.') parser.add_argument( '--allow_unsafe_triggers', - default=False, + # TODO(BEAM-9487): Set to False for Beam 2.34 + default=True, action='store_true', help='Allow the use of unsafe triggers. Unsafe triggers have the ' 'potential to cause data loss due to finishing and/or never having ' diff --git a/sdks/python/apache_beam/transforms/core.py b/sdks/python/apache_beam/transforms/core.py index 25b05df..d5a1abd 100644 --- a/sdks/python/apache_beam/transforms/core.py +++ b/sdks/python/apache_beam/transforms/core.py @@ -2318,12 +2318,12 @@ class GroupByKey(PTransform): windowing.windowfn, GlobalWindows) and isinstance(trigger, DefaultTrigger): if pcoll.pipeline.allow_unsafe_triggers: - # TODO(BEAM-9487) Change comment for Beam 2.33 + # TODO(BEAM-9487) Change comment for Beam 2.34 _LOGGER.warning( - '%s: PCollection passed to GroupByKey is unbounded, has a global ' - 'window, and uses a default trigger. This is being allowed ' - 'because --allow_unsafe_triggers is set, but it may prevent ' - 'data from making it through the pipeline.', + 'PCollection passed to GroupByKey (label: %s) is unbounded, has a ' + 'global window, and uses a default trigger. This will no longer ' + 'be allowed starting with Beam 2.34 unless ' + '--allow_unsafe_triggers is set.', self.label) else: raise ValueError( @@ -2332,19 +2332,22 @@ class GroupByKey(PTransform): unsafe_reason = trigger.may_lose_data(windowing) if unsafe_reason != DataLossReason.NO_POTENTIAL_LOSS: - reason_msg = str(unsafe_reason).replace('DataLossReason.', '') if pcoll.pipeline.allow_unsafe_triggers: + # TODO(BEAM-9487): Switch back to this log for Beam 2.34. + # _LOGGER.warning( + # 'Skipping trigger safety check. ' + # 'This could lead to incomplete or missing groups.') _LOGGER.warning( - '%s: Unsafe trigger `%s` detected (reason: %s). This is ' - 'being allowed because --allow_unsafe_triggers is set. This could ' - 'lead to missing or incomplete groups.', + '%s: Unsafe trigger type (%s) detected. Starting with ' + 'Beam 2.34, this will raise an error by default. ' + 'Either change the pipeline to use a safe trigger or ' + 'set the --allow_unsafe_triggers flag.', self.label, - trigger, - reason_msg) + unsafe_reason) else: - msg = '{}: Unsafe trigger: `{}` may lose data. '.format( - self.label, trigger) - msg += 'Reason: {}. '.format(reason_msg) + msg = 'Unsafe trigger: `{}` may lose data. '.format(trigger) + msg += 'Reason: {}. '.format( + str(unsafe_reason).replace('DataLossReason.', '')) msg += 'This can be overriden with the --allow_unsafe_triggers flag.' raise ValueError(msg)