title: How do I run Beam Python pipelines using Flink deployed on Kubernetes?
body: <p>Does anybody know how to run Beam Python pipelines with Flink when Flink is running as pods in Kubernetes?</p>
  
  <p>I have successfully managed to run a Beam Python pipeline using the Portable runner and the job service pointing to a local Flink server running in Docker containers. </p>
  
  <p>I was able to achieve that mounting the Docker socket in my Flink containers, and running Flink as root process, so the class DockerEnvironmentFactory can create the Python harness container. </p>
  
  <p>Unfortunately, I can't use the same solution when Flink is running in Kubernetes. Moreover, I don't want to create the Python harness container using the Docker command from my pods. </p>
  
  <p>It seems that Bean runner automatically selects Docker for executing  Python pipelines. However, I noticed there is an implementation called ExternalEnvironmentFactory, but I am not sure how to use it. </p>
  
  <p>Is there a way to deploy a side container and use a different factory to run the Python harness process? What is the correct approach? </p>
  
  <p>This is the patch for DockerEnvironmentFactory:</p>
  
  <pre><code>diff -pr beam-release-2.15.0/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/environment/DockerEnvironmentFactory.java beam-release-2.15.0-1/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/environment/DockerEnvironmentFactory.java
  *** beam-release-2.15.0/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/environment/DockerEnvironmentFactory.java   2019-08-14 22:33:41.000000000 +0100
  --- beam-release-2.15.0-1/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/environment/DockerEnvironmentFactory.java 2019-09-09 16:02:07.000000000 +0100
  *************** package org.apache.beam.runners.fnexecut
  *** 19,24 ****
  --- 19,26 ----
  
    import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.MoreObjects.firstNonNull;
  
  + import java.net.InetAddress;
  + import java.net.UnknownHostException;
    import java.nio.file.Files;
    import java.nio.file.Paths;
    import java.time.Duration;
  *************** public class DockerEnvironmentFactory im
  *** 127,133 ****
            ImmutableList.&lt;String&gt;builder()
                .addAll(gcsCredentialArgs())
                // NOTE: Host networking does not work on Mac, but the command line flag is accepted.
  !             .add("--network=host")
                // We need to pass on the information about Docker-on-Mac environment (due to missing
                // host networking on Mac)
                .add("--env=DOCKER_MAC_CONTAINER=" + System.getenv("DOCKER_MAC_CONTAINER"));
  --- 129,135 ----
            ImmutableList.&lt;String&gt;builder()
                .addAll(gcsCredentialArgs())
                // NOTE: Host networking does not work on Mac, but the command line flag is accepted.
  !             .add("--network=flink")
                // We need to pass on the information about Docker-on-Mac environment (due to missing
                // host networking on Mac)
                .add("--env=DOCKER_MAC_CONTAINER=" + System.getenv("DOCKER_MAC_CONTAINER"));
  *************** public class DockerEnvironmentFactory im
  *** 222,228 ****
  
        private static ServerFactory getServerFactory() {
          ServerFactory.UrlFactory dockerUrlFactory =
  !           (host, port) -&gt; HostAndPort.fromParts(DOCKER_FOR_MAC_HOST, port).toString();
          if (RUNNING_INSIDE_DOCKER_ON_MAC) {
            // If we're already running in a container, we need to use a fixed port range due to
            // non-existing host networking in Docker-for-Mac. The port range needs to be published
  --- 224,230 ----
  
        private static ServerFactory getServerFactory() {
          ServerFactory.UrlFactory dockerUrlFactory =
  !               (host, port) -&gt; HostAndPort.fromParts(getCanonicalHostName(), port).toString();
          if (RUNNING_INSIDE_DOCKER_ON_MAC) {
            // If we're already running in a container, we need to use a fixed port range due to
            // non-existing host networking in Docker-for-Mac. The port range needs to be published
  *************** public class DockerEnvironmentFactory im
  *** 237,242 ****
  --- 239,252 ----
        }
      }
  
  +   private static String getCanonicalHostName() throws RuntimeException {
  +     try {
  +       return InetAddress.getLocalHost().getCanonicalHostName();
  +     } catch (UnknownHostException e) {
  +       throw new RuntimeException(e);
  +     }
  +   }
  +
      /** Provider for DockerEnvironmentFactory. */
      public static class Provider implements EnvironmentFactory.Provider {
        private final boolean retainDockerContainer;
  *************** public class DockerEnvironmentFactory im
  *** 269,275 ****
        public ServerFactory getServerFactory() {
          switch (getPlatform()) {
            case LINUX:
  !           return ServerFactory.createDefault();
            case MAC:
              return DockerOnMac.getServerFactory();
            default:
  --- 279,286 ----
        public ServerFactory getServerFactory() {
          switch (getPlatform()) {
            case LINUX:
  !           return DockerOnMac.getServerFactory();
  ! //          return ServerFactory.createDefault();
            case MAC:
              return DockerOnMac.getServerFactory();
            default:
  </code></pre>
  
  <p>This is the Docker compose file I use to run Flink:</p>
  
  <pre><code>version: '3.4'
  services:
    jobmanager:
      image: tenx/flink:1.8.1
      command: 'jobmanager'
      environment:
        JOB_MANAGER_RPC_ADDRESS: 'jobmanager'
        DOCKER_MAC_CONTAINER: 1
        FLINK_JM_HEAP: 128
      volumes:
        - jobmanager-data:/data
        - /var/run/docker.sock:/var/run/docker.sock
      ports:
        - target: 8081
          published: 8081
          protocol: tcp
          mode: ingress
      networks:
        - flink
    taskmanager:
      image: tenx/flink:1.8.1
      command: 'taskmanager'
      environment:
        JOB_MANAGER_RPC_ADDRESS: 'jobmanager'
        DOCKER_MAC_CONTAINER: 1
        FLINK_TM_HEAP: 1024
        TASK_MANAGER_NUMBER_OF_TASK_SLOTS: 2
      networks:
        - flink
      volumes:
        - taskmanager-data:/data
        - /var/run/docker.sock:/var/run/docker.sock
        - /var/folders:/var/folders
  volumes:
      jobmanager-data:
      taskmanager-data:
  networks:
    flink:
      external: true
  </code></pre>
  
  <p>This is my Python pipeline:</p>
  
  <pre><code>import apache_beam as beam
  import logging
  
  class LogElements(beam.PTransform):
  
      class _LoggingFn(beam.DoFn):
  
          def __init__(self, prefix=''):
              super(LogElements._LoggingFn, self).__init__()
              self.prefix = prefix
  
          def process(self, element, **kwargs):
              logging.info(self.prefix + str(element))
              yield element
  
      def __init__(self, label=None, prefix=''):
          super(LogElements, self).__init__(label)
          self.prefix = prefix
  
      def expand(self, input):
          input | beam.ParDo(self._LoggingFn(self.prefix))
  
  
  from apache_beam.options.pipeline_options import PipelineOptions
  
  options = PipelineOptions(["--runner=PortableRunner", "--job_endpoint=localhost:8099"])
  
  p = beam.Pipeline(options=options)
  
  (p | beam.Create([1, 2, 3, 4, 5]) | LogElements())
  
  p.run()
  </code></pre>
  
  <p>This is how I run the job service:</p>
  
  <blockquote>
    <p>gradle :runners:flink:1.8:job-server:runShadow -PflinkMasterUrl=localhost:8081</p>
  </blockquote>
  
  <p>Docker is automatically selected for executing the Python harness.</p>
  
  <p>I can change the image used to run the Python container:</p>
  
  <blockquote>
    <p>options = PipelineOptions(["--runner=PortableRunner", "--job_endpoint=localhost:8099", "--environment_type=DOCKER", "--environment_config=beam/python:latest"])</p>
  </blockquote>
  
  <p>I can disable Docker and enable the ExternalEnvironmentFactory: </p>
  
  <blockquote>
    <p>options = PipelineOptions(["--runner=PortableRunner", "--job_endpoint=localhost:8099", "--environment_type=EXTERNAL", "--environment_config=server"])</p>
  </blockquote>
  
  <p>but I have to implement some callback answering on <a href="http://server:80" rel="nofollow noreferrer">http://server:80</a>.</p>
  
  <p>Is there an implementation available?</p>