diff --git a/airflow/providers/databricks/operators/databricks.py b/airflow/providers/databricks/operators/databricks.py
index 8956b43be1de7..6efea15be42eb 100644
--- a/airflow/providers/databricks/operators/databricks.py
+++ b/airflow/providers/databricks/operators/databricks.py
@@ -142,7 +142,9 @@ class DatabricksSubmitRunOperator(BaseOperator):
         - ``spark_jar_task``
         - ``notebook_task``
         - ``spark_python_task``
+        - ``spark_jar_task``
         - ``spark_submit_task``
+        - ``pipeline_task``
         - ``new_cluster``
         - ``existing_cluster_id``
         - ``libraries``
@@ -167,7 +169,7 @@ class DatabricksSubmitRunOperator(BaseOperator):
     :param spark_jar_task: The main class and parameters for the JAR task. Note that
         the actual JAR is specified in the ``libraries``.
         *EITHER* ``spark_jar_task`` *OR* ``notebook_task`` *OR* ``spark_python_task``
-        *OR* ``spark_submit_task`` should be specified.
+        *OR* ``spark_submit_task`` *OR* ``pipeline_task`` should be specified.
         This field will be templated.
 
         .. seealso::
@@ -175,7 +177,7 @@ class DatabricksSubmitRunOperator(BaseOperator):
     :type spark_jar_task: dict
     :param notebook_task: The notebook path and parameters for the notebook task.
         *EITHER* ``spark_jar_task`` *OR* ``notebook_task`` *OR* ``spark_python_task``
-        *OR* ``spark_submit_task`` should be specified.
+        *OR* ``spark_submit_task`` *OR* ``pipeline_task`` should be specified.
         This field will be templated.
 
         .. seealso::
@@ -183,7 +185,7 @@ class DatabricksSubmitRunOperator(BaseOperator):
     :type notebook_task: dict
     :param spark_python_task: The python file path and parameters to run the python file with.
         *EITHER* ``spark_jar_task`` *OR* ``notebook_task`` *OR* ``spark_python_task``
-        *OR* ``spark_submit_task`` should be specified.
+        *OR* ``spark_submit_task`` *OR* ``pipeline_task`` should be specified.
         This field will be templated.
 
         .. seealso::
@@ -191,21 +193,32 @@ class DatabricksSubmitRunOperator(BaseOperator):
     :type spark_python_task: dict
     :param spark_submit_task: Parameters needed to run a spark-submit command.
         *EITHER* ``spark_jar_task`` *OR* ``notebook_task`` *OR* ``spark_python_task``
-        *OR* ``spark_submit_task`` should be specified.
+        *OR* ``spark_submit_task`` *OR* ``pipeline_task`` should be specified.
         This field will be templated.
 
         .. seealso::
             https://docs.databricks.com/api/latest/jobs.html#jobssparksubmittask
     :type spark_submit_task: dict
+    :param pipeline_task: Parameters needed to execute a Delta Live Tables pipeline task.
+        The provided dictionary must contain at least ``pipeline_id`` field!
+        *EITHER* ``spark_jar_task`` *OR* ``notebook_task`` *OR* ``spark_python_task``
+        *OR* ``spark_submit_task`` *OR* ``pipeline_task`` should be specified.
+        This field will be templated.
+
+        .. seealso::
+            https://docs.databricks.com/dev-tools/api/2.0/jobs.html#jobspipelinetask
+    :type pipeline_task: dict
     :param new_cluster: Specs for a new cluster on which this task will be run.
-        *EITHER* ``new_cluster`` *OR* ``existing_cluster_id`` should be specified.
+        *EITHER* ``new_cluster`` *OR* ``existing_cluster_id`` should be specified
+        (except when ``pipeline_task`` is used).
         This field will be templated.
 
         .. seealso::
             https://docs.databricks.com/api/latest/jobs.html#jobsclusterspecnewcluster
     :type new_cluster: dict
     :param existing_cluster_id: ID for existing cluster on which to run this task.
