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 c4945d0 [BEAM-3041] Support various pip based dependency options for
python sdk (#4286)
c4945d0 is described below
commit c4945d0d1366a8a5b177994aa169b90f9c762537
Author: Ahmet Altay <[email protected]>
AuthorDate: Tue Dec 19 16:38:00 2017 -0800
[BEAM-3041] Support various pip based dependency options for python sdk
(#4286)
---
sdks/python/container/boot.go | 57 ++++++++++++++---
sdks/python/container/piputil.go | 131 +++++++++++++++++++++++++++++++++++++++
2 files changed, 178 insertions(+), 10 deletions(-)
diff --git a/sdks/python/container/boot.go b/sdks/python/container/boot.go
index 31c8267..69f031b 100644
--- a/sdks/python/container/boot.go
+++ b/sdks/python/container/boot.go
@@ -20,13 +20,15 @@ package main
import (
"context"
"flag"
+ "fmt"
"log"
"os"
"path/filepath"
"strings"
"github.com/apache/beam/sdks/go/pkg/beam/artifact"
- pb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
+ pbpipeline "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
+ pbjob "github.com/apache/beam/sdks/go/pkg/beam/model/jobmanagement_v1"
"github.com/apache/beam/sdks/go/pkg/beam/provision"
"github.com/apache/beam/sdks/go/pkg/beam/util/execx"
"github.com/apache/beam/sdks/go/pkg/beam/util/grpcx"
@@ -44,6 +46,15 @@ var (
semiPersistDir = flag.String("semi_persist_dir", "/tmp", "Local
semi-persistent directory (optional).")
)
+const (
+ sdkHarnessEntrypoint = "apache_beam.runners.worker.sdk_worker_main"
+ // Please keep these names in sync with setup dependency.py
+ workflowFile = "workflow.tar.gz"
+ requirementsFile = "requirements.txt"
+ sdkFile = "dataflow_python_sdk.tar"
+ extraPackagesFile = "extra_packages.txt"
+)
+
func main() {
flag.Parse()
if *id == "" {
@@ -81,36 +92,62 @@ func main() {
dir := filepath.Join(*semiPersistDir, "staged")
- _, err = artifact.Materialize(ctx, *artifactEndpoint, dir)
+ files, err := artifact.Materialize(ctx, *artifactEndpoint, dir)
if err != nil {
log.Fatalf("Failed to retrieve staged files: %v", err)
}
// TODO(herohde): the packages to install should be specified
explicitly. It
// would also be possible to install the SDK in the Dockerfile.
- if err := pipInstall(joinPaths(dir, "dataflow_python_sdk.tar[gcp]"));
err != nil {
- log.Fatalf("Failed to install SDK: %v", err)
+ if setupErr := installSetupPackages(files, dir); setupErr != nil {
+ log.Fatalf("Failed to install SDK: %v", setupErr)
}
// (3) Invoke python
os.Setenv("PIPELINE_OPTIONS", options)
os.Setenv("SEMI_PERSISTENT_DIRECTORY", *semiPersistDir)
- os.Setenv("LOGGING_API_SERVICE_DESCRIPTOR",
proto.MarshalTextString(&pb.ApiServiceDescriptor{Url: *loggingEndpoint}))
- os.Setenv("CONTROL_API_SERVICE_DESCRIPTOR",
proto.MarshalTextString(&pb.ApiServiceDescriptor{Url: *controlEndpoint}))
+ os.Setenv("LOGGING_API_SERVICE_DESCRIPTOR",
proto.MarshalTextString(&pbpipeline.ApiServiceDescriptor{Url:
*loggingEndpoint}))
+ os.Setenv("CONTROL_API_SERVICE_DESCRIPTOR",
proto.MarshalTextString(&pbpipeline.ApiServiceDescriptor{Url:
*controlEndpoint}))
args := []string{
"-m",
- "apache_beam.runners.worker.sdk_worker_main",
+ sdkHarnessEntrypoint,
}
log.Printf("Executing: python %v", strings.Join(args, " "))
log.Fatalf("Python exited: %v", execx.Execute("python", args...))
}
-// pipInstall runs pip install with the given args.
-func pipInstall(args []string) error {
- return execx.Execute("pip", append([]string{"install"}, args...)...)
+// installSetupPackages installs Beam SDK and user dependencies.
+func installSetupPackages(mds []*pbjob.ArtifactMetadata, workDir string) error
{
+ log.Printf("Installing setup packages ...")
+
+ files := make([]string, len(mds))
+ for i, v := range mds {
+ log.Printf("Found artifact: %s", v.Name)
+ files[i] = v.Name
+ }
+
+ // Install the Dataflow Python SDK and worker packages.
+ // We install the extra requirements in case of using the beam sdk.
These are ignored by pip
+ // if the user is using an SDK that does not provide these.
+ if err := pipInstallPackage(files, workDir, sdkFile, false, false,
[]string{"gcp"}); err != nil {
+ return fmt.Errorf("failed to install SDK: %v", err)
+ }
+ // The staged files will not disappear due to restarts because workDir
is a
+ // folder that is mapped to the host (and therefore survives restarts).
+ if err := pipInstallRequirements(files, workDir, requirementsFile); err
!= nil {
+ return fmt.Errorf("failed to install requirements: %v", err)
+ }
+ if err := installExtraPackages(files, extraPackagesFile, workDir); err
!= nil {
+ return fmt.Errorf("failed to install extra packages: %v", err)
+ }
+ if err := pipInstallPackage(files, workDir, workflowFile, false, true,
nil); err != nil {
+ return fmt.Errorf("failed to install workflow: %v", err)
+ }
+
+ return nil
}
// joinPaths joins the dir to every artifact path. Each / in the path is
diff --git a/sdks/python/container/piputil.go b/sdks/python/container/piputil.go
new file mode 100644
index 0000000..b227774
--- /dev/null
+++ b/sdks/python/container/piputil.go
@@ -0,0 +1,131 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+ "bufio"
+ "bytes"
+ "errors"
+ "fmt"
+ "io/ioutil"
+ "log"
+ "path/filepath"
+ "strings"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/util/execx"
+)
+
+const (
+ pip = "/usr/local/bin/pip"
+)
+
+// pipInstallRequirements installs the given requirement, if present.
+func pipInstallRequirements(files []string, dir, name string) error {
+ for _, file := range files {
+ if file == name {
+ // We run the install process in two rounds in order to avoid as much
+ // as possible PyPI downloads. In the first round the --find-links
+ // option will make sure that only things staged in the worker will be
+ // used without following their dependencies.
+ args := []string{"install", "-r", filepath.Join(dir, name),
"--no-index", "--no-deps", "--find-links", dir}
+ if err := execx.Execute(pip, args...); err != nil {
+ return err
+ }
+ // The second install round opens up the search for packages on PyPI and
+ // also installs dependencies. The key is that if all the packages have
+ // been installed in the first round then this command will be a no-op.
+ args = []string{"install", "-r", filepath.Join(dir, name),
"--find-links", dir}
+ return execx.Execute(pip, args...)
+ }
+ }
+ return nil
+}
+
+// pipInstallPackage installs the given package, if present.
+func pipInstallPackage(files []string, dir, name string, force, optional bool,
extras []string) error {
+ for _, file := range files {
+ if file == name {
+ var packageSpec = name
+ if extras != nil {
+ packageSpec += "[" + strings.Join(extras, ",") + "]"
+ }
+ if force {
+ // We only use force reinstallation for packages specified using the
+ // --extra_package flag. In this case, we always want to use the
+ // user-specified package, overwriting any existing package already
+ // installed. At the same time, we want to avoid reinstalling any
+ // dependencies. The "pip install" command doesn't have a clean way
to do
+ // this, so we do this in two steps.
+ //
+ // First, we use the three flags "--upgrade --force-reinstall
--no-deps"
+ // to "pip install" so as to force the package to be reinstalled, while
+ // avoiding reinstallation of dependencies. Note now that if any
needed
+ // dependencies were not installed, they will still be missing.
+ //
+ // Next, we run "pip install" on the package without any flags. Since
the
+ // installed version will match the package specified, the package
itself
+ // will not be reinstalled, but its dependencies will now be resolved
and
+ // installed if necessary. This achieves our goal outlined above.
+ args := []string{"install", "--upgrade", "--force-reinstall",
"--no-deps",
+ filepath.Join(dir, packageSpec)}
+ err := execx.Execute(pip, args...)
+ if err != nil {
+ return err
+ }
+ args = []string{"install", filepath.Join(dir, packageSpec)}
+ return execx.Execute(pip, args...)
+ }
+
+ // Case when we do not perform a forced reinstall.
+ args := []string{"install", filepath.Join(dir, packageSpec)}
+ return execx.Execute(pip, args...)
+ }
+ }
+ if optional {
+ return nil
+ }
+ return errors.New("package '" + name + "' not found")
+}
+
+// installExtraPackages installs all the packages declared in the extra
+// packages manifest file.
+func installExtraPackages(files []string, extraPackagesFile, dir string) error
{
+ // First check that extra packages manifest file is present.
+ for _, file := range files {
+ if file != extraPackagesFile {
+ continue
+ }
+
+ // Found the manifest. Install extra packages.
+ manifest, err := ioutil.ReadFile(filepath.Join(dir, extraPackagesFile))
+ if err != nil {
+ return fmt.Errorf("failed to read extra packages manifest file: %v", err)
+ }
+
+ s := bufio.NewScanner(bytes.NewReader(manifest))
+ s.Split(bufio.ScanLines)
+
+ for s.Scan() {
+ extraPackage := s.Text()
+ log.Printf("Installing extra package: %s", extraPackage)
+ if err = pipInstallPackage(files, dir, extraPackage, true, false, nil);
err != nil {
+ return fmt.Errorf("failed to install extra package %s: %v",
extraPackage, err)
+ }
+ }
+ return nil
+ }
+ return nil
+}
--
To stop receiving notification emails like this one, please contact
['"[email protected]" <[email protected]>'].
