<p>I am trying to run a beam application on spark on kubernetes.</p>
<p>beam-deployment.yaml</p>

<pre><code>apiVersion: apps/v1
kind: StatefulSet
metadata:
 name: spark-beam-jobserver
spec:
 serviceName: spark-headless
 selector:
   matchLabels:
     app: spark-beam-jobserver
 template:
   metadata:
     labels:
       app: spark-beam-jobserver
       app.kubernetes.io/instance: custom_spark
       app.kubernetes.io/name: spark
   spec:
     containers:
     - name: spark-beam-jobserver
       image: apache/beam_spark_job_server:2.33.0
       imagePullPolicy: Always
       ports:
       - containerPort: 8099
         name: jobservice
       - containerPort: 8098
         name: artifact
       - containerPort: 8097
         name: expansion
       volumeMounts:
         - name: beam-artifact-staging
           mountPath: &quot;/tmp/beam-artifact-staging&quot; 
          
       command: [
           &quot;/bin/bash&quot;, &quot;-c&quot;, &quot;./spark-job-server.sh --job-port=8099 --spark-master-url=spark://spark-primary:7077&quot;
       ]
     volumes:
     - name: beam-artifact-staging
       persistentVolumeClaim:
         claimName: spark-beam-pvc
---
apiVersion: v1
kind: Service
metadata:
 name: spark-beam-jobserver
 labels:
   app: spark-beam-jobserver
spec:
 selector:
   app: spark-beam-jobserver
 type: NodePort
 ports:
 - port: 8099
   nodePort: 32090
   name: job-service
 - port: 8098
   nodePort: 32091
   name: artifacts
#  type: ClusterIP
#  ports:
#  - port: 8099
#    name: job-service
#  - port: 8098
#    name: artifacts
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: spark-primary
spec:
  serviceName: spark-headless
  replicas: 1
  selector:
    matchLabels:
      app: spark
  template:
    metadata:
      labels:
        app: spark
        component: primary
        app.kubernetes.io/instance: custom_spark
        app.kubernetes.io/name: spark
    spec:
      containers:
      - name: primary
        image: docker.io/secondcomet/spark-custom-2.4.6
        env:
        - name: SPARK_MODE
          value: &quot;master&quot;
        - name: SPARK_RPC_AUTHENTICATION_ENABLED
          value: &quot;no&quot;
        - name: SPARK_RPC_ENCRYPTION_ENABLED
          value: &quot;no&quot;
        - name: SPARK_LOCAL_STORAGE_ENCRYPTION_ENABLED
          value: &quot;no&quot;
        - name: SPARK_SSL_ENABLED
          value: &quot;no&quot;
        ports:
        - containerPort: 7077
          name: masterendpoint
        - containerPort: 8080
          name: ui
        - containerPort: 7078
          name: driver-rpc-port
        - containerPort: 7079
          name: blockmanager
        livenessProbe:
          httpGet:
            path: /
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10                  
        resources:
          limits:
            cpu: 1.0
            memory: 1Gi
          requests:
            cpu: 0.5
            memory: 0.5Gi
---
apiVersion: v1
kind: Service
metadata:
  name: spark-primary
  labels:
    app: spark
    component: primary
spec:
  type: ClusterIP
  ports:
  - name: masterendpoint
    port: 7077
    targetPort: 7077
  - name: rest
    port: 6066
    targetPort: 6066
  - name: ui
    port: 8080
    targetPort: 8080

  - name: driver-rpc-port
    protocol: TCP 
    port: 7078
    targetPort: 7078
  - name: blockmanager
    protocol: TCP 
    port: 7079
    targetPort: 7079

  selector:
    app: spark
    component: primary
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: spark-children
  labels:
    app: spark
