Как автоматически создать идентификатор подключения spark_default - PullRequest
Новая
       4

Как автоматически создать идентификатор подключения spark_default

0 голосов
/ 23 мая 2019

Я хотел сгенерировать conn_id для spark_default . Я запускаю свой airflow на k8s, и я хотел сгенерировать conn_id на лету с spark master, который является другим контейнером, работающим в том же пространстве имен.

Есть ли способ генерировать conn_id на лету: что-то вроде:

  • переменные env
  • или используйте сам SparkSubmitOperator, чтобы написать и сгенерировать conn_id

Вот мой код: 

from airflow import DAG

from airflow.contrib.operators.spark_submit_operator import SparkSubmitOperator
from datetime import datetime, timedelta


args = {
    'owner': 'airflow',
    'start_date': datetime(2019, 5, 22)
}
dag = DAG('spark_example_new', default_args=args, schedule_interval="*/10 * * * *")

operator = SparkSubmitOperator(
    task_id='spark_submit_job_from_airflow',
    conn_id='spark_default',
    java_class='org.apache.spark.examples.JavaWordCount',
    application='local:///opt/spark/examples/jars/spark-examples_2.12-2.4.1.jar',
    total_executor_cores='1',
    executor_cores='2',
    executor_memory='2g',
    num_executors='1',
    name='airflow-spark-example-coming-from-aws-k8s',
    verbose=True,
    driver_memory='1g',
    application_args=["/opt/spark/data/graphx/users.txt"],
    dag=dag,
)

1 Ответ

1 голос
/ 23 мая 2019

Вы можете попробовать просмотреть этот ответ 

from airflow.models import Connection
from airflow import settings

def create_conn(username, password, host=None):
    new_conn = Connection(conn_id=f'{username}_connection',
                                  login=username,
                                  host=host if host else None)
    new_conn.set_password(password)

    session = settings.Session()
    session.add(new_conn)
    session.commit()
...