josh-fell commented on code in PR #32221:
URL: https://github.com/apache/airflow/pull/32221#discussion_r1352699528
##########
airflow/providers/databricks/operators/databricks.py:
##########
@@ -161,6 +163,139 @@ def get_link(
return XCom.get_value(key=XCOM_RUN_PAGE_URL_KEY, ti_key=ti_key)
+class DatabricksCreateJobsOperator(BaseOperator):
+ """
+ Creates (or resets) a Databricks job using the
+ `api/2.1/jobs/create
+
<https://docs.databricks.com/dev-tools/api/latest/jobs.html#operation/JobsCreate>`_
+ (or `api/2.1/jobs/reset
+
<https://docs.databricks.com/dev-tools/api/latest/jobs.html#operation/JobsReset>`_)
+ API endpoint.
+
+ .. seealso::
+
https://docs.databricks.com/dev-tools/api/latest/jobs.html#operation/JobsCreate
+
+ :param json: A JSON object containing API parameters which will be passed
+ directly to the ``api/2.1/jobs/create`` endpoint. The other named
parameters
+ (i.e. ``name``, ``tags``, ``tasks``, etc.) to this operator will
+ be merged with this json dictionary if they are provided.
+ If there are conflicts during the merge, the named parameters will
+ take precedence and override the top level json keys. (templated)
+
+ .. seealso::
+ For more information about templating see
:ref:`concepts:jinja-templating`.
+ :param name: An optional name for the job.
+ :param tags: A map of tags associated with the job.
+ :param tasks: A list of task specifications to be executed by this job.
+ Array of objects (JobTaskSettings).
+ :param job_clusters: A list of job cluster specifications that can be
shared and reused by
+ tasks of this job. Array of objects (JobCluster).
+ :param email_notifications: Object (JobEmailNotifications).
+ :param webhook_notifications: Object (WebhookNotifications).
+ :param timeout_seconds: An optional timeout applied to each run of this
job.
+ :param schedule: Object (CronSchedule).
+ :param max_concurrent_runs: An optional maximum allowed number of
concurrent runs of the job.
+ :param git_source: An optional specification for a remote repository
containing the notebooks
+ used by this job's notebook tasks. Object (GitSource).
+ :param access_control_list: List of permissions to set on the job. Array
of object
+ (AccessControlRequestForUser) or object (AccessControlRequestForGroup)
or object
+ (AccessControlRequestForServicePrincipal).
+
+ .. seealso::
+ This will only be used on create. In order to reset ACL consider
using the Databricks
+ UI.
+ :param databricks_conn_id: Reference to the
+ :ref:`Databricks connection <howto/connection:databricks>`. (templated)
+ :param polling_period_seconds: Controls the rate which we poll for the
result of
+ this run. By default the operator will poll every 30 seconds.
+ :param databricks_retry_limit: Amount of times retry if the Databricks
backend is
+ unreachable. Its value must be greater than or equal to 1.
+ :param databricks_retry_delay: Number of seconds to wait between retries
(it
+ might be a floating point number).
+ :param databricks_retry_args: An optional dictionary with arguments passed
to ``tenacity.Retrying`` class.
+ """
+
+ # Used in airflow.models.BaseOperator
+ template_fields: Sequence[str] = ("json", "databricks_conn_id")
+ # Databricks brand color (blue) under white text
+ ui_color = "#1CB1C2"
+ ui_fgcolor = "#fff"
+
+ def __init__(
+ self,
+ *,
+ json: dict | None = None,
+ name: str | None = None,
+ tags: dict[str, str] | None = None,
+ tasks: list[jobs.JobTaskSettings] | None = None,
+ job_clusters: list[jobs.JobCluster] | None = None,
+ email_notifications: jobs.JobEmailNotifications | None = None,
+ webhook_notifications: jobs.JobWebhookNotifications | None = None,
+ timeout_seconds: int | None = None,
+ schedule: jobs.CronSchedule | None = None,
+ max_concurrent_runs: int | None = None,
+ git_source: jobs.GitSource | None = None,
+ access_control_list: list[jobs.AccessControlRequest] | None = None,
+ databricks_conn_id: str = "databricks_default",
+ polling_period_seconds: int = 30,
+ databricks_retry_limit: int = 3,
+ databricks_retry_delay: int = 1,
+ databricks_retry_args: dict[Any, Any] | None = None,
+ **kwargs,
+ ) -> None:
+ """Creates a new ``DatabricksCreateJobsOperator``."""
+ super().__init__(**kwargs)
+ self.json = json or {}
+ self.databricks_conn_id = databricks_conn_id
+ self.polling_period_seconds = polling_period_seconds
+ self.databricks_retry_limit = databricks_retry_limit
+ self.databricks_retry_delay = databricks_retry_delay
+ self.databricks_retry_args = databricks_retry_args
+ if name is not None:
+ self.json["name"] = name
+ if tags is not None:
+ self.json["tags"] = tags
Review Comment:
Totally fair. All of the Databricks operators would need to be cleaned up,
but I also haven't heard of anyone running into this issue within their use
cases.
I can log a separate issue and many we can get some folks to contribute!
It's a solid "good first issue".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
