diff --git a/airflow/providers/databricks/operators/databricks.py b/airflow/providers/databricks/operators/databricks.py index a63eaa06c1550..7497f1b6e5f68 100644 --- a/airflow/providers/databricks/operators/databricks.py +++ b/airflow/providers/databricks/operators/databricks.py @@ -526,8 +526,8 @@ class DatabricksRunNowOperator(BaseOperator): .. seealso:: https://docs.databricks.com/dev-tools/api/latest/jobs.html#operation/JobsRunNow - :param python_named_parameters: A list of parameters for jobs with python wheel tasks, - e.g. "python_named_parameters": {"name": "john doe", "age": "35"}. + :param python_named_params: A list of named parameters for jobs with python wheel tasks, + e.g. "python_named_params": {"name": "john doe", "age": "35"}. If specified upon run-now, it would overwrite the parameters specified in job setting. This field will be templated. @@ -562,7 +562,7 @@ class DatabricksRunNowOperator(BaseOperator): token based authentication, provide the key ``token`` in the extra field for the connection and create the key ``host`` and leave the ``host`` field empty. (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. + 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 @@ -590,7 +590,7 @@ def __init__( python_params: Optional[List[str]] = None, jar_params: Optional[List[str]] = None, spark_submit_params: Optional[List[str]] = None, - python_named_parameters: Optional[Dict[str, str]] = None, + python_named_params: Optional[Dict[str, str]] = None, idempotency_token: Optional[str] = None, databricks_conn_id: str = 'databricks_default', polling_period_seconds: int = 30, @@ -621,8 +621,8 @@ def __init__( self.json['notebook_params'] = notebook_params if python_params is not None: self.json['python_params'] = python_params - if python_named_parameters is not None: - self.json['python_named_parameters'] = python_named_parameters + if python_named_params is not None: + self.json['python_named_params'] = python_named_params if jar_params is not None: self.json['jar_params'] = jar_params if spark_submit_params is not None: diff --git a/airflow/providers/databricks/operators/databricks_repos.py b/airflow/providers/databricks/operators/databricks_repos.py index 97b46b8e244d6..01461a69fcc3d 100644 --- a/airflow/providers/databricks/operators/databricks_repos.py +++ b/airflow/providers/databricks/operators/databricks_repos.py @@ -46,7 +46,7 @@ class DatabricksReposCreateOperator(BaseOperator): :param databricks_conn_id: Reference to the :ref:`Databricks connection `. By default and in the common case this will be ``databricks_default``. To use token based authentication, provide the key ``token`` in the extra field for the - connection and create the key ``host`` and leave the ``host`` field empty. + connection and create the key ``host`` and leave the ``host`` field empty. (templated) :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 @@ -54,7 +54,7 @@ class DatabricksReposCreateOperator(BaseOperator): """ # Used in airflow.models.BaseOperator - template_fields: Sequence[str] = ('repo_path', 'tag', 'branch') + template_fields: Sequence[str] = ('repo_path', 'tag', 'branch', 'databricks_conn_id') __git_providers__ = { "github.com": "gitHub", @@ -173,7 +173,7 @@ class DatabricksReposUpdateOperator(BaseOperator): :param databricks_conn_id: Reference to the :ref:`Databricks connection `. By default and in the common case this will be ``databricks_default``. To use token based authentication, provide the key ``token`` in the extra field for the - connection and create the key ``host`` and leave the ``host`` field empty. + connection and create the key ``host`` and leave the ``host`` field empty. (templated) :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 @@ -181,7 +181,7 @@ class DatabricksReposUpdateOperator(BaseOperator): """ # Used in airflow.models.BaseOperator - template_fields: Sequence[str] = ('repo_path', 'tag', 'branch') + template_fields: Sequence[str] = ('repo_path', 'tag', 'branch', 'databricks_conn_id') def __init__( self, @@ -246,7 +246,7 @@ class DatabricksReposDeleteOperator(BaseOperator): :param databricks_conn_id: Reference to the :ref:`Databricks connection `. By default and in the common case this will be ``databricks_default``. To use token based authentication, provide the key ``token`` in the extra field for the - connection and create the key ``host`` and leave the ``host`` field empty. + connection and create the key ``host`` and leave the ``host`` field empty. (templated) :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 @@ -254,7 +254,7 @@ class DatabricksReposDeleteOperator(BaseOperator): """ # Used in airflow.models.BaseOperator - template_fields: Sequence[str] = ('repo_path',) + template_fields: Sequence[str] = ('repo_path', 'databricks_conn_id') def __init__( self, diff --git a/airflow/providers/databricks/operators/databricks_sql.py b/airflow/providers/databricks/operators/databricks_sql.py index ad4add30fe8e6..cb088f634be20 100644 --- a/airflow/providers/databricks/operators/databricks_sql.py +++ b/airflow/providers/databricks/operators/databricks_sql.py @@ -42,7 +42,7 @@ class DatabricksSqlOperator(BaseOperator): :ref:`howto/operator:DatabricksSqlOperator` :param databricks_conn_id: Reference to - :ref:`Databricks connection id` + :ref:`Databricks connection id` (templated) :param http_path: Optional string specifying HTTP path of Databricks SQL Endpoint or cluster. If not specified, it should be either specified in the Databricks connection's extra parameters, or ``sql_endpoint_name`` must be specified. @@ -65,7 +65,14 @@ class DatabricksSqlOperator(BaseOperator): :param csv_params: parameters that will be passed to the ``csv.DictWriter`` class used to write CSV data. """ - template_fields: Sequence[str] = ('sql', '_output_path', 'schema', 'catalog', 'http_headers') + template_fields: Sequence[str] = ( + 'sql', + '_output_path', + 'schema', + 'catalog', + 'http_headers', + 'databricks_conn_id', + ) template_ext: Sequence[str] = ('.sql',) template_fields_renderers = {'sql': 'sql'} @@ -178,7 +185,7 @@ class DatabricksCopyIntoOperator(BaseOperator): :param file_format: Required file format. Supported formats are ``CSV``, ``JSON``, ``AVRO``, ``ORC``, ``PARQUET``, ``TEXT``, ``BINARYFILE``. :param databricks_conn_id: Reference to - :ref:`Databricks connection id` + :ref:`Databricks connection id` (templated) :param http_path: Optional string specifying HTTP path of Databricks SQL Endpoint or cluster. If not specified, it should be either specified in the Databricks connection's extra parameters, or ``sql_endpoint_name`` must be specified. @@ -209,6 +216,7 @@ class DatabricksCopyIntoOperator(BaseOperator): '_file_location', '_files', '_table_name', + 'databricks_conn_id', ) def __init__(