Databricks

Technical Preview

Overview

A client-mode connector for the Databricks REST API 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. 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: `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 (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. One connector instance talks to one workspace.

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.
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. 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

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}"

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):

{ "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:

{
  "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):

{
  "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):

{ "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):

{
  "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):

{
  "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).

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.

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:

GET /api/2.0/preview/scim/v2/Me returns 200.
POST /api/2.0/sql/statements with {"warehouse_id":"…","statement":"SELECT 1","wait_timeout":"30s"} returns 200 with status.state: SUCCEEDED.
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.
POST /api/2.2/jobs/run-now with the lab job_id returns 200 with a run_id.

Quick Start

Coreflux MQTT Broker

LoT Language

Coreflux HUB

VS Code - LoT Notebooks

Coreflux Portal

Using AI with Coreflux

Product Updates

Overview

Auth

Basic Syntax

Example MQTT payloads

Gotchas

Rate limits

Testing

​Overview

​Auth

​Basic Syntax

​Example MQTT payloads

​Gotchas

​Rate limits

​Testing

​Related

Overview

Auth

Basic Syntax

Example MQTT payloads

Gotchas

Rate limits

Testing

Related