<p>I have already deployed Spark on Kubernetes, below is the deployment.yaml,</p>

<pre><code>apiVersion: &quot;sparkoperator.k8s.io/v1beta2&quot;
kind: SparkApplication
metadata:
  name: pyspark-pi
  namespace: default
spec:
  type: Python
  pythonVersion: &quot;3&quot;
  mode: cluster
  image: &quot;user/pyspark-app:1.0&quot;
  imagePullPolicy: Always
  mainApplicationFile: local:///app/pyspark-app.py
  sparkVersion: &quot;3.1.1&quot;
  restartPolicy:
    type: OnFailure
    onFailureRetries: 3
    onFailureRetryInterval: 10
    onSubmissionFailureRetries: 5
    onSubmissionFailureRetryInterval: 20
  driver:
    cores: 1
    coreLimit: &quot;1200m&quot;
    memory: &quot;512m&quot;
    labels:
      version: 3.1.1
    serviceAccount: spark
  executor:
    cores: 1
    instances: 2
    memory: &quot;512m&quot;
    labels:
      version: 3.1.1
</code></pre>

<p>Below is the service.yaml:</p>

<pre><code>apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: spark-operator-role
  namespace: default
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: edit
subjects:
  - kind: ServiceAccount
    name: spark
    namespace: default
</code></pre>

<p>GCP Spark operator is also installed on Kubernetes</p>
<p>Below are the services running:</p>
<pre><code>pyspark-pi-84dad9839f7f5f43-driver-svc   ClusterIP   None             &lt;none&gt;        7078/TCP,7079/TCP,4040/TCP   2d14h
</code></pre>
<p>Now I want to run a beam application on this driver. Please find the sample code for beam application below:</p>

<pre><code>import apache_beam as beam

from apache_beam.options.pipeline_options import PipelineOptions

options = PipelineOptions([
    &quot;--runner=PortableRunner&quot;,
    &quot;--job_endpoint=http://127.0.0.1:4040/&quot;,
    &quot;--environment_type=DOCKER&quot;,
    &quot;--environment_config=docker.io/apache/beam_python3.7_sdk:2.33.0&quot;
])

# lets have a sample string
data = [&quot;this is sample data&quot;, &quot;this is yet another sample data&quot;]

# create a pipeline
pipeline = beam.Pipeline(options=options)
counts = (pipeline | &quot;create&quot; &gt;&gt; beam.Create(data)
    | &quot;split&quot; &gt;&gt; beam.ParDo(lambda row: row.split(&quot; &quot;))
    | &quot;pair&quot; &gt;&gt; beam.Map(lambda w: (w, 1))
    | &quot;group&quot; &gt;&gt; beam.CombinePerKey(sum))

# lets collect our result with a map transformation into output array
output = []
def collect(row):
    output.append(row)
    return True

counts | &quot;print&quot; &gt;&gt; beam.Map(collect)

# Run the pipeline
result = pipeline.run()

# lets wait until result a available
result.wait_until_finish()

# print the output
print(output)
</code></pre>

<p>When I am trying to run the above apache beam application, it is throwing the below error:</p>

<pre><code>$ python beam2.py
WARNING:root:Make sure that locally built Python SDK docker image has Python 3.7 interpreter.
Traceback (most recent call last):
  File &quot;beam2.py&quot;, line 31, in &lt;module&gt;
    result = pipeline.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 438, in run_pipeline
    job_service_handle = self.create_job_service(options)
  File &quot;C:\Users\eapasnr\Anaconda3\envs\oden2\lib\site-packages\apache_beam\runners\portability\portable_runner.py&quot;, line 317, in create_job_service
    return self.create_job_service_handle(server.start(), options)
  File &quot;C:\Users\eapasnr\Anaconda3\envs\oden2\lib\site-packages\apache_beam\runners\portability\job_server.py&quot;, line 54, in start
    grpc.channel_ready_future(channel).result(timeout=self._timeout)
  File &quot;C:\Users\eapasnr\Anaconda3\envs\oden2\lib\site-packages\grpc\_utilities.py&quot;, line 139, in result
    self._block(timeout)
  File &quot;C:\Users\eapasnr\Anaconda3\envs\oden2\lib\site-packages\grpc\_utilities.py&quot;, line 85, in _block
    raise grpc.FutureTimeoutError()
grpc.FutureTimeoutError
</code></pre>

<p>I think the problem is with pipeline options and mainly with job_endpoint. Without pipeline options the applcication is running fine and giving the output to the console.</p>
<p>Which IP address and host should I provide to the job end point to make it work on spark.</p>

user19930511
<p>There's a couple of requirements to be met to use the <code>PortableRunner</code> with Spark:</p>
<ul>
<li>You have to also run a Spark job-server alongside your cluster. It takes care of submitting the application to spark (and will function as driver). Note, the Spark version of the job-server should closely match the version of your cluster!</li>
<li>Both, <code>job_endpoint</code> and <code>artifact_endpoint</code> have to point to the job-server (using respective ports) and not to Spark itself.</li>
<li>Finally, to use <code>environment_type=DOCKER</code>, you have to make sure Docker is installed on your Spark workers.</li>
</ul>
<p>Unfortunately Beam's documentation isn't great in that area. But I suggest you have a look at <a href="https://beam.apache.org/documentation/runners/spark/#running-on-a-pre-deployed-spark-cluster" rel="nofollow noreferrer">https://beam.apache.org/documentation/runners/spark/#running-on-a-pre-deployed-spark-cluster</a></p>