spec:
  serviceName: spark-headless
  replicas: 1
  selector:
    matchLabels:
      app: spark
  template:
    metadata:
      labels:
        app: spark
        component: children
        app.kubernetes.io/instance: custom_spark
        app.kubernetes.io/name: spark
    spec:
      containers:
      - name: docker
        image: docker:19.03.5-dind
        securityContext:
          privileged: true
        volumeMounts:
          - name: dind-storage
            mountPath: /var/lib/docker
        env:
          - name: DOCKER_TLS_CERTDIR
            value: &quot;&quot;
        resources:
          limits:
            cpu: 1.0
            memory: 1Gi
          requests:
            cpu: 0.5
            memory: 100Mi
      - name: children
        image: docker.io/secondcomet/spark-custom-2.4.6
        env:
        - name: DOCKER_HOST
          value: &quot;tcp://localhost:2375&quot;
        - name: SPARK_MODE
          value: &quot;worker&quot;
        - name: SPARK_MASTER_URL
          value: &quot;spark://spark-primary:7077&quot;
        - name: SPARK_WORKER_MEMORY
          value: &quot;1G&quot;
        - name: SPARK_WORKER_CORES
          value: &quot;1&quot;
        - name: SPARK_RPC_AUTHENTICATION_ENABLED
          value: &quot;no&quot;
        - name: SPARK_RPC_ENCRYPTION_ENABLED
          value: &quot;no&quot;
        - name: SPARK_LOCAL_STORAGE_ENCRYPTION_ENABLED
          value: &quot;no&quot;
        - name: SPARK_SSL_ENABLED
          value: &quot;no&quot;
        ports:
          - containerPort: 8081
            name: ui
        volumeMounts:
          - name: beam-artifact-staging
            mountPath: &quot;/tmp/beam-artifact-staging&quot;
        resources:
          limits:
            cpu: 1
            memory: 2Gi
          requests:
            cpu: 0.5
            memory: 1Gi
      volumes:
      - name: dind-storage
        emptyDir:
      - name: beam-artifact-staging
        persistentVolumeClaim:
          claimName: spark-beam-pvc
---
apiVersion: v1
kind: Service
metadata:
  name: spark-children
  labels:
    app: spark
    component: children
spec:
  type: ClusterIP
  ports:
  - name: ui
    port: 8081
    targetPort: 8081
  selector:
    app: spark
    component: children
---
apiVersion: v1
kind: Service
metadata:
  name: spark-headless
spec:
  clusterIP: None
  selector:
    app.kubernetes.io/instance: custom_spark
    app.kubernetes.io/name: spark
  type: ClusterIP
</code></pre>


<pre><code>$ kubectl get all --namespace spark-beam
NAME                         READY   STATUS    RESTARTS   AGE
pod/spark-beam-jobserver-0   1/1     Running   0          58m
pod/spark-children-0         2/2     Running   0          58m
pod/spark-primary-0          1/1     Running   0          58m

NAME                           TYPE        CLUSTER-IP      EXTERNAL-IP   PORT(S)
      AGE
service/spark-beam-jobserver   NodePort    10.97.173.68    &lt;none&gt;        8099:32090/TCP,8098:32091/TCP
      58m
service/spark-children         ClusterIP   10.105.209.30   &lt;none&gt;        8081/TCP
      58m
service/spark-headless         ClusterIP   None            &lt;none&gt;        &lt;none&gt;
      58m
service/spark-primary          ClusterIP   10.109.32.126   &lt;none&gt;        7077/TCP,6066/TCP,8080/TCP,7078/TCP,7079/TCP   58m

NAME                                    READY   AGE
statefulset.apps/spark-beam-jobserver   1/1     58m
statefulset.apps/spark-children         1/1     58m
statefulset.apps/spark-primary          1/1     58m
</code></pre>

<p>beam-application.py</p>

<pre><code>import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

class ConvertToByteArray(beam.DoFn):
    def __init__(self):
        pass

    def setup(self):
        pass

    def process(self, row):
        try:
            yield bytearray(row + '\n', 'utf-8')

        except Exception as e:
            raise e

def run():
    options = PipelineOptions([
  
        &quot;--runner=PortableRunner&quot;,
        &quot;--job_endpoint=localhost:32090&quot;,
        &quot;--save_main_session&quot;,
        &quot;--environment_type=DOCKER&quot;,
        &quot;--environment_config=docker.io/apache/beam_python3.7_sdk:2.33.0&quot;

    ])

    with beam.Pipeline(options=options) as p:
        lines = (p
        | 'Create words' &gt;&gt; beam.Create(['this is working'])
        | 'Split words' &gt;&gt; beam.FlatMap(lambda words: words.split(' '))
        | 'Build byte array' &gt;&gt; beam.ParDo(ConvertToByteArray())
        | 'Group' &gt;&gt; beam.GroupBy() # Do future batching here
        | 'print output' &gt;&gt; beam.Map(print)
        )

