You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "ahmedabu98 (via GitHub)" <gi...@apache.org> on 2025/03/07 15:31:23 UTC
[I] [Task]: Add Python AfterSynchronizedProcessingTime trigger and add an Iceberg CDC streaming read test [beam]
ahmedabu98 opened a new issue, #34212:
URL: https://github.com/apache/beam/issues/34212
### What needs to happen?
Was trying to add a Python test for the new Iceberg streaming source (#33504) but ran into the following error:
```
@staticmethod
def from_runner_api(proto, context):
> return {
'after_all': AfterAll,
'after_any': AfterAny,
'after_each': AfterEach,
'after_end_of_window': AfterWatermark,
'after_processing_time': AfterProcessingTime,
# 'after_synchronized_processing_time': _AfterSynchronizedProcessingTime,
'always': Always,
'default': DefaultTrigger,
'element_count': AfterCount,
'never': _Never,
'or_finally': OrFinally,
'repeat': Repeatedly,
}[proto.WhichOneof('trigger')].from_runner_api(proto, context)
E KeyError: 'after_synchronized_processing_time'
apache_beam/transforms/trigger.py:301: KeyError
```
This trigger was going to be added in an old PR that went stale (#14060).
### Issue Priority
Priority: 2 (default / most normal work should be filed as P2)
### Issue Components
- [x] Component: Python SDK
- [ ] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [x] Component: IO connector
- [ ] Component: Beam YAML
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Infrastructure
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Samza Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Task]: Add Python AfterSynchronizedProcessingTime trigger and add an Iceberg CDC streaming read test [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles closed issue #34212: [Task]: Add Python AfterSynchronizedProcessingTime trigger and add an Iceberg CDC streaming read test
URL: https://github.com/apache/beam/issues/34212
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Task]: Add Python AfterSynchronizedProcessingTime trigger and add an Iceberg CDC streaming read test [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on issue #34212:
URL: https://github.com/apache/beam/issues/34212#issuecomment-2706751062
cc @kennknowles
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Task]: Add Python AfterSynchronizedProcessingTime trigger and add an Iceberg CDC streaming read test [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on issue #34212:
URL: https://github.com/apache/beam/issues/34212#issuecomment-3185630189
We need to fix this. Please keep this open.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Task]: Add Python AfterSynchronizedProcessingTime trigger and add an Iceberg CDC streaming read test [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on issue #34212:
URL: https://github.com/apache/beam/issues/34212#issuecomment-3155074235
This issue has been marked as stale due to 150 days of inactivity. It will be closed in 30 days if no further activity occurs. If you think that’s incorrect or this issue still needs to be addressed, please simply write any comment. If closed, you can reopen the issue at any time. Thank you for your contributions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Task]: Add Python AfterSynchronizedProcessingTime trigger and add an Iceberg CDC streaming read test [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on issue #34212:
URL: https://github.com/apache/beam/issues/34212#issuecomment-3338982741
> I'm having the same issue, tried all the beam sdk versions from` v2.57` to `v2.68` but I still get the same error. As a workaround I monkey-patched the code like this:
>
> # main.py
>
> # Apply Apache Beam trigger patch before importing beam modules
> def _apply_beam_trigger_patch():
> """Apply monkey patch to fix after_synchronized_processing_time KeyError."""
> import apache_beam as beam
> from apache_beam.transforms.trigger import TriggerFn, AfterProcessingTime
>
> original_from_runner_api = TriggerFn.from_runner_api
>
> @staticmethod
> def patched_from_runner_api(proto, context):
> """Patched version that handles after_synchronized_processing_time."""
> from apache_beam.transforms.trigger import (
> AfterAll, AfterAny, AfterEach, AfterWatermark, AfterProcessingTime,
> Always, DefaultTrigger, AfterCount, _Never, OrFinally, Repeatedly
> )
>
> trigger_type = proto.WhichOneof('trigger')
>
> # Handle the special case of after_synchronized_processing_time
> if trigger_type == 'after_synchronized_processing_time':
> return AfterProcessingTime(delay=0)
>
> # Handle all other trigger types
> trigger_mapping = {
> 'after_all': AfterAll,
> 'after_any': AfterAny,
> 'after_each': AfterEach,
> 'after_end_of_window': AfterWatermark,
> 'after_processing_time': AfterProcessingTime,
> 'always': Always,
> 'default': DefaultTrigger,
> 'element_count': AfterCount,
> 'never': _Never,
> 'or_finally': OrFinally,
> 'repeat': Repeatedly,
> }
>
> if trigger_type not in trigger_mapping:
> raise ValueError(f"Unknown trigger type: {trigger_type}")
>
> return trigger_mapping[trigger_type].from_runner_api(proto, context)
>
> TriggerFn.from_runner_api = patched_from_runner_api
>
> # Apply the patch before importing beam
> _apply_beam_trigger_patch()
>
> import apache_beam as beam
>
> # ... the pipeline definition and the rest of my main file goes here after patching
Not a bad hack. We haven't released this change with any version of Beam because we weren't able to execute it. I believe that the next release we could roll this change forward and it would be supported.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Task]: Add Python AfterSynchronizedProcessingTime trigger and add an Iceberg CDC streaming read test [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on issue #34212:
URL: https://github.com/apache/beam/issues/34212#issuecomment-3237881815
Reverted https://github.com/apache/beam/pull/35963 since Runner V2 does not support this yet.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Task]: Add Python AfterSynchronizedProcessingTime trigger and add an Iceberg CDC streaming read test [beam]
Posted by "tarun-google (via GitHub)" <gi...@apache.org>.
tarun-google commented on issue #34212:
URL: https://github.com/apache/beam/issues/34212#issuecomment-3633636347
Yes, we can close this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Task]: Add Python AfterSynchronizedProcessingTime trigger and add an Iceberg CDC streaming read test [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on issue #34212:
URL: https://github.com/apache/beam/issues/34212#issuecomment-2706749991
<details>
<summary>Iceberg source test</summary>
```python
def test_streaming_write_read_cdc_pipeline_using_bqms(self):
runner = self.test_pipeline.get_option('runner')
if not runner or "dataflowrunner" not in runner.lower():
self.skipTest(
"CDC streaming source requires"
"`beam:requirement:pardo:on_window_expiration:v1`, "
"which is currently only supported by the Dataflow runner")
bigquery_client = BigQueryWrapper()
dataset_id = "py_managed_iceberg_bqms_test_" + str(int(time.time()))
project = self.test_pipeline.get_option('project')
bigquery_client.get_or_create_dataset(project, dataset_id)
_LOGGER.info(
"Created dataset %s in project %s", dataset_id, project)
catalog_props = {
"warehouse": self.WAREHOUSE,
"gcp_project": project,
"gcp_location": "us-central1",
"catalog-impl": "org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog",
"io-impl": "org.apache.iceberg.gcp.gcs.GCSFileIO"}
first_table = dataset_id + ".first"
rows = [self._create_row(i) for i in range(3)]
expected_dicts = [row.as_dict() for row in rows]
# Reduce test time by forcing direct runner on pipelines that don't need Dataflow
args_with_directrunner = PipelineOptions(self.args).get_all_options()
args_with_directrunner['runner'] = 'DirectRunner'
# first, prepare an initial table (uses DirectRunner)
with beam.Pipeline(options=PipelineOptions(**args_with_directrunner)) as write_pipeline:
first_write_config = {
"table": first_table,
"catalog_properties": catalog_props
}
_ = (
write_pipeline
| beam.Create(rows)
| beam.managed.Write(beam.managed.ICEBERG, config=first_write_config))
# the intended test: stream cdc read from the first table, validate, and stream records to a second table
# (uses DataflowRunner)
second_table = dataset_id + ".second"
with beam.Pipeline(argv=self.args) as read_pipeline:
first_read_config = {
"table": first_table,
"catalog_properties": catalog_props,
"streaming": True,
"to_timestamp": int((time.time() * 1000))
}
second_write_config = {
"table": second_table,
"catalog_properties": catalog_props,
"triggering_frequency_seconds": 5
}
output_cdc_rows = (
read_pipeline
| beam.managed.Read(beam.managed.ICEBERG_CDC, config=first_read_config))
output_cdc_dicts = output_cdc_rows | beam.Map(lambda row: {"operation": row.operation, "record": row.record._asdict()})
_ = (output_cdc_rows
| "Extract records" >> beam.Map(lambda row: row.record)
.with_output_types(
RowTypeConstraint.from_fields([
("int_", int),
("str_", str),
("bytes_", bytes),
("bool_", bool),
("float_", float)
]))
| beam.managed.Write(beam.managed.ICEBERG, config=second_write_config))
expected_cdc_dicts = [{"record": record, "operation": "append"} for record in expected_dicts]
assert_that(output_cdc_dicts, equal_to(expected_cdc_dicts))
# batch read from the second table and validate records (uses DirectRunner)
with beam.Pipeline(options=PipelineOptions(**args_with_directrunner)) as read_pipeline:
second_read_config = {
"table": second_table,
"catalog_properties": catalog_props
}
output_dicts = (
read_pipeline
| beam.managed.Read(beam.managed.ICEBERG, config=second_read_config)
| beam.Map(lambda row: row._asdict()))
assert_that(output_dicts, equal_to(expected_dicts))
bigquery_client._delete_dataset(project, dataset_id)
```
</details>
The test passes when adding the `trigger.py` changes from #14060.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Task]: Add Python AfterSynchronizedProcessingTime trigger and add an Iceberg CDC streaming read test [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 closed issue #34212: [Task]: Add Python AfterSynchronizedProcessingTime trigger and add an Iceberg CDC streaming read test
URL: https://github.com/apache/beam/issues/34212
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Task]: Add Python AfterSynchronizedProcessingTime trigger and add an Iceberg CDC streaming read test [beam]
Posted by "chamikaramj (via GitHub)" <gi...@apache.org>.
chamikaramj commented on issue #34212:
URL: https://github.com/apache/beam/issues/34212#issuecomment-3358722962
https://github.com/apache/beam/pull/36285 add the AfterSynchronizedProcessingTime trigger so it should be available in Beam 2.69.0.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Task]: Add Python AfterSynchronizedProcessingTime trigger and add an Iceberg CDC streaming read test [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on issue #34212:
URL: https://github.com/apache/beam/issues/34212#issuecomment-3633627866
@tarun-google Should we consider this fixed now?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Task]: Add Python AfterSynchronizedProcessingTime trigger and add an Iceberg CDC streaming read test [beam]
Posted by "marianobrc (via GitHub)" <gi...@apache.org>.
marianobrc commented on issue #34212:
URL: https://github.com/apache/beam/issues/34212#issuecomment-3334874920
I'm having the same issue, tried all the beam sdk versions from` v2.57` to `v2.68` but I still get the same error.
As a workaround I monkey-patched the code like this:
```python
# main.py
# Apply Apache Beam trigger patch before importing beam modules
def _apply_beam_trigger_patch():
"""Apply monkey patch to fix after_synchronized_processing_time KeyError."""
import apache_beam as beam
from apache_beam.transforms.trigger import TriggerFn, AfterProcessingTime
original_from_runner_api = TriggerFn.from_runner_api
@staticmethod
def patched_from_runner_api(proto, context):
"""Patched version that handles after_synchronized_processing_time."""
from apache_beam.transforms.trigger import (
AfterAll, AfterAny, AfterEach, AfterWatermark, AfterProcessingTime,
Always, DefaultTrigger, AfterCount, _Never, OrFinally, Repeatedly
)
trigger_type = proto.WhichOneof('trigger')
# Handle the special case of after_synchronized_processing_time
if trigger_type == 'after_synchronized_processing_time':
return AfterProcessingTime(delay=0)
# Handle all other trigger types
trigger_mapping = {
'after_all': AfterAll,
'after_any': AfterAny,
'after_each': AfterEach,
'after_end_of_window': AfterWatermark,
'after_processing_time': AfterProcessingTime,
'always': Always,
'default': DefaultTrigger,
'element_count': AfterCount,
'never': _Never,
'or_finally': OrFinally,
'repeat': Repeatedly,
}
if trigger_type not in trigger_mapping:
raise ValueError(f"Unknown trigger type: {trigger_type}")
return trigger_mapping[trigger_type].from_runner_api(proto, context)
TriggerFn.from_runner_api = patched_from_runner_api
# Apply the patch before importing beam
_apply_beam_trigger_patch()
import apache_beam as beam
# ... the pipeline definition and the rest of my main file goes here after patching
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org