我的 Spark 管道是用 Scala 撰寫的，而 Airflow DAG 是用 Python 撰寫的。
我正在開發一個功能來運行任務，當 dag 觸發日期等于表中的 run_date 時，否則跳過它。我的計劃是為此使用 ShortCircuitOperator。
我在 Scala 中撰寫了一個函式來從 hive 表中獲取該日期。我想從我的 Python DAG 檔案中呼叫這個函式，以便我獲取日期值并在 ShortCircuitOperator 中使用它。所以我正在尋找一種從 Python DAG 檔案呼叫我的 Scala 函式的方法。另外請建議是否有更好的方法來做到這一點。
skip_if_not_run_date = ShortCircuitOperator(
task_id='skip_if_not_run_date',
python_callable=getLatestRunDate, #need to call scala function here
dag=dag,
provide_context=True,
)
我需要呼叫的 Spark Scala 函式
def getLatestRunDate(df: DataFrame): DataFrame = {
val df = spark.table("my_hive_schema.my_run_catalog_table")
df
.filter(col("date_of_job_run").leq(current_date())
and col("month_nbr").geq(month(current_date())-2)
and col("yr_nbr").equalTo(year()))
.select("date_of_job_run")
}
uj5u.com熱心網友回復：
我建議將其拆分為兩個任務。在第一個任務中，呼叫 Scala 函式并使用 XCOM 回傳值。第二個任務可以是短路運算子，并且可以從第一個任務接收 XCOM 值作為引數。
轉載請註明出處，本文鏈接：https://www.uj5u.com/gongcheng/362478.html