<p>I'm using Flink Kubernetes Operator 1.3.0 and need to pass some environment variables to a Python job. I have followed the <a href="https://github.com/apache/flink-kubernetes-operator/tree/release-1.3/examples/flink-python-example" rel="nofollow noreferrer">official documentation</a> and the example runs fine. How can I inject environment variables so that I can use it inside the python file?</p>
<p>EDIT:</p>
<p>Here's the yaml file that I've used. Its straight from the example link above:</p>
<pre><code>apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: python-example
spec:
  image: localhost:32000/flink-python-example:1.16.0
  flinkVersion: v1_16
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: &quot;1&quot;
  serviceAccount: flink
  jobManager:
    resource:
      memory: &quot;2048m&quot;
      cpu: 1
  taskManager:
    resource:
      memory: &quot;2048m&quot;
      cpu: 1
  job:
    jarURI: local:///opt/flink/opt/flink-python_2.12-1.16.0.jar # Note, this jarURI is actually a placeholder
    entryClass: &quot;org.apache.flink.client.python.PythonDriver&quot;
    args: [&quot;-pyclientexec&quot;, &quot;/usr/local/bin/python3&quot;, &quot;-py&quot;, &quot;/opt/flink/usrlib/python_demo.py&quot;]
    parallelism: 1
    upgradeMode: stateless
</code></pre>
<p>As you can see it's a custom resource of kind FlinkDeployment. And here's the python code:</p>
<pre class="lang-py prettyprint-override"><code>import logging
import sys

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment


def python_demo():
    env = StreamExecutionEnvironment.get_execution_environment()
    env.set_parallelism(1)

    t_env = StreamTableEnvironment.create(stream_execution_environment=env)
    t_env.execute_sql(&quot;&quot;&quot;
    CREATE TABLE orders (
      order_number BIGINT,
      price        DECIMAL(32,2),
      buyer        ROW&lt;first_name STRING, last_name STRING&gt;,
      order_time   TIMESTAMP(3)
    ) WITH (
      'connector' = 'datagen'
    )&quot;&quot;&quot;)

    t_env.execute_sql(&quot;&quot;&quot;
        CREATE TABLE print_table WITH ('connector' = 'print')
          LIKE orders&quot;&quot;&quot;)
    t_env.execute_sql(&quot;&quot;&quot;
        INSERT INTO print_table SELECT * FROM orders&quot;&quot;&quot;)


if __name__ == '__main__':
    logging.basicConfig(stream=sys.stdout, level=logging.INFO, format=&quot;%(message)s&quot;)
    python_demo()
</code></pre>

<p>Found the solution.</p>
<p>This is not detailed in the reference
<a href="https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-release-0.1/docs/custom-resource/reference" rel="nofollow noreferrer">https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-release-0.1/docs/custom-resource/reference</a></p>
<p>or example Flink Deployment
<a href="https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-release-0.1/docs/custom-resource/pod-template/" rel="nofollow noreferrer">https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-release-0.1/docs/custom-resource/pod-template/</a></p>
<p>But here it says: <a href="https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-release-0.1/docs/custom-resource/reference/#jobmanagerspec" rel="nofollow noreferrer">https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-release-0.1/docs/custom-resource/reference/#jobmanagerspec</a>
<code>JobManager pod template. It will be merged with FlinkDeploymentSpec.podTemplate</code></p>
<p>So I just added envFrom from the example in which shows you how to extend the FlinkDeployment CRD:</p>
<p><a href="https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-release-0.1/docs/custom-resource/pod-template/" rel="nofollow noreferrer">https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-release-0.1/docs/custom-resource/pod-template/</a></p>
<p>Confirmed this is working as I had to get this work for my own application now</p>
<pre><code>apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: python-example
spec:
  image: localhost:32000/flink-python-example:1.16.0
  flinkVersion: v1_16
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: &quot;1&quot;
  serviceAccount: flink
  jobManager:
    resource:
      memory: &quot;2048m&quot;
      cpu: 1
    podTemplate:
      apiVersion: v1
      kind: Pod
      metadata:
       name: pod-template
      spec:
        serviceAccount: flink
        containers:
        # Do not change the main container name
          - name: flink-main-container
            envFrom:
            - secretRef:
                name: &lt;SECRET RESOURCE NAME&gt;
  taskManager:
    resource:
      memory: &quot;2048m&quot;
      cpu: 1
  job:
    jarURI: local:///opt/flink/opt/flink-python_2.12-1.16.0.jar # Note, this jarURI is actually a placeholder
    entryClass: &quot;org.apache.flink.client.python.PythonDriver&quot;
    args: [&quot;-pyclientexec&quot;, &quot;/usr/local/bin/python3&quot;, &quot;-py&quot;, &quot;/opt/flink/usrlib/python_demo.py&quot;]
    parallelism: 1
    upgradeMode: stateless
</code></pre>
