如何发送参数SparkKubernetesOperator气流

Question

我在Kubernetes上运行了一个DAG。如何使用SparkKubernetesOperator将aws凭据发送到spark文件。

在我的DAG文件中，我从连接中获得凭据:例子:

from airflow.hooks.base import BaseHook
aws_conn = BaseHook.get_connection('aws_conn')

如何通过操作符将此aws_conn发送到spark文件?

transformation = SparkKubernetesOperator(
task_id='spark_transform_frete_new',
namespace='airflow',
application_file='spark/spark_transform_frete_new.yaml',
kubernetes_conn_id='kubernetes_default',
do_xcom_push=True,
)

文件:

apiVersion: "sparkoperator.k8s.io/v1beta2"
kind: SparkApplication
metadata:
name: "dag-example-spark-{{ macros.datetime.now().strftime("%Y-%m-%d-%H-%M-%S") }}-{{ task_instance.try_number }}"
namespace: airflow
spec:
timeToLiveSeconds: 30
volumes:
- name: ivy
persistentVolumeClaim:
claimName: dags-volume-pvc
- name: logs
persistentVolumeClaim:
claimName: logs-volume-pvc
sparkConf:
spark.jars.packages: "org.apache.hadoop:hadoop-aws:3.2.0,org.apache.spark:spark-avro_2.12:3.0.1"
spark.driver.extraJavaOptions: "-Divy.cache.dir=/tmp -Divy.home=/tmp"
"spark.kubernetes.local.dirs.tmpfs": "true"
"spark.eventLog.enabled": "true"
"spark.eventLog.dir": "/logs/spark/"
hadoopConf:
fs.s3a.impl: org.apache.hadoop.fs.s3a.S3AFileSystem
type: Python
pythonVersion: "3"
mode: cluster
image: "myimagespark/spark-dev"
imagePullPolicy: Always
mainApplicationFile: local:///dags/dag_example_python_spark/src/spark/spark_transform_frete_new.py 
sparkVersion: "3.1.1"
restartPolicy:
type: Never
driver:
cores: 1
coreLimit: "1200m"
memory: "4g"
labels:
version: 3.1.1
serviceAccount: spark
volumeMounts:
- name: ivy
mountPath: /dags
- name: logs
mountPath: /logs/spark/
executor:
cores: 2
instances: 2
memory: "3g"
labels:
version: 3.1.1
volumeMounts:
- name: ivy
mountPath: /dags
- name: logs
mountPath: /logs/spark/

Answer 1

您可以使用Jinja模板字段发送参数，如

mainApplicationFile: local:///dags/dag_example_python_spark/src/spark/spark_transform_frete_new.py       
arguments: {{ params.ARGUMENTS}}

从DAG中你可以按你想要的方式传递feed ARGUMENTS到参数