GitHub Actions / Test Results
failed
Oct 30, 2025 in 0s
3 fail, 262 skipped, 3 939 pass in 1h 9m 40s
4 files 4 suites 1h 9m 40s ⏱️
4 204 tests 3 939 ✅ 262 💤 3 ❌
8 368 runs 7 449 ✅ 916 💤 3 ❌
Results for commit bb0ae61.
Annotations
Check warning on line 0 in apache_beam.yaml.examples.testing.examples_test.ElementWiseTest
github-actions / Test Results
1 out of 2 runs failed: test_map_to_fields_with_deps_yaml (apache_beam.yaml.examples.testing.examples_test.ElementWiseTest)
sdks/python/test-suites/tox/py311/build/srcs/sdks/python/pytest_py311_no_xdist.xml [took 4m 36s]
Raw output
ValueError: Error applying transform "ToRoman" at line 30: Traceback (most recent call last):
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/portability/expansion_service.py", line 101, in Expand
result = pipeline.apply(
^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pipeline.py", line 746, in apply
return self.apply(transform, pvalueish)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pipeline.py", line 826, in apply
pvalueish_result = self.runner.apply(transform, pvalueish, self._options)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 191, in apply
return self.apply_PTransform(transform, input, options)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 195, in apply_PTransform
return transform.expand(input)
^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/transforms/fully_qualified_named_transform.py", line 78, in expand
return pinput | transform
~~~~~~~^~~~~~~~~~~
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pvalue.py", line 139, in __or__
return self.pipeline.apply(ptransform, self)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pipeline.py", line 826, in apply
pvalueish_result = self.runner.apply(transform, pvalueish, self._options)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 191, in apply
return self.apply_PTransform(transform, input, options)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 195, in apply_PTransform
return transform.expand(input)
^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/transforms/ptransform.py", line 1032, in expand
return self._fn(pcoll, *args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_errors.py", line 75, in expand
return transform_fn(wrapped_pcoll, **kwargs).as_result(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 718, in _PyJsMapToFields
**{
^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 719, in <dictcomp>
name: _as_callable(
^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 396, in _as_callable
func = _expand_python_mapping_func(original_fields, **expr)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 311, in _expand_python_mapping_func
return python_callable.PythonCallableWithSource(source)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/utils/python_callable.py", line 48, in __init__
self._callable = self.load_from_source(source)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/utils/python_callable.py", line 60, in load_from_source
return cls.load_from_script(source)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/utils/python_callable.py", line 106, in load_from_script
exec('\n'.join(lines), exec_globals)
File "<string>", line 1, in <module>
ModuleNotFoundError: No module named 'roman'
spec = {'__line__': 30, '__uuid__': '5eecdff8-725a-4998-ba8d-975cf5189f1f', 'config': {'__line__': 33, '__uuid__': 'fed16bb0-...rt(row):\n return roman.toRoman(row.year)\n'}}, ...}, 'input': {'input': 'c7cfe535-15a6-4a6d-a975-c6c31a6ecd6e'}, ...}
scope = <apache_beam.yaml.yaml_transform.Scope object at 0x7cc28ca4d8d0>
def expand_leaf_transform(spec, scope):
spec = spec.copy()
# Check for optional output_schema to verify on.
# The idea is to pass this output_schema config to the ValidateWithSchema
# transform.
output_schema_spec = {}
if 'output_schema' in spec.get('config', {}):
output_schema_spec = spec.get('config').pop('output_schema')
spec = normalize_inputs_outputs(spec)
inputs_dict = {
key: scope.get_pcollection(value)
for (key, value) in empty_if_explicitly_empty(spec['input']).items()
}
input_type = spec.get('input_type', 'default')
if input_type == 'list':
inputs = tuple(inputs_dict.values())
elif input_type == 'map':
inputs = inputs_dict
else:
if len(inputs_dict) == 0:
inputs = scope.root
elif len(inputs_dict) == 1:
inputs = next(iter(inputs_dict.values()))
else:
inputs = inputs_dict
_LOGGER.info("Expanding %s ", identify_object(spec))
ptransform = scope.create_ptransform(spec, inputs_dict.values())
try:
# TODO: Move validation to construction?
with FullyQualifiedNamedTransform.with_filter('*'):
> outputs = inputs | scope.unique_name(spec, ptransform) >> ptransform
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/yaml/yaml_transform.py:519:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
target/.tox-py311/py311/lib/python3.11/site-packages/apache_beam/pvalue.py:139: in __or__
return self.pipeline.apply(ptransform, self)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
target/.tox-py311/py311/lib/python3.11/site-packages/apache_beam/pipeline.py:731: in apply
return self.apply(
target/.tox-py311/py311/lib/python3.11/site-packages/apache_beam/pipeline.py:746: in apply
return self.apply(transform, pvalueish)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
target/.tox-py311/py311/lib/python3.11/site-packages/apache_beam/pipeline.py:826: in apply
pvalueish_result = self.runner.apply(transform, pvalueish, self._options)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
target/.tox-py311/py311/lib/python3.11/site-packages/apache_beam/runners/runner.py:191: in apply
return self.apply_PTransform(transform, input, options)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
target/.tox-py311/py311/lib/python3.11/site-packages/apache_beam/runners/runner.py:195: in apply_PTransform
return transform.expand(input)
^^^^^^^^^^^^^^^^^^^^^^^
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <ExternalTransform(PTransform) label=[ExternalTransform(beam:transforms:python:fully_qualified_named)] at 0x7cc28e1def90>
pvalueish = <PCollection[Create/Map(decode).None] at 0x7cc28ca03790>
def expand(self, pvalueish: pvalue.PCollection) -> pvalue.PCollection:
if isinstance(pvalueish, pvalue.PBegin):
self._inputs = {}
elif isinstance(pvalueish, (list, tuple)):
self._inputs = {str(ix): pvalue for ix, pvalue in enumerate(pvalueish)}
elif isinstance(pvalueish, dict):
self._inputs = pvalueish
else:
self._inputs = {'input': pvalueish}
pipeline = (
next(iter(self._inputs.values())).pipeline
if self._inputs else pvalueish.pipeline)
context = pipeline_context.PipelineContext(
component_id_map=pipeline.component_id_map)
transform_proto = beam_runner_api_pb2.PTransform(
unique_name=pipeline._current_transform().full_label,
spec=beam_runner_api_pb2.FunctionSpec(
urn=self._urn, payload=self._payload))
for tag, pcoll in self._inputs.items():
transform_proto.inputs[tag] = context.pcollections.get_id(pcoll)
# Conversion to/from proto assumes producers.
# TODO: Possibly loosen this.
context.transforms.put_proto(
'%s_%s' % (self._IMPULSE_PREFIX, tag),
beam_runner_api_pb2.PTransform(
unique_name='%s_%s' % (self._IMPULSE_PREFIX, tag),
spec=beam_runner_api_pb2.FunctionSpec(
urn=common_urns.primitives.IMPULSE.urn),
outputs={'out': transform_proto.inputs[tag]}))
output_coders = None
if self._type_hints.output_types:
if self._type_hints.output_types[0]:
output_coders = dict(
(str(k), context.coder_id_from_element_type(v))
for (k, v) in enumerate(self._type_hints.output_types[0]))
elif self._type_hints.output_types[1]:
output_coders = {
k: context.coder_id_from_element_type(v)
for (k, v) in self._type_hints.output_types[1].items()
}
components = context.to_runner_api()
request = beam_expansion_api_pb2.ExpansionRequest(
components=components,
namespace=self._external_namespace, # type: ignore[arg-type]
transform=transform_proto,
output_coder_requests=output_coders,
pipeline_options=pipeline._options.to_runner_api())
expansion_service = _maybe_use_transform_service(
self._expansion_service, pipeline.options)
with ExternalTransform.service(expansion_service) as service:
response = service.Expand(request)
if response.error:
> raise RuntimeError(_sanitize_java_traceback(response.error))
E RuntimeError: Traceback (most recent call last):
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/portability/expansion_service.py", line 101, in Expand
E result = pipeline.apply(
E ^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pipeline.py", line 746, in apply
E return self.apply(transform, pvalueish)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pipeline.py", line 826, in apply
E pvalueish_result = self.runner.apply(transform, pvalueish, self._options)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 191, in apply
E return self.apply_PTransform(transform, input, options)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 195, in apply_PTransform
E return transform.expand(input)
E ^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/transforms/fully_qualified_named_transform.py", line 78, in expand
E return pinput | transform
E ~~~~~~~^~~~~~~~~~~
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pvalue.py", line 139, in __or__
E return self.pipeline.apply(ptransform, self)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pipeline.py", line 826, in apply
E pvalueish_result = self.runner.apply(transform, pvalueish, self._options)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 191, in apply
E return self.apply_PTransform(transform, input, options)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 195, in apply_PTransform
E return transform.expand(input)
E ^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/transforms/ptransform.py", line 1032, in expand
E return self._fn(pcoll, *args, **kwargs)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_errors.py", line 75, in expand
E return transform_fn(wrapped_pcoll, **kwargs).as_result(
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 718, in _PyJsMapToFields
E **{
E ^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 719, in <dictcomp>
E name: _as_callable(
E ^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 396, in _as_callable
E func = _expand_python_mapping_func(original_fields, **expr)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 311, in _expand_python_mapping_func
E return python_callable.PythonCallableWithSource(source)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/utils/python_callable.py", line 48, in __init__
E self._callable = self.load_from_source(source)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/utils/python_callable.py", line 60, in load_from_source
E return cls.load_from_script(source)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/utils/python_callable.py", line 106, in load_from_script
E exec('\n'.join(lines), exec_globals)
E File "<string>", line 1, in <module>
E ModuleNotFoundError: No module named 'roman'
target/.tox-py311/py311/lib/python3.11/site-packages/apache_beam/transforms/external.py:832: RuntimeError
The above exception was the direct cause of the following exception:
self = <apache_beam.yaml.examples.testing.examples_test.ElementwiseExamplesTest testMethod=test_map_to_fields_with_deps_yaml>
@mock.patch('apache_beam.Pipeline', TestPipeline)
def test_yaml_example(self):
with open(pipeline_spec_file, encoding="utf-8") as f:
lines = f.readlines()
expected_key = '# Expected:\n'
if expected_key in lines:
expected = lines[lines.index('# Expected:\n') + 1:]
else:
raise ValueError(
f"Missing '# Expected:' tag in example file '{pipeline_spec_file}'")
for i, line in enumerate(expected):
expected[i] = line.replace('# ', '').replace('\n', '')
expected = [line for line in expected if line]
raw_spec_string = ''.join(lines)
# Filter for any jinja preprocessor - this has to be done before other
# preprocessors.
jinja_preprocessor = [
preprocessor for preprocessor in custom_preprocessors
if 'jinja_preprocessor' in preprocessor.__name__
]
if jinja_preprocessor:
jinja_preprocessor = jinja_preprocessor[0]
raw_spec_string = jinja_preprocessor(
raw_spec_string, self._testMethodName)
custom_preprocessors.remove(jinja_preprocessor)
pipeline_spec = yaml.load(
raw_spec_string, Loader=yaml_transform.SafeLineLoader)
with TestEnvironment() as env:
for fn in custom_preprocessors:
pipeline_spec = fn(pipeline_spec, expected, env)
with beam.Pipeline(options=PipelineOptions(
pickle_library='cloudpickle',
**yaml_transform.SafeLineLoader.strip_metadata(pipeline_spec.get(
'options', {})))) as p:
actual = [
> yaml_transform.expand_pipeline(
p,
pipeline_spec,
[
yaml_provider.InlineProvider(
TEST_PROVIDERS, INPUT_TRANSFORM_TEST_PROVIDERS)
])
]
apache_beam/yaml/examples/testing/examples_test.py:371:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
apache_beam/yaml/yaml_transform.py:1408: in expand_pipeline
providers or {})).expand(root)
^^^^^^^^^^^^
apache_beam/yaml/yaml_transform.py:1369: in expand
result = expand_transform(
apache_beam/yaml/yaml_transform.py:482: in expand_transform
return expand_composite_transform(spec, scope)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/yaml/yaml_transform.py:832: in expand_composite_transform
return transform.expand(None)
^^^^^^^^^^^^^^^^^^^^^^
apache_beam/yaml/yaml_transform.py:815: in expand
inner_scope.compute_all()
apache_beam/yaml/yaml_transform.py:209: in compute_all
self.compute_outputs(transform_id)
apache_beam/yaml/yaml_transform.py:107: in wrapper
self._cache[key] = func(self, *args)
^^^^^^^^^^^^^^^^^
apache_beam/yaml/yaml_transform.py:245: in compute_outputs
return expand_transform(self._transforms_by_uuid[transform_id], self)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/yaml/yaml_transform.py:484: in expand_transform
return expand_leaf_transform(spec, scope)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
spec = {'__line__': 30, '__uuid__': '5eecdff8-725a-4998-ba8d-975cf5189f1f', 'config': {'__line__': 33, '__uuid__': 'fed16bb0-...rt(row):\n return roman.toRoman(row.year)\n'}}, ...}, 'input': {'input': 'c7cfe535-15a6-4a6d-a975-c6c31a6ecd6e'}, ...}
scope = <apache_beam.yaml.yaml_transform.Scope object at 0x7cc28ca4d8d0>
def expand_leaf_transform(spec, scope):
spec = spec.copy()
# Check for optional output_schema to verify on.
# The idea is to pass this output_schema config to the ValidateWithSchema
# transform.
output_schema_spec = {}
if 'output_schema' in spec.get('config', {}):
output_schema_spec = spec.get('config').pop('output_schema')
spec = normalize_inputs_outputs(spec)
inputs_dict = {
key: scope.get_pcollection(value)
for (key, value) in empty_if_explicitly_empty(spec['input']).items()
}
input_type = spec.get('input_type', 'default')
if input_type == 'list':
inputs = tuple(inputs_dict.values())
elif input_type == 'map':
inputs = inputs_dict
else:
if len(inputs_dict) == 0:
inputs = scope.root
elif len(inputs_dict) == 1:
inputs = next(iter(inputs_dict.values()))
else:
inputs = inputs_dict
_LOGGER.info("Expanding %s ", identify_object(spec))
ptransform = scope.create_ptransform(spec, inputs_dict.values())
try:
# TODO: Move validation to construction?
with FullyQualifiedNamedTransform.with_filter('*'):
outputs = inputs | scope.unique_name(spec, ptransform) >> ptransform
except Exception as exn:
> raise ValueError(
f"Error applying transform {identify_object(spec)}: {exn}") from exn
E ValueError: Error applying transform "ToRoman" at line 30: Traceback (most recent call last):
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/portability/expansion_service.py", line 101, in Expand
E result = pipeline.apply(
E ^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pipeline.py", line 746, in apply
E return self.apply(transform, pvalueish)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pipeline.py", line 826, in apply
E pvalueish_result = self.runner.apply(transform, pvalueish, self._options)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 191, in apply
E return self.apply_PTransform(transform, input, options)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 195, in apply_PTransform
E return transform.expand(input)
E ^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/transforms/fully_qualified_named_transform.py", line 78, in expand
E return pinput | transform
E ~~~~~~~^~~~~~~~~~~
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pvalue.py", line 139, in __or__
E return self.pipeline.apply(ptransform, self)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pipeline.py", line 826, in apply
E pvalueish_result = self.runner.apply(transform, pvalueish, self._options)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 191, in apply
E return self.apply_PTransform(transform, input, options)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 195, in apply_PTransform
E return transform.expand(input)
E ^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/transforms/ptransform.py", line 1032, in expand
E return self._fn(pcoll, *args, **kwargs)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_errors.py", line 75, in expand
E return transform_fn(wrapped_pcoll, **kwargs).as_result(
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 718, in _PyJsMapToFields
E **{
E ^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 719, in <dictcomp>
E name: _as_callable(
E ^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 396, in _as_callable
E func = _expand_python_mapping_func(original_fields, **expr)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 311, in _expand_python_mapping_func
E return python_callable.PythonCallableWithSource(source)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/utils/python_callable.py", line 48, in __init__
E self._callable = self.load_from_source(source)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/utils/python_callable.py", line 60, in load_from_source
E return cls.load_from_script(source)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/utils/python_callable.py", line 106, in load_from_script
E exec('\n'.join(lines), exec_globals)
E File "<string>", line 1, in <module>
E ModuleNotFoundError: No module named 'roman'
apache_beam/yaml/yaml_transform.py:521: ValueError
Check warning on line 0 in apache_beam.yaml.examples.testing.examples_test.MLTest
github-actions / Test Results
1 out of 2 runs failed: test_ml_preprocessing_yaml (apache_beam.yaml.examples.testing.examples_test.MLTest)
sdks/python/test-suites/tox/py311/build/srcs/sdks/python/pytest_py311_no_xdist.xml [took 10m 1s]
Raw output
Failed: Timeout (>600.0s) from pytest-timeout.
self = <apache_beam.yaml.examples.testing.examples_test.MLExamplesTest testMethod=test_ml_preprocessing_yaml>
@mock.patch('apache_beam.Pipeline', TestPipeline)
def test_yaml_example(self):
with open(pipeline_spec_file, encoding="utf-8") as f:
lines = f.readlines()
expected_key = '# Expected:\n'
if expected_key in lines:
expected = lines[lines.index('# Expected:\n') + 1:]
else:
raise ValueError(
f"Missing '# Expected:' tag in example file '{pipeline_spec_file}'")
for i, line in enumerate(expected):
expected[i] = line.replace('# ', '').replace('\n', '')
expected = [line for line in expected if line]
raw_spec_string = ''.join(lines)
# Filter for any jinja preprocessor - this has to be done before other
# preprocessors.
jinja_preprocessor = [
preprocessor for preprocessor in custom_preprocessors
if 'jinja_preprocessor' in preprocessor.__name__
]
if jinja_preprocessor:
jinja_preprocessor = jinja_preprocessor[0]
raw_spec_string = jinja_preprocessor(
raw_spec_string, self._testMethodName)
custom_preprocessors.remove(jinja_preprocessor)
pipeline_spec = yaml.load(
raw_spec_string, Loader=yaml_transform.SafeLineLoader)
with TestEnvironment() as env:
for fn in custom_preprocessors:
pipeline_spec = fn(pipeline_spec, expected, env)
with beam.Pipeline(options=PipelineOptions(
pickle_library='cloudpickle',
**yaml_transform.SafeLineLoader.strip_metadata(pipeline_spec.get(
'options', {})))) as p:
actual = [
> yaml_transform.expand_pipeline(
p,
pipeline_spec,
[
yaml_provider.InlineProvider(
TEST_PROVIDERS, INPUT_TRANSFORM_TEST_PROVIDERS)
])
]
apache_beam/yaml/examples/testing/examples_test.py:371:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
apache_beam/yaml/yaml_transform.py:1408: in expand_pipeline
providers or {})).expand(root)
^^^^^^^^^^^^
apache_beam/yaml/yaml_transform.py:1369: in expand
result = expand_transform(
apache_beam/yaml/yaml_transform.py:482: in expand_transform
return expand_composite_transform(spec, scope)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/yaml/yaml_transform.py:832: in expand_composite_transform
return transform.expand(None)
^^^^^^^^^^^^^^^^^^^^^^
apache_beam/yaml/yaml_transform.py:815: in expand
inner_scope.compute_all()
apache_beam/yaml/yaml_transform.py:209: in compute_all
self.compute_outputs(transform_id)
apache_beam/yaml/yaml_transform.py:107: in wrapper
self._cache[key] = func(self, *args)
^^^^^^^^^^^^^^^^^
apache_beam/yaml/yaml_transform.py:245: in compute_outputs
return expand_transform(self._transforms_by_uuid[transform_id], self)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/yaml/yaml_transform.py:484: in expand_transform
return expand_leaf_transform(spec, scope)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/yaml/yaml_transform.py:515: in expand_leaf_transform
ptransform = scope.create_ptransform(spec, inputs_dict.values())
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/yaml/yaml_transform.py:408: in create_ptransform
provider.create_transform(
apache_beam/yaml/yaml_provider.py:239: in create_transform
if urn in self.schema_transforms():
^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/yaml/yaml_provider.py:212: in schema_transforms
for config in external.SchemaAwareExternalTransform.discover(
target/.tox-py311/py311/lib/python3.11/site-packages/apache_beam/transforms/external.py:522: in discover
return list(cls.discover_iter(expansion_service, ignore_errors))
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
target/.tox-py311/py311/lib/python3.11/site-packages/apache_beam/transforms/external.py:526: in discover_iter
with ExternalTransform.service(expansion_service) as service:
/opt/hostedtoolcache/Python/3.11.12/x64/lib/python3.11/contextlib.py:137: in __enter__
return next(self.gen)
^^^^^^^^^^^^^^
target/.tox-py311/py311/lib/python3.11/site-packages/apache_beam/transforms/external.py:884: in service
with expansion_service as stub:
apache_beam/yaml/yaml_provider.py:1411: in __enter__
self._service = self._service_provider.__enter__()
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
target/.tox-py311/py311/lib/python3.11/site-packages/apache_beam/utils/subprocess_server.py:179: in __enter__
return self.start()
^^^^^^^^^^^^
target/.tox-py311/py311/lib/python3.11/site-packages/apache_beam/utils/subprocess_server.py:199: in start
channel_ready.result(timeout=wait_secs)
target/.tox-py311/py311/lib/python3.11/site-packages/grpc/_utilities.py:162: in result
self._block(timeout)
target/.tox-py311/py311/lib/python3.11/site-packages/grpc/_utilities.py:108: in _block
self._condition.wait(timeout=remaining)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <Condition(<unlocked _thread.RLock object owner=0 count=0 at 0x7cc28f790f00>, 0)>
timeout = 102.06745934486389
def wait(self, timeout=None):
"""Wait until notified or until a timeout occurs.
If the calling thread has not acquired the lock when this method is
called, a RuntimeError is raised.
This method releases the underlying lock, and then blocks until it is
awakened by a notify() or notify_all() call for the same condition
variable in another thread, or until the optional timeout occurs. Once
awakened or timed out, it re-acquires the lock and returns.
When the timeout argument is present and not None, it should be a
floating point number specifying a timeout for the operation in seconds
(or fractions thereof).
When the underlying lock is an RLock, it is not released using its
release() method, since this may not actually unlock the lock when it
was acquired multiple times recursively. Instead, an internal interface
of the RLock class is used, which really unlocks it even when it has
been recursively acquired several times. Another internal interface is
then used to restore the recursion level when the lock is reacquired.
"""
if not self._is_owned():
raise RuntimeError("cannot wait on un-acquired lock")
waiter = _allocate_lock()
waiter.acquire()
self._waiters.append(waiter)
saved_state = self._release_save()
gotit = False
try: # restore state no matter what (e.g., KeyboardInterrupt)
if timeout is None:
waiter.acquire()
gotit = True
else:
if timeout > 0:
> gotit = waiter.acquire(True, timeout)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E Failed: Timeout (>600.0s) from pytest-timeout.
/opt/hostedtoolcache/Python/3.11.12/x64/lib/python3.11/threading.py:331: Failed
Check warning on line 0 in apache_beam.yaml.examples.testing.examples_test.MLTest
github-actions / Test Results
1 out of 2 runs failed: test_streaming_sentiment_analysis_yaml (apache_beam.yaml.examples.testing.examples_test.MLTest)
sdks/python/test-suites/tox/py311/build/srcs/sdks/python/pytest_py311_no_xdist.xml [took 1m 4s]
Raw output
ValueError: Error applying transform "Truncating" at line 145: Traceback (most recent call last):
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/portability/expansion_service.py", line 101, in Expand
result = pipeline.apply(
^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pipeline.py", line 746, in apply
return self.apply(transform, pvalueish)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pipeline.py", line 826, in apply
pvalueish_result = self.runner.apply(transform, pvalueish, self._options)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 191, in apply
return self.apply_PTransform(transform, input, options)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 195, in apply_PTransform
return transform.expand(input)
^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/transforms/fully_qualified_named_transform.py", line 78, in expand
return pinput | transform
~~~~~~~^~~~~~~~~~~
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pvalue.py", line 139, in __or__
return self.pipeline.apply(ptransform, self)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pipeline.py", line 826, in apply
pvalueish_result = self.runner.apply(transform, pvalueish, self._options)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 191, in apply
return self.apply_PTransform(transform, input, options)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 195, in apply_PTransform
return transform.expand(input)
^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/transforms/ptransform.py", line 1032, in expand
return self._fn(pcoll, *args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_errors.py", line 75, in expand
return transform_fn(wrapped_pcoll, **kwargs).as_result(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 718, in _PyJsMapToFields
**{
^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 719, in <dictcomp>
name: _as_callable(
^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 396, in _as_callable
func = _expand_python_mapping_func(original_fields, **expr)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 311, in _expand_python_mapping_func
return python_callable.PythonCallableWithSource(source)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/utils/python_callable.py", line 48, in __init__
self._callable = self.load_from_source(source)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/utils/python_callable.py", line 60, in load_from_source
return cls.load_from_script(source)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/utils/python_callable.py", line 106, in load_from_script
exec('\n'.join(lines), exec_globals)
File "<string>", line 1, in <module>
ModuleNotFoundError: No module named 'transformers'
spec = {'__line__': 145, '__uuid__': '389dc950-f0fd-4237-9911-2fe0c1d62131', 'config': {'__line__': 149, '__uuid__': '228ab25...uncated_sentence\n'}, 'likes': {'expression': 'likes'}, ...}, ...}, 'input': {'input': 'FilterForProperComments'}, ...}
scope = <apache_beam.yaml.yaml_transform.Scope object at 0x7cc28c85f110>
def expand_leaf_transform(spec, scope):
spec = spec.copy()
# Check for optional output_schema to verify on.
# The idea is to pass this output_schema config to the ValidateWithSchema
# transform.
output_schema_spec = {}
if 'output_schema' in spec.get('config', {}):
output_schema_spec = spec.get('config').pop('output_schema')
spec = normalize_inputs_outputs(spec)
inputs_dict = {
key: scope.get_pcollection(value)
for (key, value) in empty_if_explicitly_empty(spec['input']).items()
}
input_type = spec.get('input_type', 'default')
if input_type == 'list':
inputs = tuple(inputs_dict.values())
elif input_type == 'map':
inputs = inputs_dict
else:
if len(inputs_dict) == 0:
inputs = scope.root
elif len(inputs_dict) == 1:
inputs = next(iter(inputs_dict.values()))
else:
inputs = inputs_dict
_LOGGER.info("Expanding %s ", identify_object(spec))
ptransform = scope.create_ptransform(spec, inputs_dict.values())
try:
# TODO: Move validation to construction?
with FullyQualifiedNamedTransform.with_filter('*'):
> outputs = inputs | scope.unique_name(spec, ptransform) >> ptransform
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/yaml/yaml_transform.py:519:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
target/.tox-py311/py311/lib/python3.11/site-packages/apache_beam/pvalue.py:139: in __or__
return self.pipeline.apply(ptransform, self)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
target/.tox-py311/py311/lib/python3.11/site-packages/apache_beam/pipeline.py:731: in apply
return self.apply(
target/.tox-py311/py311/lib/python3.11/site-packages/apache_beam/pipeline.py:746: in apply
return self.apply(transform, pvalueish)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
target/.tox-py311/py311/lib/python3.11/site-packages/apache_beam/pipeline.py:826: in apply
pvalueish_result = self.runner.apply(transform, pvalueish, self._options)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
target/.tox-py311/py311/lib/python3.11/site-packages/apache_beam/runners/runner.py:191: in apply
return self.apply_PTransform(transform, input, options)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
target/.tox-py311/py311/lib/python3.11/site-packages/apache_beam/runners/runner.py:195: in apply_PTransform
return transform.expand(input)
^^^^^^^^^^^^^^^^^^^^^^^
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <ExternalTransform(PTransform) label=[ExternalTransform(beam:transforms:python:fully_qualified_named)] at 0x7cc28c7c9810>
pvalueish = <PCollection[FilterForProperComments/Filter(filter).None] at 0x7cc28ad83750>
def expand(self, pvalueish: pvalue.PCollection) -> pvalue.PCollection:
if isinstance(pvalueish, pvalue.PBegin):
self._inputs = {}
elif isinstance(pvalueish, (list, tuple)):
self._inputs = {str(ix): pvalue for ix, pvalue in enumerate(pvalueish)}
elif isinstance(pvalueish, dict):
self._inputs = pvalueish
else:
self._inputs = {'input': pvalueish}
pipeline = (
next(iter(self._inputs.values())).pipeline
if self._inputs else pvalueish.pipeline)
context = pipeline_context.PipelineContext(
component_id_map=pipeline.component_id_map)
transform_proto = beam_runner_api_pb2.PTransform(
unique_name=pipeline._current_transform().full_label,
spec=beam_runner_api_pb2.FunctionSpec(
urn=self._urn, payload=self._payload))
for tag, pcoll in self._inputs.items():
transform_proto.inputs[tag] = context.pcollections.get_id(pcoll)
# Conversion to/from proto assumes producers.
# TODO: Possibly loosen this.
context.transforms.put_proto(
'%s_%s' % (self._IMPULSE_PREFIX, tag),
beam_runner_api_pb2.PTransform(
unique_name='%s_%s' % (self._IMPULSE_PREFIX, tag),
spec=beam_runner_api_pb2.FunctionSpec(
urn=common_urns.primitives.IMPULSE.urn),
outputs={'out': transform_proto.inputs[tag]}))
output_coders = None
if self._type_hints.output_types:
if self._type_hints.output_types[0]:
output_coders = dict(
(str(k), context.coder_id_from_element_type(v))
for (k, v) in enumerate(self._type_hints.output_types[0]))
elif self._type_hints.output_types[1]:
output_coders = {
k: context.coder_id_from_element_type(v)
for (k, v) in self._type_hints.output_types[1].items()
}
components = context.to_runner_api()
request = beam_expansion_api_pb2.ExpansionRequest(
components=components,
namespace=self._external_namespace, # type: ignore[arg-type]
transform=transform_proto,
output_coder_requests=output_coders,
pipeline_options=pipeline._options.to_runner_api())
expansion_service = _maybe_use_transform_service(
self._expansion_service, pipeline.options)
with ExternalTransform.service(expansion_service) as service:
response = service.Expand(request)
if response.error:
> raise RuntimeError(_sanitize_java_traceback(response.error))
E RuntimeError: Traceback (most recent call last):
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/portability/expansion_service.py", line 101, in Expand
E result = pipeline.apply(
E ^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pipeline.py", line 746, in apply
E return self.apply(transform, pvalueish)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pipeline.py", line 826, in apply
E pvalueish_result = self.runner.apply(transform, pvalueish, self._options)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 191, in apply
E return self.apply_PTransform(transform, input, options)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 195, in apply_PTransform
E return transform.expand(input)
E ^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/transforms/fully_qualified_named_transform.py", line 78, in expand
E return pinput | transform
E ~~~~~~~^~~~~~~~~~~
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pvalue.py", line 139, in __or__
E return self.pipeline.apply(ptransform, self)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pipeline.py", line 826, in apply
E pvalueish_result = self.runner.apply(transform, pvalueish, self._options)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 191, in apply
E return self.apply_PTransform(transform, input, options)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 195, in apply_PTransform
E return transform.expand(input)
E ^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/transforms/ptransform.py", line 1032, in expand
E return self._fn(pcoll, *args, **kwargs)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_errors.py", line 75, in expand
E return transform_fn(wrapped_pcoll, **kwargs).as_result(
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 718, in _PyJsMapToFields
E **{
E ^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 719, in <dictcomp>
E name: _as_callable(
E ^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 396, in _as_callable
E func = _expand_python_mapping_func(original_fields, **expr)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 311, in _expand_python_mapping_func
E return python_callable.PythonCallableWithSource(source)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/utils/python_callable.py", line 48, in __init__
E self._callable = self.load_from_source(source)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/utils/python_callable.py", line 60, in load_from_source
E return cls.load_from_script(source)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/utils/python_callable.py", line 106, in load_from_script
E exec('\n'.join(lines), exec_globals)
E File "<string>", line 1, in <module>
E ModuleNotFoundError: No module named 'transformers'
target/.tox-py311/py311/lib/python3.11/site-packages/apache_beam/transforms/external.py:832: RuntimeError
The above exception was the direct cause of the following exception:
self = <apache_beam.yaml.examples.testing.examples_test.MLExamplesTest testMethod=test_streaming_sentiment_analysis_yaml>
@mock.patch('apache_beam.Pipeline', TestPipeline)
def test_yaml_example(self):
with open(pipeline_spec_file, encoding="utf-8") as f:
lines = f.readlines()
expected_key = '# Expected:\n'
if expected_key in lines:
expected = lines[lines.index('# Expected:\n') + 1:]
else:
raise ValueError(
f"Missing '# Expected:' tag in example file '{pipeline_spec_file}'")
for i, line in enumerate(expected):
expected[i] = line.replace('# ', '').replace('\n', '')
expected = [line for line in expected if line]
raw_spec_string = ''.join(lines)
# Filter for any jinja preprocessor - this has to be done before other
# preprocessors.
jinja_preprocessor = [
preprocessor for preprocessor in custom_preprocessors
if 'jinja_preprocessor' in preprocessor.__name__
]
if jinja_preprocessor:
jinja_preprocessor = jinja_preprocessor[0]
raw_spec_string = jinja_preprocessor(
raw_spec_string, self._testMethodName)
custom_preprocessors.remove(jinja_preprocessor)
pipeline_spec = yaml.load(
raw_spec_string, Loader=yaml_transform.SafeLineLoader)
with TestEnvironment() as env:
for fn in custom_preprocessors:
pipeline_spec = fn(pipeline_spec, expected, env)
with beam.Pipeline(options=PipelineOptions(
pickle_library='cloudpickle',
**yaml_transform.SafeLineLoader.strip_metadata(pipeline_spec.get(
'options', {})))) as p:
actual = [
> yaml_transform.expand_pipeline(
p,
pipeline_spec,
[
yaml_provider.InlineProvider(
TEST_PROVIDERS, INPUT_TRANSFORM_TEST_PROVIDERS)
])
]
apache_beam/yaml/examples/testing/examples_test.py:371:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
apache_beam/yaml/yaml_transform.py:1408: in expand_pipeline
providers or {})).expand(root)
^^^^^^^^^^^^
apache_beam/yaml/yaml_transform.py:1369: in expand
result = expand_transform(
apache_beam/yaml/yaml_transform.py:482: in expand_transform
return expand_composite_transform(spec, scope)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/yaml/yaml_transform.py:832: in expand_composite_transform
return transform.expand(None)
^^^^^^^^^^^^^^^^^^^^^^
apache_beam/yaml/yaml_transform.py:815: in expand
inner_scope.compute_all()
apache_beam/yaml/yaml_transform.py:209: in compute_all
self.compute_outputs(transform_id)
apache_beam/yaml/yaml_transform.py:107: in wrapper
self._cache[key] = func(self, *args)
^^^^^^^^^^^^^^^^^
apache_beam/yaml/yaml_transform.py:245: in compute_outputs
return expand_transform(self._transforms_by_uuid[transform_id], self)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/yaml/yaml_transform.py:484: in expand_transform
return expand_leaf_transform(spec, scope)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
spec = {'__line__': 145, '__uuid__': '389dc950-f0fd-4237-9911-2fe0c1d62131', 'config': {'__line__': 149, '__uuid__': '228ab25...uncated_sentence\n'}, 'likes': {'expression': 'likes'}, ...}, ...}, 'input': {'input': 'FilterForProperComments'}, ...}
scope = <apache_beam.yaml.yaml_transform.Scope object at 0x7cc28c85f110>
def expand_leaf_transform(spec, scope):
spec = spec.copy()
# Check for optional output_schema to verify on.
# The idea is to pass this output_schema config to the ValidateWithSchema
# transform.
output_schema_spec = {}
if 'output_schema' in spec.get('config', {}):
output_schema_spec = spec.get('config').pop('output_schema')
spec = normalize_inputs_outputs(spec)
inputs_dict = {
key: scope.get_pcollection(value)
for (key, value) in empty_if_explicitly_empty(spec['input']).items()
}
input_type = spec.get('input_type', 'default')
if input_type == 'list':
inputs = tuple(inputs_dict.values())
elif input_type == 'map':
inputs = inputs_dict
else:
if len(inputs_dict) == 0:
inputs = scope.root
elif len(inputs_dict) == 1:
inputs = next(iter(inputs_dict.values()))
else:
inputs = inputs_dict
_LOGGER.info("Expanding %s ", identify_object(spec))
ptransform = scope.create_ptransform(spec, inputs_dict.values())
try:
# TODO: Move validation to construction?
with FullyQualifiedNamedTransform.with_filter('*'):
outputs = inputs | scope.unique_name(spec, ptransform) >> ptransform
except Exception as exn:
> raise ValueError(
f"Error applying transform {identify_object(spec)}: {exn}") from exn
E ValueError: Error applying transform "Truncating" at line 145: Traceback (most recent call last):
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/portability/expansion_service.py", line 101, in Expand
E result = pipeline.apply(
E ^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pipeline.py", line 746, in apply
E return self.apply(transform, pvalueish)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pipeline.py", line 826, in apply
E pvalueish_result = self.runner.apply(transform, pvalueish, self._options)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 191, in apply
E return self.apply_PTransform(transform, input, options)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 195, in apply_PTransform
E return transform.expand(input)
E ^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/transforms/fully_qualified_named_transform.py", line 78, in expand
E return pinput | transform
E ~~~~~~~^~~~~~~~~~~
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pvalue.py", line 139, in __or__
E return self.pipeline.apply(ptransform, self)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/pipeline.py", line 826, in apply
E pvalueish_result = self.runner.apply(transform, pvalueish, self._options)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 191, in apply
E return self.apply_PTransform(transform, input, options)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/runners/runner.py", line 195, in apply_PTransform
E return transform.expand(input)
E ^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/transforms/ptransform.py", line 1032, in expand
E return self._fn(pcoll, *args, **kwargs)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_errors.py", line 75, in expand
E return transform_fn(wrapped_pcoll, **kwargs).as_result(
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 718, in _PyJsMapToFields
E **{
E ^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 719, in <dictcomp>
E name: _as_callable(
E ^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 396, in _as_callable
E func = _expand_python_mapping_func(original_fields, **expr)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/yaml/yaml_mapping.py", line 311, in _expand_python_mapping_func
E return python_callable.PythonCallableWithSource(source)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/utils/python_callable.py", line 48, in __init__
E self._callable = self.load_from_source(source)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/utils/python_callable.py", line 60, in load_from_source
E return cls.load_from_script(source)
E ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py311/build/srcs/sdks/python/apache_beam/utils/python_callable.py", line 106, in load_from_script
E exec('\n'.join(lines), exec_globals)
E File "<string>", line 1, in <module>
E ModuleNotFoundError: No module named 'transformers'
apache_beam/yaml/yaml_transform.py:521: ValueError
Loading