[
https://issues.apache.org/jira/browse/BEAM-1833?focusedWorklogId=628075&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-628075
]
ASF GitHub Bot logged work on BEAM-1833:
----------------------------------------
Author: ASF GitHub Bot
Created on: 26/Jul/21 23:59
Start Date: 26/Jul/21 23:59
Worklog Time Spent: 10m
Work Description: robertwb commented on a change in pull request #15202:
URL: https://github.com/apache/beam/pull/15202#discussion_r677019161
##########
File path: sdks/python/apache_beam/pipeline.py
##########
@@ -670,15 +671,18 @@ def apply(
pvalueish, inputs = transform._extract_input_pvalues(pvalueish)
try:
- inputs = tuple(inputs)
- for leaf_input in inputs:
- if not isinstance(leaf_input, pvalue.PValue):
- raise TypeError
+ if not isinstance(inputs, dict):
+ inputs = {str(ix): input for (ix, input) in enumerate(inputs)}
except TypeError:
Review comment:
This catches a possible failure in enumerate.
##########
File path: sdks/python/apache_beam/pipeline.py
##########
@@ -437,11 +438,11 @@ def visit_transform(self, transform_node):
output_replacements[transform_node].append((tag, replacement))
if replace_input:
- new_input = [
- input if not input in output_map else output_map[input]
- for input in transform_node.inputs
- ]
- input_replacements[transform_node] = new_input
+ new_inputs = {
+ tag: input if not input in output_map else output_map[input]
+ for (tag, input) in transform_node.main_inputs.items()
+ }
+ input_replacements[transform_node] = new_inputs
Review comment:
Good catch. Done.
##########
File path: sdks/python/apache_beam/transforms/ptransform.py
##########
@@ -262,16 +263,21 @@ def get_named_nested_pvalues(pvalueish):
else:
tagged_values = enumerate(pvalueish)
elif isinstance(pvalueish, list):
+ if as_inputs:
+ yield None, pvalueish
Review comment:
This is for backwards compatibility with the "eager" mode (e.g. `[1, 2,
3] | beam.Map(lambda x: x*x)`)
##########
File path: sdks/python/apache_beam/transforms/ptransform.py
##########
@@ -253,7 +254,7 @@ def visit(self, node):
return self.visit_nested(node)
-def get_named_nested_pvalues(pvalueish):
+def get_named_nested_pvalues(pvalueish, as_inputs=False):
Review comment:
This is done elsewhere as needed.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 628075)
Time Spent: 7h 20m (was: 7h 10m)
> Restructure Python pipeline construction to better follow the Runner API
> ------------------------------------------------------------------------
>
> Key: BEAM-1833
> URL: https://issues.apache.org/jira/browse/BEAM-1833
> Project: Beam
> Issue Type: Improvement
> Components: sdk-py-core
> Reporter: Robert Bradshaw
> Assignee: Sam Rohde
> Priority: P2
> Fix For: 2.20.0
>
> Time Spent: 7h 20m
> Remaining Estimate: 0h
>
> The most important part is removing the runner.apply overrides, but there are
> also various other improvements (e.g. all inputs and outputs should be named).
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
