airflow_flink_kubernetes_operator_sql_job_submit.py

1import datetime
2
3from airflow import DAG
4from airflow.providers.apache.flink.operators.flink_kubernetes import FlinkKubernetesOperator
5
6with DAG(
7    dag_id="example_flink_kubernetes_operator",
8    schedule=None,
9    start_date=datetime.datetime(2021, 1, 1),
10    template_searchpath=["/usr/local/airflow/sql/"],
11    catchup=False,
12) as dag:
13
14    # [START howto_operator_flink_kubernetes]
15    submit_flink_job = FlinkKubernetesOperator(
16        task_id="submit_flink_job",
17        name="flink-job-name",
18        namespace="default",
19        application_file="flink_job_definition.yaml",
20        kubernetes_conn_id="kubernetes_default",
21    )
22    # [END howto_operator_flink_kubernetes]