You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2020/03/30 23:41:01 UTC
[GitHub] [beam] pabloem opened a new pull request #11270: [WIP] Improvements
for FnApiRunner
pabloem opened a new pull request #11270: [WIP] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270
**Please** add a meaningful description for your change here
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Apex | Dataflow | Flink | Gearpump | Samza | Spark
--- | --- | --- | --- | --- | --- | --- | ---
Go | [](https://builds.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/)
Java | [](https://builds.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Apex/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Gearpump/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/)
Python | [](https://builds.apache.org/job/beam_PostCommit_Python2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/) | --- | [](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python2_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/)
XLang | --- | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/)
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website
--- | --- | --- | --- | ---
Non-portable | [](https://builds.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/)
Portable | --- | [](https://builds.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | ---
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11270: [BEAM-9639][BEAM-9608]
Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#issuecomment-607428460
this latest commit is purely aesthetic
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11270:
[BEAM-9639][BEAM-9608] Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#discussion_r407666877
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/translations.py
##########
@@ -149,15 +155,26 @@ def no_overlap(a, b):
return (
not consumer.forced_root and not self in consumer.must_follow and
not self.is_runner_urn(context) and
- not consumer.is_runner_urn(context) and
- no_overlap(self.downstream_side_inputs, consumer.side_inputs()))
+ not consumer.is_runner_urn(context) and no_overlap(
+ set(self.downstream_side_inputs.keys()),
+ {i
+ for i, _, _ in consumer.side_inputs()}))
+
+ def _fuse_downstream_side_inputs(self, other):
+ res = dict(self.downstream_side_inputs)
+ for si, other_si_ids in other.downstream_side_inputs.items():
+ if si in res:
+ res[si] = union(res[si], other_si_ids)
+ else:
+ res[si] = other_si_ids
+ return res
def fuse(self, other):
# type: (Stage) -> Stage
return Stage(
"(%s)+(%s)" % (self.name, other.name),
self.transforms + other.transforms,
- union(self.downstream_side_inputs, other.downstream_side_inputs),
+ self._fuse_downstream_side_inputs(other),
Review comment:
Nit: this sounds like it mutates self.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11270: [BEAM-9639][BEAM-9608]
Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#issuecomment-615521653
Building the side input index elsewhere. LMK what you think.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11270: [BEAM-9639][BEAM-9608]
Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#issuecomment-614228294
Run Python PreCommit
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on a change in pull request #11270:
[BEAM-9639][BEAM-9608] Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#discussion_r408507669
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/translations.py
##########
@@ -149,15 +155,26 @@ def no_overlap(a, b):
return (
not consumer.forced_root and not self in consumer.must_follow and
not self.is_runner_urn(context) and
- not consumer.is_runner_urn(context) and
- no_overlap(self.downstream_side_inputs, consumer.side_inputs()))
+ not consumer.is_runner_urn(context) and no_overlap(
+ set(self.downstream_side_inputs.keys()),
+ {i
+ for i, _, _ in consumer.side_inputs()}))
+
+ def _fuse_downstream_side_inputs(self, other):
+ res = dict(self.downstream_side_inputs)
+ for si, other_si_ids in other.downstream_side_inputs.items():
+ if si in res:
+ res[si] = union(res[si], other_si_ids)
+ else:
+ res[si] = other_si_ids
+ return res
def fuse(self, other):
# type: (Stage) -> Stage
return Stage(
"(%s)+(%s)" % (self.name, other.name),
self.transforms + other.transforms,
- union(self.downstream_side_inputs, other.downstream_side_inputs),
+ self._fuse_downstream_side_inputs(other),
Review comment:
renamed to `_get_fused_downstream_side_inputs` thoughts?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11270:
[BEAM-9639][BEAM-9608] Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#discussion_r407667700
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/translations.py
##########
@@ -75,21 +75,27 @@
IMPULSE_BUFFER = b'impulse'
+# SideInputId is identified by a consumer ParDo + tag.
+SideInputId = Tuple[str, str]
+
+DataSideInput = Dict[SideInputId,
Review comment:
What does the value represent?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11270: [BEAM-9639][BEAM-9608]
Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#issuecomment-614184800
Run Python PreCommit
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on a change in pull request #11270:
[BEAM-9639][BEAM-9608] Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#discussion_r408507854
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py
##########
@@ -326,8 +327,8 @@ def commit_side_inputs_to_state(
data_side_input, # type: DataSideInput
):
# type: (...) -> None
- for (consuming_transform_id, tag), (buffer_id, func_spec) \
- in data_side_input.items():
+ for (consuming_transform_id, tag), (buffer_id,
Review comment:
it did not : ( hehe
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11270: [BEAM-9639][BEAM-9608]
Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#issuecomment-612497517
Run Python PreCommit
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11270: [BEAM-9639][BEAM-9608]
Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#issuecomment-614269899
Run Python PreCommit
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11270:
[BEAM-9639][BEAM-9608] Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#discussion_r407663916
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py
##########
@@ -367,6 +413,73 @@ def _build_process_bundle_descriptor(self):
state_api_service_descriptor=self.state_api_service_descriptor(),
timer_api_service_descriptor=self.data_api_service_descriptor())
+ def commit_output_views_to_state(self):
+ """Commit bundle outputs to state to be consumed as side inputs later.
+
+ Only the outputs that should be side inputs are committed to state.
+ """
+ data_side_input = {} # type: DataSideInput
+ for pcoll, si_ids in self.stage.downstream_side_inputs.items():
+ for (consumer_transform_name, tag), access_pattern in si_ids.items():
+ data_side_input[consumer_transform_name, tag] = (
+ translations.create_buffer_id(pcoll), access_pattern)
+ self.execution_context.commit_side_inputs_to_state(data_side_input)
+
+ def extract_bundle_inputs(self):
+ # type: (...) -> Tuple[Dict[str, PartitionableBuffer], DataOutput]
+
+ """Returns maps of transform names to PCollection identifiers.
+
+ Also mutates IO stages to point to the data ApiServiceDescriptor.
+
+ Returns:
+ A tuple of (data_input, data_output) dictionaries.
+ `data_input` is a dictionary mapping (transform_name, output_name) to a
+ PCollection buffer; `data_output` is a dictionary mapping
+ (transform_name, output_name) to a PCollection ID.
+ """
+ data_input = {} # type: Dict[str, PartitionableBuffer]
+ data_output = {} # type: DataOutput
+ # A mapping of {(transform_id, timer_family_id) : buffer_id}
+ expected_timer_output = {} # type: Dict[Tuple(str, str), str]
+ for transform in self.stage.transforms:
+ if transform.spec.urn in (bundle_processor.DATA_INPUT_URN,
+ bundle_processor.DATA_OUTPUT_URN):
+ pcoll_id = transform.spec.payload
+ if transform.spec.urn == bundle_processor.DATA_INPUT_URN:
+ coder_id = self.execution_context.data_channel_coders[only_element(
+ transform.outputs.values())]
+ coder = self.execution_context.pipeline_context.coders[
+ self.execution_context.safe_coders.get(coder_id, coder_id)]
+ if pcoll_id == translations.IMPULSE_BUFFER:
+ data_input[transform.unique_name] = ListBuffer(
+ coder_impl=coder.get_impl())
+ data_input[transform.unique_name].append(ENCODED_IMPULSE_VALUE)
+ else:
+ if pcoll_id not in self.execution_context.pcoll_buffers:
+ self.execution_context.pcoll_buffers[pcoll_id] = ListBuffer(
+ coder_impl=coder.get_impl())
+ data_input[transform.unique_name] = \
+ self.execution_context.pcoll_buffers[pcoll_id]
+ elif transform.spec.urn == bundle_processor.DATA_OUTPUT_URN:
+ data_output[transform.unique_name] = pcoll_id
+ coder_id = self.execution_context.data_channel_coders[only_element(
+ transform.inputs.values())]
+ else:
+ raise NotImplementedError
+ data_spec = beam_fn_api_pb2.RemoteGrpcPort(coder_id=coder_id)
+ data_api_service_descriptor = \
+ self.data_api_service_descriptor()
+ if data_api_service_descriptor:
+ data_spec.api_service_descriptor.url = (
+ data_api_service_descriptor.url)
+ transform.spec.payload = data_spec.SerializeToString()
+ elif transform.spec.urn in translations.PAR_DO_URNS:
+ for timer_family_id in payload.timer_family_specs.keys():
+ expected_timer_output[(transform.unique_name, timer_family_id)] = (
+ create_buffer_id(timer_family_id, 'timers'))
+ return data_input, data_output, expected_timer_output
Review comment:
Update docs to match.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11270: [WIP] Improvements for
FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11270: [WIP] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#issuecomment-606919524
Run Portable_Python PreCommit
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11270: [BEAM-9639][BEAM-9608]
Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#issuecomment-612497313
Run Python PreCommit
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11270: [BEAM-9639][BEAM-9608]
Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#issuecomment-612271558
attempting to rebase. let's see how far that takes me...
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11270: [BEAM-9639][BEAM-9608]
Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#issuecomment-615628912
Run Python PreCommit
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on a change in pull request #11270:
[BEAM-9639][BEAM-9608] Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#discussion_r408507730
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py
##########
@@ -367,6 +413,73 @@ def _build_process_bundle_descriptor(self):
state_api_service_descriptor=self.state_api_service_descriptor(),
timer_api_service_descriptor=self.data_api_service_descriptor())
+ def commit_output_views_to_state(self):
+ """Commit bundle outputs to state to be consumed as side inputs later.
+
+ Only the outputs that should be side inputs are committed to state.
+ """
+ data_side_input = {} # type: DataSideInput
+ for pcoll, si_ids in self.stage.downstream_side_inputs.items():
+ for (consumer_transform_name, tag), access_pattern in si_ids.items():
+ data_side_input[consumer_transform_name, tag] = (
+ translations.create_buffer_id(pcoll), access_pattern)
+ self.execution_context.commit_side_inputs_to_state(data_side_input)
+
+ def extract_bundle_inputs(self):
+ # type: (...) -> Tuple[Dict[str, PartitionableBuffer], DataOutput]
+
+ """Returns maps of transform names to PCollection identifiers.
+
+ Also mutates IO stages to point to the data ApiServiceDescriptor.
+
+ Returns:
+ A tuple of (data_input, data_output) dictionaries.
+ `data_input` is a dictionary mapping (transform_name, output_name) to a
+ PCollection buffer; `data_output` is a dictionary mapping
+ (transform_name, output_name) to a PCollection ID.
+ """
+ data_input = {} # type: Dict[str, PartitionableBuffer]
+ data_output = {} # type: DataOutput
+ # A mapping of {(transform_id, timer_family_id) : buffer_id}
+ expected_timer_output = {} # type: Dict[Tuple(str, str), str]
+ for transform in self.stage.transforms:
+ if transform.spec.urn in (bundle_processor.DATA_INPUT_URN,
+ bundle_processor.DATA_OUTPUT_URN):
+ pcoll_id = transform.spec.payload
+ if transform.spec.urn == bundle_processor.DATA_INPUT_URN:
+ coder_id = self.execution_context.data_channel_coders[only_element(
+ transform.outputs.values())]
+ coder = self.execution_context.pipeline_context.coders[
+ self.execution_context.safe_coders.get(coder_id, coder_id)]
+ if pcoll_id == translations.IMPULSE_BUFFER:
+ data_input[transform.unique_name] = ListBuffer(
+ coder_impl=coder.get_impl())
+ data_input[transform.unique_name].append(ENCODED_IMPULSE_VALUE)
+ else:
+ if pcoll_id not in self.execution_context.pcoll_buffers:
+ self.execution_context.pcoll_buffers[pcoll_id] = ListBuffer(
+ coder_impl=coder.get_impl())
+ data_input[transform.unique_name] = \
+ self.execution_context.pcoll_buffers[pcoll_id]
+ elif transform.spec.urn == bundle_processor.DATA_OUTPUT_URN:
+ data_output[transform.unique_name] = pcoll_id
+ coder_id = self.execution_context.data_channel_coders[only_element(
+ transform.inputs.values())]
+ else:
+ raise NotImplementedError
+ data_spec = beam_fn_api_pb2.RemoteGrpcPort(coder_id=coder_id)
+ data_api_service_descriptor = \
+ self.data_api_service_descriptor()
+ if data_api_service_descriptor:
+ data_spec.api_service_descriptor.url = (
+ data_api_service_descriptor.url)
+ transform.spec.payload = data_spec.SerializeToString()
+ elif transform.spec.urn in translations.PAR_DO_URNS:
+ for timer_family_id in payload.timer_family_specs.keys():
+ expected_timer_output[(transform.unique_name, timer_family_id)] = (
+ create_buffer_id(timer_family_id, 'timers'))
+ return data_input, data_output, expected_timer_output
Review comment:
done
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11270:
[BEAM-9639][BEAM-9608] Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#discussion_r407668033
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/translations.py
##########
@@ -149,15 +155,26 @@ def no_overlap(a, b):
return (
not consumer.forced_root and not self in consumer.must_follow and
not self.is_runner_urn(context) and
- not consumer.is_runner_urn(context) and
- no_overlap(self.downstream_side_inputs, consumer.side_inputs()))
+ not consumer.is_runner_urn(context) and no_overlap(
+ set(self.downstream_side_inputs.keys()),
+ {i
+ for i, _, _ in consumer.side_inputs()}))
+
+ def _fuse_downstream_side_inputs(self, other):
+ res = dict(self.downstream_side_inputs)
+ for si, other_si_ids in other.downstream_side_inputs.items():
+ if si in res:
+ res[si] = union(res[si], other_si_ids)
Review comment:
So this is actually a dict mapping to sets?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11270: [BEAM-9639][BEAM-9608]
Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#issuecomment-614268696
failed test is streaming wordcount test
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on a change in pull request #11270:
[BEAM-9639][BEAM-9608] Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#discussion_r408507693
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py
##########
@@ -367,6 +413,73 @@ def _build_process_bundle_descriptor(self):
state_api_service_descriptor=self.state_api_service_descriptor(),
timer_api_service_descriptor=self.data_api_service_descriptor())
+ def commit_output_views_to_state(self):
Review comment:
done
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on a change in pull request #11270:
[BEAM-9639][BEAM-9608] Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#discussion_r408508964
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/translations.py
##########
@@ -75,21 +75,27 @@
IMPULSE_BUFFER = b'impulse'
+# SideInputId is identified by a consumer ParDo + tag.
+SideInputId = Tuple[str, str]
+
+DataSideInput = Dict[SideInputId,
+ Tuple[bytes, beam_runner_api_pb2.FunctionSpec]]
+
class Stage(object):
"""A set of Transforms that can be sent to the worker for processing."""
def __init__(self,
name, # type: str
transforms, # type: List[beam_runner_api_pb2.PTransform]
- downstream_side_inputs=None, # type: Optional[FrozenSet[str]]
+ downstream_side_inputs=None, # type: Optional[Dict[str, SideInputId]]
Review comment:
Hm so this change breaks that, so the memory requirements would be larger. I would think that they would not be too bad, since most graphs don't have many side inputs going many places. What do you think? I'm willing to find a better solution for this, but I wonder if it's worth the extra time.
The reason that this is made into a dict is to contain more information about downstream side inputs. specifically, it contains which transforms will consume the side inputs. this is used to commit the side inputs to state after they are calculated (rather than before they are consumed). This will be necessary for streaming, because side inputs will need to be added to state as they are computed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11270:
[BEAM-9639][BEAM-9608] Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#discussion_r407673895
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py
##########
@@ -914,14 +898,16 @@ def process_bundle(self,
expected_outputs, # type: DataOutput
fired_timers, # type: Mapping[Tuple[str, str], PartitionableBuffer]
expected_output_timers # type: Dict[str, Dict[str, str]]
+ dry_run=False
Review comment:
This should be the default, we shouldn't have to pass it.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11270: [BEAM-9639][BEAM-9608]
Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#issuecomment-610150292
@robertwb ptal
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11270:
[BEAM-9639][BEAM-9608] Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#discussion_r407664777
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py
##########
@@ -367,6 +413,73 @@ def _build_process_bundle_descriptor(self):
state_api_service_descriptor=self.state_api_service_descriptor(),
timer_api_service_descriptor=self.data_api_service_descriptor())
+ def commit_output_views_to_state(self):
Review comment:
Not sure what "output views" means. Maybe call this commit_side_inputs_to_state as well?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11270: [BEAM-9639][BEAM-9608]
Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#issuecomment-612513921
Run Python PreCommit
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on a change in pull request #11270:
[BEAM-9639][BEAM-9608] Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#discussion_r408507610
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/translations.py
##########
@@ -75,21 +75,27 @@
IMPULSE_BUFFER = b'impulse'
+# SideInputId is identified by a consumer ParDo + tag.
+SideInputId = Tuple[str, str]
+
+DataSideInput = Dict[SideInputId,
Review comment:
The value is a tuple with the encoded data. I've moved these to translations.py, and updated the comment
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on a change in pull request #11270:
[BEAM-9639][BEAM-9608] Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#discussion_r408507780
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py
##########
@@ -367,6 +413,73 @@ def _build_process_bundle_descriptor(self):
state_api_service_descriptor=self.state_api_service_descriptor(),
timer_api_service_descriptor=self.data_api_service_descriptor())
+ def commit_output_views_to_state(self):
+ """Commit bundle outputs to state to be consumed as side inputs later.
+
+ Only the outputs that should be side inputs are committed to state.
+ """
+ data_side_input = {} # type: DataSideInput
+ for pcoll, si_ids in self.stage.downstream_side_inputs.items():
+ for (consumer_transform_name, tag), access_pattern in si_ids.items():
+ data_side_input[consumer_transform_name, tag] = (
+ translations.create_buffer_id(pcoll), access_pattern)
+ self.execution_context.commit_side_inputs_to_state(data_side_input)
+
+ def extract_bundle_inputs(self):
Review comment:
done
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11270: [WIP] Improvements for
FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11270: [WIP] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#issuecomment-607028810
Run Python PreCommit
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on a change in pull request #11270:
[BEAM-9639][BEAM-9608] Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#discussion_r408507558
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/translations.py
##########
@@ -149,15 +155,26 @@ def no_overlap(a, b):
return (
not consumer.forced_root and not self in consumer.must_follow and
not self.is_runner_urn(context) and
- not consumer.is_runner_urn(context) and
- no_overlap(self.downstream_side_inputs, consumer.side_inputs()))
+ not consumer.is_runner_urn(context) and no_overlap(
+ set(self.downstream_side_inputs.keys()),
+ {i
+ for i, _, _ in consumer.side_inputs()}))
+
+ def _fuse_downstream_side_inputs(self, other):
+ res = dict(self.downstream_side_inputs)
+ for si, other_si_ids in other.downstream_side_inputs.items():
+ if si in res:
+ res[si] = union(res[si], other_si_ids)
Review comment:
woah this is a bug.
downstream side input is a dictionary mapping to dictionaries.
Dict[Output Pcollection, Dict[Side input ID, Access pattern]]
Where Side input ID is Tuple[consumer ptransform, input index]. Added appropriate annotations, and fixed the bug.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11270:
[BEAM-9639][BEAM-9608] Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#discussion_r407663602
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py
##########
@@ -367,6 +413,73 @@ def _build_process_bundle_descriptor(self):
state_api_service_descriptor=self.state_api_service_descriptor(),
timer_api_service_descriptor=self.data_api_service_descriptor())
+ def commit_output_views_to_state(self):
+ """Commit bundle outputs to state to be consumed as side inputs later.
+
+ Only the outputs that should be side inputs are committed to state.
+ """
+ data_side_input = {} # type: DataSideInput
+ for pcoll, si_ids in self.stage.downstream_side_inputs.items():
+ for (consumer_transform_name, tag), access_pattern in si_ids.items():
+ data_side_input[consumer_transform_name, tag] = (
+ translations.create_buffer_id(pcoll), access_pattern)
+ self.execution_context.commit_side_inputs_to_state(data_side_input)
+
+ def extract_bundle_inputs(self):
Review comment:
extract_bundle_inputs_and_outputs?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on a change in pull request #11270:
[BEAM-9639][BEAM-9608] Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#discussion_r408507914
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py
##########
@@ -914,14 +898,16 @@ def process_bundle(self,
expected_outputs, # type: DataOutput
fired_timers, # type: Mapping[Tuple[str, str], PartitionableBuffer]
expected_output_timers # type: Dict[str, Dict[str, str]]
+ dry_run=False
Review comment:
Done
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11270: [BEAM-9639][BEAM-9608]
Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#issuecomment-607477435
Run Python PreCommit
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11270:
[BEAM-9639][BEAM-9608] Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#discussion_r407670612
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/translations.py
##########
@@ -75,21 +75,27 @@
IMPULSE_BUFFER = b'impulse'
+# SideInputId is identified by a consumer ParDo + tag.
+SideInputId = Tuple[str, str]
+
+DataSideInput = Dict[SideInputId,
+ Tuple[bytes, beam_runner_api_pb2.FunctionSpec]]
+
class Stage(object):
"""A set of Transforms that can be sent to the worker for processing."""
def __init__(self,
name, # type: str
transforms, # type: List[beam_runner_api_pb2.PTransform]
- downstream_side_inputs=None, # type: Optional[FrozenSet[str]]
+ downstream_side_inputs=None, # type: Optional[Dict[str, SideInputId]]
Review comment:
The goal of this (which, yes, should have been better documented) is to quickly be able to prohibit fusion. But the reason we defined our own union was so that memory didn't grow as O(n^2) in the common case because many stages were able to share this set (rather than have their own copy). These changes seem to break that.
Also, could you clarify why this was made into a dict?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11270:
[BEAM-9639][BEAM-9608] Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#discussion_r407620241
##########
File path: sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py
##########
@@ -326,8 +327,8 @@ def commit_side_inputs_to_state(
data_side_input, # type: DataSideInput
):
# type: (...) -> None
- for (consuming_transform_id, tag), (buffer_id, func_spec) \
- in data_side_input.items():
+ for (consuming_transform_id, tag), (buffer_id,
Review comment:
I wonder if
`((consuming_transform_id, tag), (buffer_id, func_spec))`
would make both yapf and humans happy.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11270: [BEAM-9639][BEAM-9608]
Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#issuecomment-615629278
Run Python PreCommit
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11270: [BEAM-9639][BEAM-9608]
Improvements for FnApiRunner
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11270: [BEAM-9639][BEAM-9608] Improvements for FnApiRunner
URL: https://github.com/apache/beam/pull/11270#issuecomment-614275628
Run Python PreCommit
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services