> ## Documentation Index
> Fetch the complete documentation index at: https://docs.coreflux.org/llms.txt
> Use this file to discover all available pages before exploring further.

# Databricks

> Client-mode REST API connector for Databricks: run SQL (INSERT or SELECT) on a SQL warehouse with the Statement Execution API, trigger a job run when a batch completes, and land files in Unity Catalog volumes with the Files API, plus statement polling, run status and volume listing

Technical Preview

## Overview

A client-mode connector for the [Databricks REST API](https://docs.databricks.com/api/workspace/) that writes MQTT data into the lakehouse and kicks off downstream pipelines:

| Catalog connector event | Endpoint | Notes |
| - | - | - |
| Run SQL (INSERT or SELECT) | `POST /api/2.0/sql/statements` | [Statement Execution API](https://docs.databricks.com/api/workspace/statementexecution/executestatement). Needs `warehouse_id`. Execution errors come back as `HTTP 200` with `status.state: FAILED`. |
| Poll a statement | `GET /api/2.0/sql/statements/{statement_id}` | Status, manifest and first result chunk. Use it after `PENDING` / `RUNNING`. |
| Next result chunk | `GET /api/2.0/sql/statements/{statement_id}/result/chunks/{chunk_index}` | Follow `next_chunk_index`. |
| Cancel a statement | `POST /api/2.0/sql/statements/{statement_id}/cancel` | Empty response; poll to see `CANCELED`. |
| Trigger a pipeline | `POST /api/2.2/jobs/run-now` | [Jobs run-now](https://docs.databricks.com/api/workspace/jobs/runnow): `job_id`, `job_parameters`, `idempotency_token`. Returns `run_id`. |
| Run status | `GET /api/2.2/jobs/runs/get?run_id=…` | `status.state` (`QUEUED`, `RUNNING`, `TERMINATED`, …) and `termination_details`. |
| Land a file in a volume | `PUT /api/2.0/fs/files/Volumes/…?overwrite=…` | [Files API](https://docs.databricks.com/api/workspace/files/upload) (Public Preview): raw bytes, up to 5 GiB, `204 No Content`. |
| Read a file / metadata | `GET` / `HEAD /api/2.0/fs/files/Volumes/…` | Bytes, or `content-length` + `last-modified` headers only. |
| List / create a directory | `GET` / `PUT /api/2.0/fs/directories/Volumes/…` | `contents[]` listing; create is `mkdir -p`. |
| Who am I / warehouses | `GET /api/2.0/preview/scim/v2/Me`, `GET /api/2.0/sql/warehouses` | Credential smoke and `warehouse_id` lookup. |

`BASE_ADDRESS` is your workspace URL, with no trailing slash. Replace `YOUR_WORKSPACE.cloud.databricks.com` with yours. Topics use `data/databricks/<object>/<verb>`, and replies go to `<topic>/response`.

| Cloud | Workspace URL shape |
| - | - |
| AWS | `https://dbc-xxxxxxxx-xxxx.cloud.databricks.com` (or a named `https://<name>.cloud.databricks.com`) |
| Azure | `https://adb-<workspace-id>.<n>.azuredatabricks.net` |
| GCP | `https://<workspace-id>.<n>.gcp.databricks.com` |

See [Get identifiers for workspace objects](https://docs.databricks.com/aws/en/workspace/workspace-details). One connector instance talks to one workspace.

> [!WARNING]
>
> **API versions differ by service.** SQL Statement Execution, Files, SCIM and Warehouses are under `/api/2.0`. Jobs uses `/api/2.2` (the `/api/2.1` paths still work, but 2.2 paginates the `tasks` array in `runs/get`). The catalog base URL `…/api/2.0/` is not a single prefix for every event, so each endpoint carries its own version.

## Auth

Every call sends `Authorization: Bearer …`. The template reads the token from the MQTT payload field `access_token`. Keep it in a Hub secret such as `GET SECRET "DATABRICKS_TOKEN"` and never commit it.

* **Personal access token (PAT):** a `dapi…` token from **Settings → Developer → Access tokens**. Use a dedicated service principal's token in production, not a person's. See [Databricks personal access tokens](https://docs.databricks.com/aws/en/dev-tools/auth/pat).
* **OAuth M2M (service principal):** mint an access token out of band with `POST https://<workspace>/oidc/v1/token`, `grant_type=client_credentials`, `scope=all-apis`, and the service principal's client id and secret. Tokens last about **1 hour**, so the publisher must refresh them. See [OAuth M2M](https://docs.databricks.com/aws/en/dev-tools/auth/oauth-m2m). On Azure, a Microsoft Entra ID token for the Azure Databricks resource also works.
* **Permissions:** `CAN USE` on the SQL warehouse, table privileges (`USE CATALOG`, `USE SCHEMA`, `SELECT`, `MODIFY`) for SQL, `CAN MANAGE RUN` on the job, and `READ VOLUME` / `WRITE VOLUME` on the Unity Catalog volume.

A `401` response with `Credential was not sent or was of an unsupported type for this API.` means the token is missing or malformed. `403` with `Invalid access token` means it was rejected or has expired.

## Basic Syntax

```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
DEFINE ROUTE DatabricksRestApi WITH TYPE REST_API
    ADD REST_API_CONFIG
        WITH BASE_ADDRESS "https://YOUR_WORKSPACE.cloud.databricks.com"
        WITH ENABLE_CLIENT "true"
        WITH USE_SSL "true"
        WITH REQUEST_TIMEOUT '60'
        WITH DEFAULT_CONTENT_TYPE "application/json"
        # Auth (prefer Hub secrets / GET SECRET "DATABRICKS_TOKEN"):
        #   PAT or OAuth access_token on each MQTT payload → Bearer header below (refresh OAuth before ~1 h expiry).
        #   Aliases: DATABRICKS_ACCESS_TOKEN, DATABRICKS_PAT.
        # Smoke: GET /api/2.0/preview/scim/v2/Me → 200; POST /api/2.0/sql/statements
        #   {"warehouse_id":"…","statement":"SELECT 1","wait_timeout":"30s"} → 200 + status.state SUCCEEDED.
        # REQUEST_TIMEOUT 60 s covers the 50 s max wait_timeout.
    ADD EVENT GetCurrentUser
        WITH SOURCE_TOPIC "data/databricks/user/me"
        WITH DESTINATION_TOPIC "data/databricks/user/me/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/api/2.0/preview/scim/v2/Me"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
    ADD EVENT ListWarehouses
        WITH SOURCE_TOPIC "data/databricks/warehouses/list"
        WITH DESTINATION_TOPIC "data/databricks/warehouses/list/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/api/2.0/sql/warehouses"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
    ADD EVENT ExecuteStatement
        WITH SOURCE_TOPIC "data/databricks/statement/execute"
        WITH DESTINATION_TOPIC "data/databricks/statement/execute/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/api/2.0/sql/statements"
        WITH HEADERS
            ADD "Content-Type" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
        WITH BODY "{value.json.body}"
    ADD EVENT GetStatement
        WITH SOURCE_TOPIC "data/databricks/statement/get"
        WITH DESTINATION_TOPIC "data/databricks/statement/get/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/api/2.0/sql/statements/{value.json.statement_id}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
    ADD EVENT GetStatementResultChunk
        WITH SOURCE_TOPIC "data/databricks/statement/chunk"
        WITH DESTINATION_TOPIC "data/databricks/statement/chunk/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/api/2.0/sql/statements/{value.json.statement_id}/result/chunks/{value.json.chunk_index}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
    ADD EVENT CancelStatement
        WITH SOURCE_TOPIC "data/databricks/statement/cancel"
        WITH DESTINATION_TOPIC "data/databricks/statement/cancel/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/api/2.0/sql/statements/{value.json.statement_id}/cancel"
        WITH HEADERS
            ADD "Content-Type" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
        WITH BODY "{value.json.body}"
    ADD EVENT RunJobNow
        WITH SOURCE_TOPIC "data/databricks/job/run-now"
        WITH DESTINATION_TOPIC "data/databricks/job/run-now/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/api/2.2/jobs/run-now"
        WITH HEADERS
            ADD "Content-Type" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
        WITH BODY "{value.json.body}"
    ADD EVENT GetJobRun
        WITH SOURCE_TOPIC "data/databricks/run/get"
        WITH DESTINATION_TOPIC "data/databricks/run/get/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/api/2.2/jobs/runs/get?run_id={value.json.run_id}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
    ADD EVENT UploadFile
        WITH SOURCE_TOPIC "data/databricks/file/upload"
        WITH DESTINATION_TOPIC "data/databricks/file/upload/response"
        WITH METHOD "PUT"
        WITH ENDPOINT "/api/2.0/fs/files{value.json.file_path}?overwrite={value.json.overwrite}"
        WITH HEADERS
            ADD "Content-Type" "application/octet-stream"
            ADD "Authorization" "Bearer {value.json.access_token}"
        WITH BODY "{value.json.content}"
    ADD EVENT DownloadFile
        WITH SOURCE_TOPIC "data/databricks/file/download"
        WITH DESTINATION_TOPIC "data/databricks/file/download/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/api/2.0/fs/files{value.json.file_path}"
        WITH HEADERS
            ADD "Accept" "application/octet-stream"
            ADD "Authorization" "Bearer {value.json.access_token}"
    ADD EVENT GetFileMetadata
        WITH SOURCE_TOPIC "data/databricks/file/metadata"
        WITH DESTINATION_TOPIC "data/databricks/file/metadata/response"
        WITH METHOD "HEAD"
        WITH ENDPOINT "/api/2.0/fs/files{value.json.file_path}"
        WITH HEADERS
            ADD "Authorization" "Bearer {value.json.access_token}"
    ADD EVENT ListDirectory
        WITH SOURCE_TOPIC "data/databricks/directory/list"
        WITH DESTINATION_TOPIC "data/databricks/directory/list/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/api/2.0/fs/directories{value.json.directory_path}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
    ADD EVENT CreateDirectory
        WITH SOURCE_TOPIC "data/databricks/directory/create"
        WITH DESTINATION_TOPIC "data/databricks/directory/create/response"
        WITH METHOD "PUT"
        WITH ENDPOINT "/api/2.0/fs/directories{value.json.directory_path}"
        WITH HEADERS
            ADD "Authorization" "Bearer {value.json.access_token}"
```

> [!NOTE]
>
> JSON request bodies come from `{value.json.body}`, not `{payload}`. The MQTT payload carries control fields (`access_token`, `statement_id`, `chunk_index`, `run_id`, `file_path`, `directory_path`, `overwrite`) next to a `body` object, and only `body` is sent to Databricks. `UploadFile` sends `{value.json.content}` as the raw file body with `Content-Type: application/octet-stream`.

## Example MQTT payloads

**Find a warehouse (topic `data/databricks/warehouses/list`):**

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{ "access_token": "<from DATABRICKS_TOKEN>" }
```

Use the `id` of a `RUNNING` (or serverless, auto-starting) warehouse as `warehouse_id`.

**Insert telemetry with SQL (topic `data/databricks/statement/execute`).** Use named parameter markers instead of string-building SQL:

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{
  "access_token": "<from DATABRICKS_TOKEN>",
  "body": {
    "warehouse_id": "abcdef0123456789",
    "catalog": "main",
    "schema": "iiot",
    "statement": "INSERT INTO telemetry (ts, site, temperature) VALUES (:ts, :site, :temperature)",
    "parameters": [
      { "name": "ts", "value": "2026-10-04T22:30:00Z", "type": "TIMESTAMP" },
      { "name": "site", "value": "porto" },
      { "name": "temperature", "value": "21.5", "type": "DOUBLE" }
    ],
    "wait_timeout": "30s"
  }
}
```

A finished insert returns `status.state: SUCCEEDED` with `num_affected_rows` / `num_inserted_rows` in `result.data_array`. One row per call is fine for low rates. For high-rate telemetry, batch rows into one multi-row `INSERT`, or land files with `UploadFile` and load them with `COPY INTO` or Auto Loader in a job.

**Query (same topic):**

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{
  "access_token": "<from DATABRICKS_TOKEN>",
  "body": {
    "warehouse_id": "abcdef0123456789",
    "statement": "SELECT site, avg(temperature) AS avg_t FROM main.iiot.telemetry WHERE ts > current_timestamp() - INTERVAL 1 HOUR GROUP BY site",
    "wait_timeout": "30s",
    "on_wait_timeout": "CONTINUE",
    "disposition": "INLINE",
    "format": "JSON_ARRAY"
  }
}
```

Rows come back in `result.data_array` as arrays of strings (`null` for NULL), with column names in `manifest.schema.columns`.

**Poll, page, or cancel (topics `data/databricks/statement/get`, `…/statement/chunk`, `…/statement/cancel`):**

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{ "access_token": "<from DATABRICKS_TOKEN>", "statement_id": "01eda0e7-e315-1846-84e2-79a963ffad44", "chunk_index": 1, "body": {} }
```

`GetStatement` only needs `statement_id`, `GetStatementResultChunk` also needs `chunk_index`, and `CancelStatement` needs `"body": {}`.

**Trigger a pipeline when a batch completes (topic `data/databricks/job/run-now`):**

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{
  "access_token": "<from DATABRICKS_TOKEN>",
  "body": {
    "job_id": 11223344,
    "job_parameters": { "batch_id": "porto-2026-10-04T22", "source": "coreflux" },
    "idempotency_token": "porto-2026-10-04T22"
  }
}
```

The response is `{"run_id": 455644833, "number_in_job": 455644833}`. Poll `data/databricks/run/get` with `{ "access_token": "…", "run_id": 455644833 }`.

**Land a batch file in a volume (topic `data/databricks/file/upload`):**

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{
  "access_token": "<from DATABRICKS_TOKEN>",
  "file_path": "/Volumes/main/iiot/landing/porto/2026-10-04T22.csv",
  "overwrite": "true",
  "content": "ts,site,temperature\n2026-10-04T22:30:00Z,porto,21.5\n"
}
```

A successful upload returns `204` with an empty body. Check it with `data/databricks/file/metadata` (`HEAD`, `{ "access_token": "…", "file_path": "…" }`), or list the folder with `data/databricks/directory/list` (`directory_path`).

> [!TIP]
>
> `file_path` and `directory_path` must start with `/` (for example `/Volumes/<catalog>/<schema>/<volume>/…`). The template joins them straight onto `/api/2.0/fs/files` and `/api/2.0/fs/directories` with no extra slash. Always send `overwrite` (`"true"` or `"false"`), because the template puts it in the query string. With `"false"`, uploading to an existing path fails with `409`.

## Gotchas

* **`warehouse_id` is required** for every statement. SQL runs on a SQL warehouse (serverless or pro recommended), not on an all-purpose cluster. A stopped classic warehouse starts on the first statement, which can take minutes, so expect `PENDING`.
* **`wait_timeout`** is `"0s"` (asynchronous: returns `PENDING` and a `statement_id` straight away) or `"5s"` to `"50s"` (synchronous up to that limit; default `10s`). Any other value returns `400`. If the limit passes, `on_wait_timeout: CONTINUE` (default) keeps it running for you to poll, and `CANCEL` stops it. The connector's `REQUEST_TIMEOUT` is 60 s, which is above the 50 s maximum.
* **Check `status.state`, not just the HTTP code.** SQL errors such as syntax errors or missing tables come back as `HTTP 200` with `status.state: FAILED` and `status.error.message`. Statements stay readable for at least 12 hours after they finish, then `GetStatement` returns `404`.
* **`INLINE` vs `EXTERNAL_LINKS`:** `INLINE` supports `JSON_ARRAY` only, up to 25 MiB per result; larger results are aborted. `EXTERNAL_LINKS` (up to 100 GiB; `JSON_ARRAY`, `CSV` or `ARROW_STREAM`) returns short-lived presigned / SAS / signed URLs (≤ 15 minutes) that must be fetched **without** the `Authorization` header. That second fetch is outside this connector.
* **Parameters** use named markers (`:name`); positional `?` is not supported. Values are strings, with an optional `type`.
* **`idempotency_token`** (max 64 chars) on `run-now`: if a run with that token exists, Databricks returns the same `run_id` instead of starting a second run. Derive it from the batch id so MQTT redelivery doesn't trigger duplicate pipelines.
* **Files API body is raw bytes.** Databricks stores exactly what it receives. `{value.json.content}` suits text (CSV, JSON lines); it does **not** decode base64, so binary formats such as Parquet need a different path. Max 5 GiB per `PUT`. `overwrite` defaults to `true` on the Databricks side when omitted.
* **Files API is Public Preview** and works on Unity Catalog volumes (`/Volumes/…`), workspace files, and DBFS paths. The legacy DBFS API (`/api/2.0/dbfs/put`, base64 in 1 MB blocks) is not templated.
* Deletes (files, directories, jobs, warehouses), job create / reset / update, and `runs/submit` are left out on purpose.

## Rate limits

Databricks applies per-workspace rate limits per endpoint and returns `429` with `Retry-After` when you exceed them. Statement Execution and `jobs/run-now` are meant for moderate rates (batches, not one call per sensor sample). Back off on `429` and `503 TEMPORARILY_UNAVAILABLE`. See [Resource limits](https://docs.databricks.com/aws/en/resources/limits).

## Testing

**Status: blocked on credentials.** The factory box has no Databricks workspace or token, so the connector has not been smoke-tested against Databricks. There is no credible local Databricks emulator, so no wire-format emulator smoke was run either. Unauthenticated probes of real Databricks hosts (`community.cloud.databricks.com`, `accounts.cloud.databricks.com`) returned `401` `Credential was not sent or was of an unsupported type for this API.` for `GET /api/2.0/preview/scim/v2/Me`, `GET /api/2.0/sql/warehouses` and `POST /api/2.0/sql/statements`, with and without a placeholder token. That only shows the hosts enforce auth on the same paths and error shape the template uses. `lot_lint` passed with no findings.

To clear the gate, use a real workspace URL, a token, a SQL warehouse id, a lab job id and a lab volume path:

1. `GET /api/2.0/preview/scim/v2/Me` returns `200`.
2. `POST /api/2.0/sql/statements` with `{"warehouse_id":"…","statement":"SELECT 1","wait_timeout":"30s"}` returns `200` with `status.state: SUCCEEDED`.
3. `PUT /api/2.0/fs/files/Volumes/<c>/<s>/<v>/coreflux-smoke.csv?overwrite=true` returns `204`, and `HEAD` on the same path returns `200`.
4. `POST /api/2.2/jobs/run-now` with the lab `job_id` returns `200` with a `run_id`.

## Related

* [REST API Connectors](/latest/lot-language/routes/rest-api-routes)
* [REST API Apps](/latest/lot-language/routes/rest-api-apps/index)
* [Google BigQuery](/latest/lot-language/routes/rest-api-apps/google-bigquery)
* [Amazon S3](/latest/lot-language/routes/rest-api-apps/amazon-s3)
* [Azure Blob Storage and Data Lake Gen2](/latest/lot-language/routes/rest-api-apps/azure-blob-storage-and-data-lake-gen2)


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.