如何使用气流触发 google dataproc 作业并传递参数答案

【问题标题】：How to trigger google dataproc job using airflow and pass parameter as well如何使用气流触发 google dataproc 作业并传递参数
【发布时间】：2021-05-14 02:27:02
【问题描述】：

作为 DAG 的一部分，我正在使用以下代码触发 gcp pyspark dataproc 作业，

   dag=dag,
   gcp_conn_id=gcp_conn_id,
   region=region,
   main=pyspark_script_location_gcs,
   task_id='pyspark_job_1_submit',
   cluster_name=cluster_name,
   job_name="job_1"
)

如何将变量作为参数传递给可在脚本中访问的 pyspark 作业？

【问题讨论】：

标签： apache-spark google-cloud-platform pyspark airflow google-cloud-dataproc

【解决方案1】：

你可以使用DataProcPySparkOperator的参数arguments：

arguments（列表）- 工作的参数。（模板）

job = DataProcPySparkOperator(
    gcp_conn_id=gcp_conn_id,
    region=region,
    main=pyspark_script_location_gcs,
    task_id='pyspark_job_1_submit',
    cluster_name=cluster_name,
    job_name="job_1",
    arguments=[
        "-arg1=arg1_value", # or just "arg1_value" for non named args
        "-arg2=arg2_value"
    ],
    dag=dag
)

【讨论】：

我有一个字符串变量要传递.. 那么这里的格式应该是什么？在 spark 脚本中我如何访问它？
@j' 如果要传递一个变量，只需在运算符中使用arguments=[string_var]。要在 pyspark 主作业中获取变量，您可以使用 sys.argv 或更好地使用 argparse 包。你可以看到例子here关于如何传递python args