-        *EITHER* ``new_cluster`` *OR* ``existing_cluster_id`` should be specified.
+        *EITHER* ``new_cluster`` *OR* ``existing_cluster_id`` should be specified
+        (except when ``pipeline_task`` is used).
         This field will be templated.
     :type existing_cluster_id: str
     :param libraries: Libraries which this run will use.
@@ -255,6 +268,7 @@ def __init__(
         notebook_task: Optional[Dict[str, str]] = None,
         spark_python_task: Optional[Dict[str, Union[str, List[str]]]] = None,
         spark_submit_task: Optional[Dict[str, List[str]]] = None,
+        pipeline_task: Optional[Dict[str, str]] = None,
         new_cluster: Optional[Dict[str, object]] = None,
         existing_cluster_id: Optional[str] = None,
         libraries: Optional[List[Dict[str, str]]] = None,
@@ -282,6 +296,8 @@ def __init__(
             self.json['spark_python_task'] = spark_python_task
         if spark_submit_task is not None:
             self.json['spark_submit_task'] = spark_submit_task
+        if pipeline_task is not None:
+            self.json['pipeline_task'] = pipeline_task
         if new_cluster is not None:
             self.json['new_cluster'] = new_cluster
         if existing_cluster_id is not None:
@@ -356,12 +372,15 @@ class DatabricksRunNowOperator(BaseOperator):
 
         python_params = ["douglas adams", "42"]
 
+        jar_params = ["douglas adams", "42"]
+
         spark_submit_params = ["--class", "org.apache.spark.examples.SparkPi"]
 
         notebook_run = DatabricksRunNowOperator(
             job_id=job_id,
             notebook_params=notebook_params,
             python_params=python_params,
+            jar_params=jar_params,
             spark_submit_params=spark_submit_params
         )
 
@@ -374,6 +393,7 @@ class DatabricksRunNowOperator(BaseOperator):
         - ``json``
         - ``notebook_params``
         - ``python_params``
+        - ``jar_params``
         - ``spark_submit_params``
 
 
@@ -420,6 +440,18 @@ class DatabricksRunNowOperator(BaseOperator):
         .. seealso::
             https://docs.databricks.com/api/latest/jobs.html#run-now
     :type python_params: list[str]
+    :param jar_params: A list of parameters for jobs with JAR tasks,
+        e.g. "jar_params": ["john doe", "35"].
+        The parameters will be passed to JAR file as command line parameters.
+        If specified upon run-now, it would overwrite the parameters specified in
+        job setting.
+        The json representation of this field (i.e. {"jar_params":["john doe","35"]})
+        cannot exceed 10,000 bytes.
+        This field will be templated.
+
+        .. seealso::
+            https://docs.databricks.com/api/latest/jobs.html#run-now
+    :type jar_params: list[str]
     :param spark_submit_params: A list of parameters for jobs with spark submit task,
         e.g. "spark_submit_params": ["--class", "org.apache.spark.examples.SparkPi"].
         The parameters will be passed to spark-submit script as command line parameters.
@@ -463,6 +495,7 @@ def __init__(
         json: Optional[Any] = None,
         notebook_params: Optional[Dict[str, str]] = None,
         python_params: Optional[List[str]] = None,
+        jar_params: Optional[List[str]] = None,
         spark_submit_params: Optional[List[str]] = None,
         databricks_conn_id: str = 'databricks_default',
         polling_period_seconds: int = 30,
@@ -485,6 +518,8 @@ def __init__(
             self.json['notebook_params'] = notebook_params
         if python_params is not None:
             self.json['python_params'] = python_params
+        if jar_params is not None:
+            self.json['jar_params'] = jar_params
         if spark_submit_params is not None:
             self.json['spark_submit_params'] = spark_submit_params
 
diff --git a/docs/apache-airflow-providers-databricks/operators.rst b/docs/apache-airflow-providers-databricks/operators.rst
index eb5170f839d84..0b8f8ac26f0ea 100644
--- a/docs/apache-airflow-providers-databricks/operators.rst
+++ b/docs/apache-airflow-providers-databricks/operators.rst
@@ -19,11 +19,12 @@
 
 .. _howto/operator:DatabricksSubmitRunOperator:
 
+
 DatabricksSubmitRunOperator
 ===========================
 
 Use the :class:`~airflow.providers.databricks.operators.DatabricksSubmitRunOperator` to submit
-an existing Spark job run to Databricks api/2.0/jobs/runs/submit <https://docs.databricks.com/api/latest/jobs.html#runs-submit> API endpoint.
+a new Databricks job via Databricks `api/2.0/jobs/runs/submit <https://docs.databricks.com/api/latest/jobs.html#runs-submit>`_ API endpoint.
 
 
 Using the Operator
@@ -42,13 +43,15 @@ one named parameter for each top level parameter in the ``runs/submit`` endpoint
    * - Parameter
      - Input
    * - spark_jar_task: dict
-     - main class and parameters for the JAR task
+     - `main class and parameters for the JAR task <https://docs.databricks.com/dev-tools/api/2.0/jobs.html#jobssparkjartask>`_
    * - notebook_task: dict
-     - notebook path and parameters for the task
+     - `notebook path and parameters for the task <https://docs.databricks.com/dev-tools/api/2.0/jobs.html#jobsnotebooktask>`_
    * - spark_python_task: dict
-     - python file path and parameters to run the python file with
+     - `python file path and parameters to run the python file with <https://docs.databricks.com/dev-tools/api/2.0/jobs.html#jobssparkpythontask>`_
    * - spark_submit_task: dict
-     - parameters needed to run a spark-submit command
+     - `parameters needed to run a spark-submit command <https://docs.databricks.com/dev-tools/api/2.0/jobs.html#jobssparksubmittask>`_
+   * - pipeline_task: dict
+     - `parameters needed to run a Delta Live Tables pipeline <https://docs.databricks.com/dev-tools/api/2.0/jobs.html#jobspipelinetask>`_
    * - new_cluster: dict
      - specs for a new cluster on which this task will be run
    * - existing_cluster_id: string
@@ -83,3 +86,49 @@ You can also use named parameters to initialize the operator and run the job.
     :language: python
     :start-after: [START howto_operator_databricks_named]
     :end-before: [END howto_operator_databricks_named]
+
+
+DatabricksRunNowOperator
+===========================
+
+Use the :class:`~airflow.providers.databricks.operators.DatabricksRunNowOperator` to trigger run of existing Databricks job
+via `api/2.0/jobs/runs/run-now <https://docs.databricks.com/dev-tools/api/2.0/jobs.html#run-now>`_ API endpoint.
+
+
+Using the Operator
+^^^^^^^^^^^^^^^^^^
+
+There are two ways to instantiate this operator. In the first way, you can take the JSON payload that you typically use
+to call the ``api/2.0/jobs/run-now`` endpoint and pass it directly to our ``DatabricksRunNowOperator`` through the ``json`` parameter.
+
+Another way to accomplish the same thing is to use the named parameters of the ``DatabricksRunNowOperator`` directly.
+Note that there is exactly one named parameter for each top level parameter in the ``jobs/run-now`` endpoint.
+
+.. list-table:: Databricks Airflow Connection Metadata
+   :widths: 15 25
+   :header-rows: 1
+
+   * - Parameter
+     - Input
+   * - job_id: str
+     - ID of the existing Databricks jobs (required)
+   * - jar_params: list[str]
+     - A list of parameters for jobs with JAR tasks, e.g. ``"jar_params": ["john doe", "35"]``. The parameters will be passed to JAR file as command line parameters. If specified upon run-now, it would        overwrite the parameters specified in job setting. The json representation of this field (i.e. ``{"jar_params":["john doe","35"]}``) cannot exceed 10,000 bytes. This field will be templated.
+   * - notebook_params: dict[str,str]
+     - A dict from keys to values for jobs with notebook task, e.g.``"notebook_params": {"name": "john doe", "age":  "35"}```. The map is passed to the notebook and will be accessible through the ``dbutils.widgets.get function``. See `Widgets <https://docs.databricks.com/notebooks/widgets.html>`_ for more information. If not specified upon run-now, the triggered run will use the job’s base parameters. ``notebook_params`` cannot be specified in conjunction with ``jar_params``. The json representation of this field (i.e. ``{"notebook_params":{"name":"john doe","age":"35"}}``) cannot exceed 10,000 bytes. This field will be templated.
+   * - python_params: list[str]
+     - A list of parameters for jobs with python tasks, e.g. ``"python_params": ["john doe", "35"]``. The parameters will be passed to python file as command line parameters. If specified upon run-now, it would overwrite the parameters specified in job setting. The json representation of this field (i.e. ``{"python_params":["john doe","35"]}``) cannot exceed 10,000 bytes. This field will be templated.
+   * - spark_submit_params: list[str]
+     - A list of parameters for jobs with spark submit task,  e.g. ``"spark_submit_params": ["--class", "org.apache.spark.examples.SparkPi"]``. The parameters will be passed to spark-submit script as command line parameters. If specified upon run-now, it would overwrite the parameters specified in job setting. The json representation of this field cannot exceed 10,000 bytes. This field will be templated.
+   * - timeout_seconds: int
+     - The timeout for this run. By default a value of 0 is used  which means to have no timeout. This field will be templated.
+   * - databricks_conn_id: string
+     - the name of the Airflow connection to use
+   * - polling_period_seconds: integer
+     - controls the rate which we poll for the result of this run
+   * - databricks_retry_limit: integer
+     - amount of times retry if the Databricks backend is unreachable
+   * - databricks_retry_delay: decimal
+     - number of seconds to wait between retries
+   * - do_xcom_push: boolean
+     - whether we should push run_id and run_page_url to xcom
diff --git a/tests/providers/databricks/operators/test_databricks.py b/tests/providers/databricks/operators/test_databricks.py
index e920a7ad96027..bf356ba1f991b 100644
--- a/tests/providers/databricks/operators/test_databricks.py
+++ b/tests/providers/databricks/operators/test_databricks.py
@@ -46,7 +46,7 @@
 EXISTING_CLUSTER_ID = 'existing-cluster-id'
 RUN_NAME = 'run-name'
 RUN_ID = 1
-JOB_ID = 42
+JOB_ID = "42"
 NOTEBOOK_PARAMS = {"dry-run": "true", "oldest-time-to-consider": "1457570074236"}
 JAR_PARAMS = ["param1", "param2"]
 RENDERED_TEMPLATED_JAR_PARAMS = [f'/test-{DATE}']
@@ -137,6 +137,18 @@ def test_init_with_specified_run_name(self):
         )
         assert expected == op.json
 
+    def test_pipeline_task(self):
+        """
+        Test the initializer with a specified run_name.
+        """
+        pipeline_task = {"pipeline_id": "test-dlt"}
+        json = {'new_cluster': NEW_CLUSTER, 'run_name': RUN_NAME, "pipeline_task": pipeline_task}
+        op = DatabricksSubmitRunOperator(task_id=TASK_ID, json=json)
+        expected = databricks_operator._deep_string_coerce(
+            {'new_cluster': NEW_CLUSTER, "pipeline_task": pipeline_task, 'run_name': RUN_NAME}
+        )
+        assert expected == op.json
+
     def test_init_with_merging(self):
         """
         Test the initializer when json and other named parameters are both
@@ -301,7 +313,8 @@ def test_init_with_merging(self):
         provided. The named parameters should override top level keys in the
         json dict.
         """
-        override_notebook_params = {'workers': 999}
+        override_notebook_params = {'workers': "999"}
+        override_jar_params = ['workers', "998"]
         json = {'notebook_params': NOTEBOOK_PARAMS, 'jar_params': JAR_PARAMS}
 
         op = DatabricksRunNowOperator(
@@ -310,13 +323,14 @@ def test_init_with_merging(self):
             job_id=JOB_ID,
             notebook_params=override_notebook_params,
             python_params=PYTHON_PARAMS,
+            jar_params=override_jar_params,
             spark_submit_params=SPARK_SUBMIT_PARAMS,
         )
 
         expected = databricks_operator._deep_string_coerce(
             {
                 'notebook_params': override_notebook_params,
-                'jar_params': JAR_PARAMS,
+                'jar_params': override_jar_params,
                 'python_params': PYTHON_PARAMS,
                 'spark_submit_params': SPARK_SUBMIT_PARAMS,
                 'job_id': JOB_ID,