if __name__ == &quot;__main__&quot;:
    run()
</code></pre>

<p>When I am trying to run the python application in my conda environment:
python beam-application.py</p>
<p>I am getting the below error :</p>

<pre><code>  File &quot;beam.py&quot;, line 39, in &lt;module&gt;
    run()
  File &quot;beam.py&quot;, line 35, in run
    | 'print output' &gt;&gt; beam.Map(print)
  File &quot;C:\Users\eapasnr\Anaconda3\envs\oden2\lib\site-packages\apache_beam\pipeline.py&quot;, line 586, in __exit__   
    self.result = self.run()
  File &quot;C:\Users\eapasnr\Anaconda3\envs\oden2\lib\site-packages\apache_beam\pipeline.py&quot;, line 565, in run        
    return self.runner.run_pipeline(self, self._options)
  File &quot;C:\Users\eapasnr\Anaconda3\envs\oden2\lib\site-packages\apache_beam\runners\portability\portable_runner.py&quot;, line 440, in run_pipeline
    job_service_handle.submit(proto_pipeline)
  File &quot;C:\Users\eapasnr\Anaconda3\envs\oden2\lib\site-packages\apache_beam\runners\portability\portable_runner.py&quot;, line 114, in submit
    prepare_response.staging_session_token)
  File &quot;C:\Users\eapasnr\Anaconda3\envs\oden2\lib\site-packages\apache_beam\runners\portability\portable_runner.py&quot;, line 218, in stage
    staging_session_token)
  File &quot;C:\Users\eapasnr\Anaconda3\envs\oden2\lib\site-packages\apache_beam\runners\portability\artifact_service.py&quot;, line 237, in offer_artifacts
    for request in requests:
  File &quot;C:\Users\eapasnr\Anaconda3\envs\oden2\lib\site-packages\grpc\_channel.py&quot;, line 426, in __next__
    return self._next()
  File &quot;C:\Users\eapasnr\Anaconda3\envs\oden2\lib\site-packages\grpc\_channel.py&quot;, line 826, in _next
    raise self
grpc._channel._MultiThreadedRendezvous: &lt;_MultiThreadedRendezvous of RPC that terminated with:
        status = StatusCode.UNAVAILABLE
        details = &quot;failed to connect to all addresses; last error: UNAVAILABLE: WSA Error&quot;
        debug_error_string = &quot;UNKNOWN:Failed to pick subchannel {created_time:&quot;2022-10-10T14:38:39.520460502+00:00&quot;, children:[UNKNOWN:failed to connect to all addresses; last error: UNAVAILABLE: WSA Error {grpc_status:14, created_time:&quot;2022-10-10T14:38:39.520457024+00:00&quot;}]}&quot;
&gt;
</code></pre>

<p>I am not sure where exactly is the problem.</p>
<p>What should I pass in job_endpoint and artifact_endpoint?</p>
<p>I also tried port-forwarding :</p>
<pre><code>kubectl port-forward service/spark-beam-jobserver 32090:8099 --namespace spark-beam
kubectl port-forward service/spark-primary 8080:8080 --namespace spark-beam
kubectl port-forward service/spark-children 8081:8081 --namespace spark-beam
</code></pre>

<p>I suppose this is based on <a href="https://github.com/cometta/python-apache-beam-spark" rel="nofollow noreferrer">https://github.com/cometta/python-apache-beam-spark</a>?</p>
<p><code>spark-beam-jobserver</code> is using service type <code>NodePort</code>. So, if running in a local (minikube) cluster, you won't need any port forwarding to reach the job server.</p>
<p>You should be able to submit a Python job from your local shell using the following pipeline options:</p>
<pre><code>  --job_endpoint=localhost:32090
  --artifact_endpoint=localhost:32091
</code></pre>
<p>Note, your python code above misses the <code>artifact_endpoint</code>. You have to provide both endpoints.</p>
