# Databricks > Send conversation messages stored in Databricks to Dimension Labs with the universal API. If your conversations already land in a Databricks table, you can send them to Dimension Labs from there. A scheduled job on your side queries the rows that are new since its last run and posts each message to the [universal API](/api/tracker). ## What you need - A Databricks workspace admin, to create a service principal and grant it access. - A SQL warehouse, and the service principal's OAuth client id and secret. - A Dimension Labs API key for a new **Universal API** source. See [Your API key](/api/tracker#your-api-key). ## 1. Get access Use a service principal with OAuth machine-to-machine (M2M) authentication ([Authorize service principal access to Databricks with OAuth](https://docs.databricks.com/aws/en/dev-tools/auth/oauth-m2m)). 1. Add a service principal to your account, then give it access to your workspace. 2. Create an OAuth secret: open **Settings** from your username menu, select **Identity and access**, click **Manage** next to Service principals, and select the service principal. On the **Secrets** tab, click **Generate secret**, set the lifetime and scopes, and click **Generate**. 3. Copy the secret and the client id. Databricks shows the secret only once. A secret can be valid for up to 730 days, and a service principal can have up to five secrets (Databricks' limits, from the same page). 4. Give the service principal permission to use the SQL warehouse. To read the table, it needs `USE CATALOG` on the catalog, `USE SCHEMA` on the schema and `SELECT` on the table: Databricks says all three are required ([Unity Catalog permissions model concepts](https://docs.databricks.com/aws/en/data-governance/unity-catalog/access-control/permissions-concepts)). 5. Copy the warehouse's connection details: in the sidebar, under **SQL**, open **SQL Warehouses**, click the warehouse, and copy the **Server hostname** and **HTTP path** from the **Connection Details** tab ([Get connection details for a Databricks compute resource](https://docs.databricks.com/aws/en/integrations/compute-details)). ## 2. Query new rows Use the official Node.js driver, `@databricks/sql`, with `authType: 'databricks-oauth'` ([Databricks SQL Driver for Node.js](https://docs.databricks.com/aws/dev-tools/nodejs-sql-driver)). Each run asks for the rows changed since the last run, using `updated_at` as the cursor. `unix_millis()` returns the milliseconds since `1970-01-01 00:00:00 UTC` for a timestamp ([unix_millis function](https://docs.databricks.com/aws/en/sql/language-manual/functions/unix_millis)), so the cursor and the message time are both plain numbers. ```sql SELECT message_id, conversation_id, customer_id, sender_type, body, channel, unix_millis(sent_at) AS sent_ms, unix_millis(updated_at) AS updated_ms FROM main.support.conversation_messages WHERE unix_millis(updated_at) >= ? ORDER BY updated_at ``` Pass the cursor in `ordinalParameters` when you call `session.executeStatement`, then read the rows with `fetchAll()`. ## 3. Map the columns This guide assumes a table with one row per message, with these columns. Rename them to match yours. | Column | Holds | |---|---| | `message_id` | The message's own id | | `conversation_id` | The conversation, ticket or chat the message belongs to | | `customer_id` | Your internal id for the customer in that conversation, on every row, including agent and bot rows. Not an email address or phone number. | | `sender_type` | Who wrote the message, for example `customer`, `agent` or `bot` | | `body` | The message text | | `channel` | Chat, email, voice and so on | | `sent_at` | When the message was sent (TIMESTAMP) | | `updated_at` | When the row was written or last changed (TIMESTAMP) | | Dimension Labs | Databricks column | Notes | |---|---|---| | `text` | `body` | Skip rows where it is empty. Strip HTML if bodies are HTML. | | `userId` | `customer_id` | If it is null, use the `sessionId`. | | `incoming` | `sender_type` | `true` only when the customer wrote it. Agent and bot rows are `false`. | | `sessionId` | `conversation_id` | Replace any colons. | | `conversationId` | `conversation_id` | Same value as `sessionId`. | | `dimensionlabs_timestamp` | `sent_ms` | Already milliseconds. Pass it through `toMillis`. | | `platformJson` | `message_id`, `channel` | Add any join keys you have, such as an account or store id. If your only customer key is an email, send a consistent hash of it instead. | ## 4. Send it Install the driver in your project first: `npm install @databricks/sql`. ```js // databricks.mjs. Environment: DATABRICKS_SERVER_HOSTNAME, DATABRICKS_HTTP_PATH, DATABRICKS_CLIENT_ID, DATABRICKS_CLIENT_SECRET, DL_API_KEY. const HOUR = 60 * 60 * 1000; const SQL = `SELECT message_id, conversation_id, customer_id, sender_type, body, channel, unix_millis(sent_at) AS sent_ms, unix_millis(updated_at) AS updated_ms FROM main.support.conversation_messages WHERE unix_millis(updated_at) >= ? ORDER BY updated_at`; const clean = (s) => String(s ?? '').replace(/<[^>]+>/g, ' ').replace(/\s+/g, ' ').trim(); async function readRows(since) { const client = new DBSQLClient(); await client.connect({ host: process.env.DATABRICKS_SERVER_HOSTNAME, path: process.env.DATABRICKS_HTTP_PATH, authType: 'databricks-oauth', oauthClientId: process.env.DATABRICKS_CLIENT_ID, oauthClientSecret: process.env.DATABRICKS_CLIENT_SECRET, }); try { const session = await client.openSession(); const operation = await session.executeStatement(SQL, { ordinalParameters: [since] }); const rows = await operation.fetchAll(); await operation.close(); await session.close(); return rows; } finally { await client.close(); } } // Sends every row written or changed at or after `since` (Unix ms). Returns the new `since`. export async function syncDatabricks({ since, historical = false }) { let newest = since; for (const row of await readRows(since)) { const text = clean(row.body); const id = String(row.conversation_id).replaceAll(':', '-'); if (text) { const ts = toMillis(Number(row.sent_ms)); await sendToDimensionLabs({ text, userId: row.customer_id ? String(row.customer_id) : id, incoming: row.sender_type === 'customer', sessionId: id, conversationId: id, dimensionlabs_timestamp: ts, platformJson: { message_id: String(row.message_id), channel: row.channel }, }, { historical: historical || ts < Date.now() - HOUR }); } newest = Math.max(newest, Number(row.updated_ms)); } return newest; } ``` The cursor follows `updated_at`, so a row that changes later (a status column, say) is sent again. That's fine: duplicates are removed automatically. ## 5. Run it Save the code as `databricks.mjs` in a project set up as in [Set up your project](/api/tracker#set-up-your-project), and run it with the [scheduled sync](/api/scheduled-sync) runner: `import { syncDatabricks as sync } from './databricks.mjs';`. - **History first:** `node run.mjs 2025-01-01` sends every row since that date to `/trackhistorical`. Then start a [backfill](/dimensions/backfill) from the **Backfill** tab. - **Then every 15 minutes:** the scheduled job sends new rows to `/track`, and they are enriched automatically every day. A message whose `sent_at` is more than an hour old when the job sees it (a late-loaded batch, say) goes to `/trackhistorical`, and needs a backfill. - **Re-runs are safe:** duplicates are removed automatically. You can also export the rows to a file and [upload it](/data/upload-a-file). To keep the job inside Databricks instead, write the same logic in Python, with `from dimensionlabs import send_to_dimension_labs, to_millis` ([dimensionlabs.py](/templates/dimensionlabs.py), next to your notebook or in a workspace file), and run it as a job with a **Scheduled** trigger: open the job, click **Add trigger** under **Schedules & Triggers**, select **Scheduled**, and save ([Automate jobs with schedules and triggers](https://docs.databricks.com/aws/en/jobs/triggers)). Keep the cursor in a small Delta table. The job's compute needs outbound access to `tracker.dimensionlabs.io`. ## Write Dimensions back To bring Dimensions into Databricks, pull them with the [Export API](/api/export) and load the files into a table. You need an Export API key, from **Settings** in the app (not the tracker key), and the integration ID of the source. ```js // export.mjs. Environment: DL_EXPORT_KEY, DL_INTEGRATION_ID. Usage: node export.mjs 2026-10-05 2026-10-13 const [startDate, endDate] = process.argv.slice(2); // endDate is exclusive const params = new URLSearchParams({ startDate, endDate, botIds: process.env.DL_INTEGRATION_ID }); const res = await fetch(`https://api.dimensionlabs.io/export?${params}`, { headers: { Authorization: process.env.DL_EXPORT_KEY } }); if (!res.ok) throw new Error(`Export returned ${res.status}: ${await res.text()}`); const { zipFile } = await res.json(); // zipFile is either a download link or the zip itself as base64 text; this handles both. const zip = /^https?:\/\//.test(zipFile) ? Buffer.from(await (await fetch(zipFile)).arrayBuffer()) : Buffer.from(zipFile, 'base64'); await writeFile('dimensions.zip', zip); ``` Then `unzip -o dimensions.zip -d dimensions/`. The zip holds the enrichments by day. Open one day's file to see its format and columns, and create your table to match. Put the files in a Unity Catalog volume or cloud storage location, and load them into a Delta table with `COPY INTO`, which skips files it has already loaded ([Get started using COPY INTO to load data](https://docs.databricks.com/aws/en/ingestion/copy-into/)). Give each export's files their own folder or names, so a re-exported day is loaded again. - `endDate` is exclusive, and dates are partitioned by US Pacific day. Overlap each export with the last by a day, so nothing falls between them. - The overlap returns the same rows again. Replace each exported day's rows in your table rather than appending them. ## Databricks reference - [Authorize service principal access to Databricks with OAuth](https://docs.databricks.com/aws/en/dev-tools/auth/oauth-m2m) - [Unity Catalog permissions model concepts](https://docs.databricks.com/aws/en/data-governance/unity-catalog/access-control/permissions-concepts) - [Get connection details for a Databricks compute resource](https://docs.databricks.com/aws/en/integrations/compute-details) - [Databricks SQL Driver for Node.js](https://docs.databricks.com/aws/dev-tools/nodejs-sql-driver) - [unix_millis function](https://docs.databricks.com/aws/en/sql/language-manual/functions/unix_millis) - [Automate jobs with schedules and triggers](https://docs.databricks.com/aws/en/jobs/triggers) - [Get started using COPY INTO to load data](https://docs.databricks.com/aws/en/ingestion/copy-into/) Checked against Databricks' documentation on 2026-10-13.