DatabricksRunOperator Execution date

Viewed 200
opr_run_now = DatabricksRunNowOperator(
task_id = 'run_now',
databricks_conn_id = 'databricks_default',
job_id = 754377,
notebook_params = meta_data,
dag = dag

) here

Is there way to pass execution date using databricks run operator.

2 Answers

What do you want to pass the execution_date to? What are you trying to achieve in the end? The following doc was helpful for me:

https://www.astronomer.io/guides/airflow-databricks

And here is an example where I am passing execution_date to be used in a python file run in Databricks. I'm capturing the execution_date using sys.argv.

from airflow import DAG
from airflow.providers.databricks.operators.databricks import (
    DatabricksRunNowOperator,
)
from datetime import datetime, timedelta

spark_python_task = {
    "python_file": "dbfs:/FileStore/sandbox/databricks_test_python_task.py"
}

# Define params for Run Now Operator
python_params = [
    "{{ execution_date }}",
    "{{ execution_date.subtract(hours=1) }}",
]

default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=2),
}

with DAG(
    dag_id="databricks_dag",
    start_date=datetime(2022, 3, 11),
    schedule_interval="@hourly",
    catchup=False,
    default_args=default_args,
    max_active_runs=1,
) as dag:

opr_run_now = DatabricksRunNowOperator(
    task_id="run_now",
    databricks_conn_id="Databricks",
    job_id=2060,
    python_params=python_params,
)

opr_run_now

There are two ways to set DatabricksRunOperator. One with named arguments (as you did) - which doesn't support templating. The second way is to use JSON payload that you typically use to call the api/2.0/jobs/run-now - This way also gives you the ability to pass execution_date as the json parameter is templated.

notebook_task_params = {
    'new_cluster': new_cluster,
    'notebook_task': {
        'notebook_path': '/test-{{ ds }}',
    }
DatabricksSubmitRunOperator(task_id='notebook_task', json=notebook_task_params)

For more information see the operator docs.

Related