[
https://issues.apache.org/jira/browse/BEAM-3490?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16336611#comment-16336611
]
ASF GitHub Bot commented on BEAM-3490:
--------------------------------------
robertwb closed pull request #4454: [BEAM-3490] Explicitly use direct runner in
non-portable tests.
URL: https://github.com/apache/beam/pull/4454
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/sdks/python/apache_beam/pipeline_test.py
b/sdks/python/apache_beam/pipeline_test.py
index 22f89721ef5..634483e4160 100644
--- a/sdks/python/apache_beam/pipeline_test.py
+++ b/sdks/python/apache_beam/pipeline_test.py
@@ -160,8 +160,9 @@ def test_create_singleton_pcollection(self):
# TODO(BEAM-1555): Test is failing on the service, with FakeSource.
# @attr('ValidatesRunner')
- def test_metrics_in_source(self):
- pipeline = TestPipeline()
+ def test_metrics_in_fake_source(self):
+ # FakeSource mock requires DirectRunner.
+ pipeline = TestPipeline(runner='DirectRunner')
pcoll = pipeline | Read(FakeSource([1, 2, 3, 4, 5, 6]))
assert_that(pcoll, equal_to([1, 2, 3, 4, 5, 6]))
res = pipeline.run()
@@ -171,8 +172,9 @@ def test_metrics_in_source(self):
self.assertEqual(outputs_counter.key.metric.name, 'outputs')
self.assertEqual(outputs_counter.committed, 6)
- def test_read(self):
- pipeline = TestPipeline()
+ def test_fake_read(self):
+ # FakeSource mock requires DirectRunner.
+ pipeline = TestPipeline(runner='DirectRunner')
pcoll = pipeline | 'read' >> Read(FakeSource([1, 2, 3]))
assert_that(pcoll, equal_to([1, 2, 3]))
pipeline.run()
@@ -326,7 +328,8 @@ def get_overrides():
file_system_override_mock.side_effect = get_overrides
- with Pipeline() as p:
+ # Specify DirectRunner as it's the one patched above.
+ with Pipeline(runner='DirectRunner') as p:
pcoll = p | beam.Create([1, 2, 3]) | 'Multiply' >> DoubleParDo()
assert_that(pcoll, equal_to([3, 6, 9]))
@@ -512,7 +515,7 @@ def __reduce__(self):
class DirectRunnerRetryTests(unittest.TestCase):
def test_retry_fork_graph(self):
- p = beam.Pipeline()
+ p = beam.Pipeline(runner='DirectRunner')
# TODO(mariagh): Remove the use of globals from the test.
global count_b, count_c # pylint: disable=global-variable-undefined
diff --git a/sdks/python/apache_beam/runners/dataflow/native_io/iobase_test.py
b/sdks/python/apache_beam/runners/dataflow/native_io/iobase_test.py
index 01fd35f9cf9..ce772572874 100644
--- a/sdks/python/apache_beam/runners/dataflow/native_io/iobase_test.py
+++ b/sdks/python/apache_beam/runners/dataflow/native_io/iobase_test.py
@@ -182,7 +182,8 @@ def __exit__(self, *unused_args):
def Write(self, value):
self.written_values.append(value)
- p = TestPipeline()
+ # Records in-memory writes, only works on Direct runner.
+ p = TestPipeline(runner='DirectRunner')
sink = FakeSink()
p | Create(['a', 'b', 'c']) | _NativeWrite(sink) # pylint:
disable=expression-not-assigned
p.run()
diff --git a/sdks/python/apache_beam/runners/direct/direct_runner_test.py
b/sdks/python/apache_beam/runners/direct/direct_runner_test.py
index 1c8b7855aad..1b51d05aae0 100644
--- a/sdks/python/apache_beam/runners/direct/direct_runner_test.py
+++ b/sdks/python/apache_beam/runners/direct/direct_runner_test.py
@@ -27,7 +27,7 @@ class DirectPipelineResultTest(unittest.TestCase):
def test_waiting_on_result_stops_executor_threads(self):
pre_test_threads = set(t.ident for t in threading.enumerate())
- pipeline = test_pipeline.TestPipeline()
+ pipeline = test_pipeline.TestPipeline(runner='DirectRunner')
_ = (pipeline | beam.Create([{'foo': 'bar'}]))
result = pipeline.run()
result.wait_until_finish()
diff --git a/sdks/python/apache_beam/runners/direct/sdf_direct_runner_test.py
b/sdks/python/apache_beam/runners/direct/sdf_direct_runner_test.py
index 7ab6dde9397..c1df7da52c4 100644
--- a/sdks/python/apache_beam/runners/direct/sdf_direct_runner_test.py
+++ b/sdks/python/apache_beam/runners/direct/sdf_direct_runner_test.py
@@ -145,7 +145,7 @@ def run_sdf_read_pipeline(
assert len(expected_data) > 0
- with TestPipeline() as p:
+ with TestPipeline(runner='DirectRunner') as p:
pc1 = (p
| 'Create1' >> beam.Create(file_names)
| 'SDF' >> beam.ParDo(ReadFiles(resume_count)))
@@ -205,7 +205,7 @@ def test_sdf_with_resume_multiple_elements(self):
resume_count)
def test_sdf_with_windowed_timestamped_input(self):
- with TestPipeline() as p:
+ with TestPipeline(runner='DirectRunner') as p:
result = (p
| beam.Create([1, 3, 5, 10])
| beam.FlatMap(lambda t: [TimestampedValue(('A', t), t),
@@ -221,7 +221,7 @@ def test_sdf_with_windowed_timestamped_input(self):
assert_that(result, equal_to(expected_result))
def test_sdf_with_side_inputs(self):
- with TestPipeline() as p:
+ with TestPipeline(runner='DirectRunner') as p:
result = (p
| 'create_main' >> beam.Create(['1', '3', '5'])
| beam.ParDo(ExpandStrings(), side=['1', '3']))
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
> Reasonable Python direct runner batch performance.
> --------------------------------------------------
>
> Key: BEAM-3490
> URL: https://issues.apache.org/jira/browse/BEAM-3490
> Project: Beam
> Issue Type: Bug
> Components: sdk-py-core
> Reporter: Robert Bradshaw
> Assignee: Robert Bradshaw
> Priority: Major
>
> The plan is to migrate to the FnApi Runner for batch workloads.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
