<p>Attempting to run the <a href="https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/wordcount.py" rel="nofollow noreferrer">word count example</a> on minikube following the Flink Kubernetes instructions <a href="https://ci.apache.org/projects/flink/flink-docs-release-1.10/ops/deployment/kubernetes.html#setup-kubernetes" rel="nofollow noreferrer">here</a>, but the job never completes. The Python Beam SDK worker pooler doesn't appear to do any work.</p>
<p>In addition to the instructions for configuring a Flink Kubernetes cluster, I added a Python SDK worker pool to the taskmanager deployment. If I understand correctly, the purpose of the worker pool is to execute the Python portions of the pipeline. See full k8s deployment.</p>
<pre class="lang-yaml prettyprint-override"><code>apiVersion: apps/v1
kind: Deployment
metadata:
  name: flink-taskmanager
  namespace: flink-test
spec:
  replicas: 1
  selector:
    matchLabels:
      app: flink
      component: taskmanager
  template:
    metadata:
      labels:
        app: flink
        component: taskmanager
    spec:
      containers:
      - name: taskmanager
        image: flink:1.10.2-scala_2.11
        workingDir: /opt/flink
        command: [&quot;/bin/bash&quot;, &quot;-c&quot;, &quot;$FLINK_HOME/bin/taskmanager.sh start; \
          while :;
          do
            if [[ -f $(find log -name '*taskmanager*.log' -print -quit) ]];
              then tail -f -n +1 log/*taskmanager*.log;
            fi;
          done&quot;]
        ports:
        - containerPort: 6122
          name: rpc
        livenessProbe:
          tcpSocket:
            port: 6122
          initialDelaySeconds: 30
          periodSeconds: 60
        volumeMounts:
        - name: flink-config-volume
          mountPath: /opt/flink/conf/
        securityContext:
          runAsUser: 9999  # refers to user _flink_ from official flink image, change if necessary
      - name: beam-worker-pool
        image: apache/beam_python3.7_sdk:2.24.0
        args: [&quot;--worker_pool&quot;]
        ports:
        - containerPort: 50000
          name: pool
        livenessProbe:
          tcpSocket:
            port: 50000
          initialDelaySeconds: 30
          periodSeconds: 60
        volumeMounts:
        - name: flink-config-volume
          mountPath: /opt/flink/conf/
        securityContext:
          runAsUser: 9999
      volumes:
      - name: flink-config-volume
        configMap:
          name: flink-config
          items:
          - key: flink-conf.yaml
            path: flink-conf.yaml
          - key: log4j.properties
            path: log4j.properties
</code></pre>
<p>I am running the example as follows:</p>
<pre class="lang-none prettyprint-override"><code>python -m apache_beam.examples.wordcount \
--output /tmp/results/count \
--runner FlinkRunner \
--flink_master=localhost:8081 \
--environment_type=EXTERNAL \
--environment_config=localhost:50000
</code></pre>
<p>I used the documentation at <a href="https://beam.apache.org/documentation/runtime/sdk-harness-config/" rel="nofollow noreferrer">https://beam.apache.org/documentation/runtime/sdk-harness-config/</a> to set the <code>environment_type</code> and <code>environment_config</code> values.</p>
<p>The job gets added to the job manager and I can view it in the Flink UI, but the job never completes. I started poking around the container logs as noticed the beam-worker-pool has the following logs:</p>
<pre class="lang-none prettyprint-override"><code>Starting worker with command ['/opt/apache/beam/boot', '--id=1-1', '--logging_endpoint=localhost:46005', '--artifact_endpoint=localhost:43851', '--provision_endpoint=localhost:37079', '--control_endpoint=localhost:37961']
2020/09/28 16:44:00 Provision info:
pipeline_options:&lt;fields: fields: &gt; fields: &gt; fields: &gt; fields: &gt; &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; &gt; &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; fields: &gt; &gt; logging_endpoint: artifact_endpoint: control_endpoint: dependencies: 
2020/09/28 16:44:00 Initializing python harness: /opt/apache/beam/boot --id=1-1 --logging_endpoint=localhost:46005 --artifact_endpoint=localhost:43851 --provision_endpoint=localhost:37079 --control_endpoint=localhost:37961
2020/09/28 16:44:08 Failed to retrieve staged files: failed to retrieve /tmp/staged in 3 attempts: failed to retrieve chunk for /tmp/staged/pickled_main_session
    caused by:
rpc error: code = Unknown desc = ; failed to retrieve chunk for /tmp/staged/pickled_main_session
    caused by:
rpc error: code = Unknown desc = ; failed to retrieve chunk for /tmp/staged/pickled_main_session
    caused by:
rpc error: code = Unknown desc = ; failed to retrieve chunk for /tmp/staged/pickled_main_session
    caused by:
rpc error: code = Unknown desc = 
</code></pre>
<p>Likewise the taskmanger is logging:</p>
<pre class="lang-none prettyprint-override"><code>2020-09-28 16:46:00,155 INFO  org.apache.beam.runners.fnexecution.environment.ExternalEnvironmentFactory  - Still waiting for startup of environment from localhost:50000 for worker id 1-1
</code></pre>
<p>Not sure what I am missing. I checked the <code>/tmp/staging/pickled_main_session</code> on the worker pool and it is empty.</p>
<p>Note this issue is similar to these previous questions.
<a href="https://stackoverflow.com/questions/57851158/how-do-i-run-beam-python-pipelines-using-flink-deployed-on-kubernetes">How do I run Beam Python pipelines using Flink deployed on Kubernetes?</a>
<a href="https://stackoverflow.com/questions/60411012/running-apache-beam-python-pipelines-in-kubernetes">Running Apache Beam python pipelines in Kubernetes</a></p>

Nathan B
<p>By default (as of this writing), Beam stages runtime dependencies (&quot;artifacts&quot;) to a certain directory (/tmp/staged by default) that needs to be accessible to both the job server (in your case, the client) and the Beam worker.</p>
<p>You can get around this by setting the <code>--flink_submit_uber_jar</code> pipeline option. When <code>--flink_submit_uber_jar</code> is set, Beam wraps all your dependencies in a jar that is submitted to Flink.</p>
