[
https://issues.apache.org/jira/browse/BEAM-3883?focusedWorklogId=117109&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-117109
]
ASF GitHub Bot logged work on BEAM-3883:
----------------------------------------
Author: ASF GitHub Bot
Created on: 28/Jun/18 22:58
Start Date: 28/Jun/18 22:58
Worklog Time Spent: 10m
Work Description: robertwb closed pull request #5817: [BEAM-3883] Stage
files in the portability runner.
URL: https://github.com/apache/beam/pull/5817
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/runners/dataflow/internal/apiclient.py
b/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py
index 784166cafda..ccce9a99ef7 100644
--- a/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py
+++ b/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py
@@ -442,10 +442,11 @@ def _stage_resources(self, options):
raise RuntimeError('The --temp_location option must be specified.')
resource_stager = _LegacyDataflowStager(self)
- return resource_stager.stage_job_resources(
+ _, resources = resource_stager.stage_job_resources(
options,
temp_dir=tempfile.mkdtemp(),
staging_location=google_cloud_options.staging_location)
+ return resources
def stage_file(self, gcs_or_local_path, file_name, stream,
mime_type='application/octet-stream'):
diff --git a/sdks/python/apache_beam/runners/portability/portable_runner.py
b/sdks/python/apache_beam/runners/portability/portable_runner.py
index 0da33fa7f4a..26376c93337 100644
--- a/sdks/python/apache_beam/runners/portability/portable_runner.py
+++ b/sdks/python/apache_beam/runners/portability/portable_runner.py
@@ -24,12 +24,11 @@
from apache_beam import metrics
from apache_beam.options.pipeline_options import PortableOptions
from apache_beam.portability import common_urns
-from apache_beam.portability.api import beam_artifact_api_pb2
-from apache_beam.portability.api import beam_artifact_api_pb2_grpc
from apache_beam.portability.api import beam_job_api_pb2
from apache_beam.portability.api import beam_job_api_pb2_grpc
from apache_beam.runners import pipeline_context
from apache_beam.runners import runner
+from apache_beam.runners.portability import portable_stager
__all__ = ['PortableRunner']
@@ -92,16 +91,12 @@ def run_pipeline(self, pipeline):
beam_job_api_pb2.PrepareJobRequest(
job_name='job', pipeline=proto_pipeline))
if prepare_response.artifact_staging_endpoint.url:
- # Must commit something to get a retrieval token,
- # committing empty manifest for now.
- # TODO(BEAM-3883): Actually stage required files.
- artifact_service = beam_artifact_api_pb2_grpc.ArtifactStagingServiceStub(
-
grpc.insecure_channel(prepare_response.artifact_staging_endpoint.url))
- commit_manifest = artifact_service.CommitManifest(
- beam_artifact_api_pb2.CommitManifestRequest(
- manifest=beam_artifact_api_pb2.Manifest(),
- staging_session_token=prepare_response.staging_session_token))
- retrieval_token = commit_manifest.retrieval_token
+ stager = portable_stager.PortableStager(
+
grpc.insecure_channel(prepare_response.artifact_staging_endpoint.url),
+ prepare_response.staging_session_token)
+ retrieval_token, _ = stager.stage_job_resources(
+ pipeline._options,
+ staging_location='')
else:
retrieval_token = None
run_response = job_service.Run(
diff --git a/sdks/python/apache_beam/runners/portability/portable_stager.py
b/sdks/python/apache_beam/runners/portability/portable_stager.py
index f556811425b..3761373fb42 100644
--- a/sdks/python/apache_beam/runners/portability/portable_stager.py
+++ b/sdks/python/apache_beam/runners/portability/portable_stager.py
@@ -20,6 +20,8 @@
from __future__ import division
from __future__ import print_function
+import base64
+import hashlib
import os
from apache_beam.portability.api import beam_artifact_api_pb2
@@ -69,7 +71,8 @@ def stage_artifact(self, local_path_to_artifact,
artifact_name):
def artifact_request_generator():
artifact_metadata = beam_artifact_api_pb2.ArtifactMetadata(
- name=artifact_name)
+ name=artifact_name,
+ md5=_get_file_hash(local_path_to_artifact))
metadata = beam_artifact_api_pb2.PutArtifactMetadata(
staging_session_token=self._staging_session_token,
metadata=artifact_metadata)
@@ -90,7 +93,18 @@ def artifact_request_generator():
def commit_manifest(self):
manifest = beam_artifact_api_pb2.Manifest(artifact=self._artifacts)
self._artifacts = []
- self._artifact_staging_stub.CommitManifest(
+ return self._artifact_staging_stub.CommitManifest(
beam_artifact_api_pb2.CommitManifestRequest(
manifest=manifest,
- staging_session_token=self._staging_session_token))
+ staging_session_token=self._staging_session_token)).retrieval_token
+
+
+def _get_file_hash(path):
+ hasher = hashlib.md5()
+ with open(path) as f:
+ while True:
+ chunk = f.read(1 << 21)
+ if chunk:
+ hasher.update(chunk)
+ else:
+ return base64.b64encode(hasher.digest())
diff --git a/sdks/python/apache_beam/runners/portability/stager.py
b/sdks/python/apache_beam/runners/portability/stager.py
index a6bfa497bd5..e938970404a 100644
--- a/sdks/python/apache_beam/runners/portability/stager.py
+++ b/sdks/python/apache_beam/runners/portability/stager.py
@@ -219,6 +219,10 @@ def stage_job_resources(self,
resources.extend(
self._stage_beam_sdk(sdk_remote_location, staging_location,
temp_dir))
+ elif setup_options.sdk_location == 'container':
+ # Use the SDK that's built into the container, rather than re-staging
+ # it.
+ pass
else:
# This branch is also used by internal tests running with the SDK built
# at head.
@@ -252,8 +256,8 @@ def stage_job_resources(self,
# Delete all temp files created while staging job resources.
shutil.rmtree(temp_dir)
- self.commit_manifest()
- return resources
+ retrieval_token = self.commit_manifest()
+ return retrieval_token, resources
@staticmethod
def _download_file(from_url, to_path):
diff --git a/sdks/python/apache_beam/runners/portability/stager_test.py
b/sdks/python/apache_beam/runners/portability/stager_test.py
index 56b57d16796..f6ee55b5249 100644
--- a/sdks/python/apache_beam/runners/portability/stager_test.py
+++ b/sdks/python/apache_beam/runners/portability/stager_test.py
@@ -133,7 +133,7 @@ def test_no_main_session(self):
self.assertEqual([],
self.stager.stage_job_resources(
- options, staging_location=staging_dir))
+ options, staging_location=staging_dir)[1])
def test_with_main_session(self):
staging_dir = self.make_temp_dir()
@@ -144,7 +144,7 @@ def test_with_main_session(self):
self.assertEqual([names.PICKLED_MAIN_SESSION_FILE],
self.stager.stage_job_resources(
- options, staging_location=staging_dir))
+ options, staging_location=staging_dir)[1])
self.assertTrue(
os.path.isfile(
os.path.join(staging_dir, names.PICKLED_MAIN_SESSION_FILE)))
@@ -156,7 +156,7 @@ def test_default_resources(self):
self.assertEqual([],
self.stager.stage_job_resources(
- options, staging_location=staging_dir))
+ options, staging_location=staging_dir)[1])
def test_with_requirements_file(self):
staging_dir = self.make_temp_dir()
@@ -176,7 +176,7 @@ def test_with_requirements_file(self):
self.stager.stage_job_resources(
options,
populate_requirements_cache=self.populate_requirements_cache,
- staging_location=staging_dir)))
+ staging_location=staging_dir)[1]))
self.assertTrue(
os.path.isfile(os.path.join(staging_dir, stager.REQUIREMENTS_FILE)))
self.assertTrue(os.path.isfile(os.path.join(staging_dir, 'abc.txt')))
@@ -214,7 +214,7 @@ def test_with_requirements_file_and_cache(self):
self.stager.stage_job_resources(
options,
populate_requirements_cache=self.populate_requirements_cache,
- staging_location=staging_dir)))
+ staging_location=staging_dir)[1]))
self.assertTrue(
os.path.isfile(os.path.join(staging_dir, stager.REQUIREMENTS_FILE)))
self.assertTrue(os.path.isfile(os.path.join(staging_dir, 'abc.txt')))
@@ -244,7 +244,7 @@ def test_with_setup_file(self):
os.path.join(source_dir, stager.WORKFLOW_TARBALL_FILE)
],
temp_dir=source_dir,
- staging_location=staging_dir))
+ staging_location=staging_dir)[1])
self.assertTrue(
os.path.isfile(os.path.join(staging_dir,
stager.WORKFLOW_TARBALL_FILE)))
@@ -287,7 +287,7 @@ def test_sdk_location_default(self):
with mock.patch(
'apache_beam.utils.processes.check_call',
self.build_fake_pip_download_command_handler(has_wheels=False)):
- staged_resources = self.stager.stage_job_resources(
+ _, staged_resources = self.stager.stage_job_resources(
options, temp_dir=self.make_temp_dir(), staging_location=staging_dir)
self.assertEqual([names.DATAFLOW_SDK_TARBALL_FILE], staged_resources)
@@ -305,7 +305,7 @@ def test_sdk_location_default_with_wheels(self):
with mock.patch(
'apache_beam.utils.processes.check_call',
self.build_fake_pip_download_command_handler(has_wheels=True)):
- staged_resources = self.stager.stage_job_resources(
+ _, staged_resources = self.stager.stage_job_resources(
options, temp_dir=self.make_temp_dir(), staging_location=staging_dir)
self.assertEqual(len(staged_resources), 2)
@@ -329,7 +329,7 @@ def test_sdk_location_local_directory(self):
self.assertEqual([names.DATAFLOW_SDK_TARBALL_FILE],
self.stager.stage_job_resources(
- options, staging_location=staging_dir))
+ options, staging_location=staging_dir)[1])
tarball_path = os.path.join(staging_dir, names.DATAFLOW_SDK_TARBALL_FILE)
with open(tarball_path) as f:
self.assertEqual(f.read(), 'Package content.')
@@ -347,7 +347,7 @@ def test_sdk_location_local_source_file(self):
self.assertEqual([names.DATAFLOW_SDK_TARBALL_FILE],
self.stager.stage_job_resources(
- options, staging_location=staging_dir))
+ options, staging_location=staging_dir)[1])
tarball_path = os.path.join(staging_dir, names.DATAFLOW_SDK_TARBALL_FILE)
with open(tarball_path) as f:
self.assertEqual(f.read(), 'Package content.')
@@ -365,7 +365,7 @@ def test_sdk_location_local_wheel_file(self):
self.assertEqual([sdk_filename],
self.stager.stage_job_resources(
- options, staging_location=staging_dir))
+ options, staging_location=staging_dir)[1])
tarball_path = os.path.join(staging_dir, sdk_filename)
with open(tarball_path) as f:
self.assertEqual(f.read(), 'Package content.')
@@ -399,7 +399,7 @@ def test_sdk_location_remote_source_file(self,
*unused_mocks):
self.assertEqual([names.DATAFLOW_SDK_TARBALL_FILE],
self.stager.stage_job_resources(
- options, staging_location=staging_dir))
+ options, staging_location=staging_dir)[1])
@mock.patch(
'apache_beam.runners.portability.stager_test.TestStager.stage_artifact')
@@ -425,7 +425,7 @@ def is_remote_path(path):
'.stager.Stager._is_remote_path', staticmethod(is_remote_path)):
self.assertEqual([sdk_filename],
self.stager.stage_job_resources(
- options, staging_location=staging_dir))
+ options, staging_location=staging_dir)[1])
def test_sdk_location_http(self):
staging_dir = self.make_temp_dir()
@@ -445,7 +445,7 @@ def file_download(_, to_path):
'.stager.Stager._download_file', staticmethod(file_download)):
self.assertEqual([names.DATAFLOW_SDK_TARBALL_FILE],
self.stager.stage_job_resources(
- options, staging_location=staging_dir))
+ options, staging_location=staging_dir)[1])
tarball_path = os.path.join(staging_dir, names.DATAFLOW_SDK_TARBALL_FILE)
with open(tarball_path) as f:
@@ -500,7 +500,7 @@ def file_copy(from_path, to_path):
'abc.tar.gz', 'xyz.tar.gz', 'xyz2.tar', 'whl.whl',
'remote_file.tar.gz', stager.EXTRA_PACKAGES_FILE
], self.stager.stage_job_resources(
- options, staging_location=staging_dir))
+ options, staging_location=staging_dir)[1])
with open(os.path.join(staging_dir, stager.EXTRA_PACKAGES_FILE)) as f:
self.assertEqual([
'abc.tar.gz\n', 'xyz.tar.gz\n', 'xyz2.tar\n', 'whl.whl\n',
----------------------------------------------------------------
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:
us...@infra.apache.org
Issue Time Tracking
-------------------
Worklog Id: (was: 117109)
Time Spent: 19h 40m (was: 19.5h)
> Python SDK stages artifacts when talking to job server
> ------------------------------------------------------
>
> Key: BEAM-3883
> URL: https://issues.apache.org/jira/browse/BEAM-3883
> Project: Beam
> Issue Type: Sub-task
> Components: sdk-py-core
> Reporter: Ben Sidhom
> Assignee: Ankur Goenka
> Priority: Major
> Fix For: 2.6.0
>
> Time Spent: 19h 40m
> Remaining Estimate: 0h
>
> The Python SDK does not currently stage its user-defined functions or
> dependencies when talking to the job API. Artifacts that need to be staged
> include the user code itself, any SDK components not included in the
> container image, and the list of Python packages that must be installed at
> runtime.
>
> Artifacts that are currently expected can be found in the harness boot code:
> [https://github.com/apache/beam/blob/58e3b06bee7378d2d8db1c8dd534b415864f63e1/sdks/python/container/boot.go#L52.]
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
