DatabricksSubmitRunOperator¶
Use the DatabricksSubmitRunOperator
to submit
a new Databricks job via Databricks api/2.1/jobs/runs/submit API endpoint.
Using the Operator¶
There are three ways to instantiate this operator. In the first way, you can take the JSON payload that you typically use
to call the api/2.1/jobs/runs/submit
endpoint and pass it directly to our DatabricksSubmitRunOperator
through the
json
parameter. With this approach you get full control over the underlying payload to Jobs REST API, including
execution of Databricks jobs with multiple tasks, but it’s harder to detect errors because of the lack of the type checking.
json = {
"new_cluster": {"spark_version": "2.1.0-db3-scala2.11", "num_workers": 2},
"notebook_task": {
"notebook_path": "/Users/airflow@example.com/PrepareData",
},
}
notebook_run = DatabricksSubmitRunOperator(task_id="notebook_run", json=json)
The second way to accomplish the same thing is to use the named parameters of the DatabricksSubmitRunOperator
directly. Note that there is exactly
one named parameter for each top level parameter in the runs/submit
endpoint. When using named parameters you must to specify following:
Task specification - it should be one of:
spark_jar_task
- main class and parameters for the JAR tasknotebook_task
- notebook path and parameters for the taskspark_python_task
- python file path and parameters to run the python file withspark_submit_task
- parameters needed to run aspark-submit
commandpipeline_task
- parameters needed to run a Delta Live Tables pipelinedbt_task
- parameters needed to run a dbt project
Cluster specification - it should be one of: *
new_cluster
- specs for a new cluster on which this task will be run *existing_cluster_id
- ID for existing cluster on which to run this taskpipeline_task
- may refer to either apipeline_id
orpipeline_name
In the case where both the json parameter AND the named parameters
are provided, they will be merged together. If there are conflicts during the merge,
the named parameters will take precedence and override the top level json
keys.
- Currently the named parameters that
DatabricksSubmitRunOperator
supports are spark_jar_task
notebook_task
spark_python_task
spark_submit_task
pipeline_task
dbt_task
git_source
new_cluster
existing_cluster_id
libraries
run_name
timeout_seconds
new_cluster = {"spark_version": "10.1.x-scala2.12", "num_workers": 2}
notebook_task = {
"notebook_path": "/Users/airflow@example.com/PrepareData",
}
notebook_run = DatabricksSubmitRunOperator(
task_id="notebook_run", new_cluster=new_cluster, notebook_task=notebook_task
)
Another way to do is use the param tasks to pass array of objects to instantiate this operator. Here the value of tasks param that is used to invoke api/2.1/jobs/runs/submit
endpoint is passed through the tasks
param in DatabricksSubmitRunOperator
. Instead of invoking single task, you can pass array of task and submit a one-time run.
tasks = [
{
"new_cluster": {"spark_version": "2.1.0-db3-scala2.11", "num_workers": 2},
"notebook_task": {"notebook_path": "/Users/airflow@example.com/PrepareData"},
}
]
notebook_run = DatabricksSubmitRunOperator(task_id="notebook_run", tasks=tasks)
Examples¶
Specifying parameters as JSON¶
An example usage of the DatabricksSubmitRunOperator is as follows:
# Example of using the JSON parameter to initialize the operator.
new_cluster = {
"spark_version": "9.1.x-scala2.12",
"node_type_id": "r3.xlarge",
"aws_attributes": {"availability": "ON_DEMAND"},
"num_workers": 8,
}
notebook_task_params = {
"new_cluster": new_cluster,
"notebook_task": {
"notebook_path": "/Users/airflow@example.com/PrepareData",
},
}
notebook_task = DatabricksSubmitRunOperator(task_id="notebook_task", json=notebook_task_params)
Using named parameters¶
You can also use named parameters to initialize the operator and run the job.
# Example of using the named parameters of DatabricksSubmitRunOperator
# to initialize the operator.
spark_jar_task = DatabricksSubmitRunOperator(
task_id="spark_jar_task",
new_cluster=new_cluster,
spark_jar_task={"main_class_name": "com.example.ProcessData"},
libraries=[{"jar": "dbfs:/lib/etl-0.1.jar"}],
)
DatabricksSubmitRunDeferrableOperator¶
Deferrable version of the DatabricksSubmitRunOperator
operator.
It allows to utilize Airflow workers more effectively using new functionality introduced in Airflow 2.2.0