beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From al...@apache.org
Subject [beam] branch master updated: Makes Dataflow runner use python3-fnapi container when applicable.
Date Wed, 13 Feb 2019 18:42:37 GMT
This is an automated email from the ASF dual-hosted git repository.

altay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new 3ecd61c  Makes Dataflow runner use python3-fnapi container when applicable.
     new b8183e4  Merge pull request #7829 from tvalentyn/use_py3_container_for_fnapi
3ecd61c is described below

commit 3ecd61c974ad716189b0a728572a7e87abae031c
Author: Valentyn Tymofieiev <valentyn@google.com>
AuthorDate: Tue Feb 12 21:28:33 2019 -0800

    Makes Dataflow runner use python3-fnapi container when applicable.
---
 .../runners/dataflow/internal/apiclient.py         | 23 +++++++++++-----
 .../runners/dataflow/internal/apiclient_test.py    | 32 +++++++++++++++-------
 2 files changed, 38 insertions(+), 17 deletions(-)

diff --git a/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py b/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py
index a1873ea..6d5e140 100644
--- a/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py
+++ b/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py
@@ -874,16 +874,25 @@ def get_default_container_image_for_current_sdk(job_type):
     Returns:
       str: Google Cloud Dataflow container image for remote execution.
     """
+  if sys.version_info[0] == 2:
+    version_suffix = ''
+  elif sys.version_info[0] == 3:
+    version_suffix = '3'
+  else:
+    raise Exception('Dataflow only supports Python versions 2 and 3, got: %s'
+                    % sys.version_info[0])
+
   # TODO(tvalentyn): Use enumerated type instead of strings for job types.
   if job_type == 'FNAPI_BATCH' or job_type == 'FNAPI_STREAMING':
-    image_name = names.DATAFLOW_CONTAINER_IMAGE_REPOSITORY + '/python-fnapi'
+    fnapi_suffix = '-fnapi'
   else:
-    if sys.version_info[0] == 2:
-      image_name = names.DATAFLOW_CONTAINER_IMAGE_REPOSITORY + '/python'
-    elif sys.version_info[0] == 3:
-      image_name = names.DATAFLOW_CONTAINER_IMAGE_REPOSITORY + '/python3'
-    else:
-      raise Exception('Dataflow only supports Python versions 2 and 3.')
+    fnapi_suffix = ''
+
+  image_name = '{repository}/python{version_suffix}{fnapi_suffix}'.format(
+      repository=names.DATAFLOW_CONTAINER_IMAGE_REPOSITORY,
+      version_suffix=version_suffix,
+      fnapi_suffix=fnapi_suffix)
+
   image_tag = _get_required_container_version(job_type)
   return image_name + ':' + image_tag
 
diff --git a/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py b/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py
index ac66fc6..0e37a0a 100644
--- a/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py
+++ b/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py
@@ -248,10 +248,16 @@ class UtilTest(unittest.TestCase):
                                 pipeline_options,
                                 '2.0.0', #any environment version
                                 FAKE_PIPELINE_URL)
-    self.assertEqual(
-        env.proto.workerPools[0].workerHarnessContainerImage,
-        (names.DATAFLOW_CONTAINER_IMAGE_REPOSITORY +
-         '/python-fnapi:' + names.BEAM_FNAPI_CONTAINER_VERSION))
+    if sys.version_info[0] == 3:
+      self.assertEqual(
+          env.proto.workerPools[0].workerHarnessContainerImage,
+          (names.DATAFLOW_CONTAINER_IMAGE_REPOSITORY +
+           '/python3-fnapi:' + names.BEAM_FNAPI_CONTAINER_VERSION))
+    else:
+      self.assertEqual(
+          env.proto.workerPools[0].workerHarnessContainerImage,
+          (names.DATAFLOW_CONTAINER_IMAGE_REPOSITORY +
+           '/python-fnapi:' + names.BEAM_FNAPI_CONTAINER_VERSION))
 
     # batch, legacy pipeline.
     pipeline_options = PipelineOptions(
@@ -281,10 +287,16 @@ class UtilTest(unittest.TestCase):
                                 pipeline_options,
                                 '2.0.0', #any environment version
                                 FAKE_PIPELINE_URL)
-    self.assertEqual(
-        env.proto.workerPools[0].workerHarnessContainerImage,
-        (names.DATAFLOW_CONTAINER_IMAGE_REPOSITORY +
-         '/python-fnapi:2.2.0'))
+    if sys.version_info[0] == 3:
+      self.assertEqual(
+          env.proto.workerPools[0].workerHarnessContainerImage,
+          (names.DATAFLOW_CONTAINER_IMAGE_REPOSITORY +
+           '/python3-fnapi:2.2.0'))
+    else:
+      self.assertEqual(
+          env.proto.workerPools[0].workerHarnessContainerImage,
+          (names.DATAFLOW_CONTAINER_IMAGE_REPOSITORY +
+           '/python-fnapi:2.2.0'))
 
     # batch, legacy pipeline.
     pipeline_options = PipelineOptions(
@@ -391,10 +403,10 @@ class UtilTest(unittest.TestCase):
          '--temp_location', 'gs://test-location/temp',
          '--experiments', 'beam_fn_api',
          '--experiments', 'use_multiple_sdk_containers'])
+    mock_sys.version_info = [2, 333]
     environment = apiclient.Environment(
         [], pipeline_options, 1, FAKE_PIPELINE_URL)
-    mock_sys.version_info = [22, 333]
-    self.assertEqual('Apache Beam Python 22.333 SDK',
+    self.assertEqual('Apache Beam Python 2.333 SDK',
                      environment._get_python_sdk_name())
 
 


Mime
View raw message