Databricks

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.

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.

1. Get access

Use a service principal with OAuth machine-to-machine (M2M) authentication (Authorize service principal access to Databricks with OAuth).

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

2. Query new rows

Use the official Node.js driver, @databricks/sql, with authType: 'databricks-oauth' (Databricks SQL Driver for Node.js). 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), so the cursor and the message time are both plain numbers.

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.

// databricks.mjs. Environment: DATABRICKS_SERVER_HOSTNAME, DATABRICKS_HTTP_PATH, DATABRICKS_CLIENT_ID, DATABRICKS_CLIENT_SECRET, DL_API_KEY.
import { DBSQLClient } from '@databricks/sql';
import { sendToDimensionLabs, toMillis } from './dimensionlabs.mjs';

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, and run it with the 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 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.

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

// export.mjs. Environment: DL_EXPORT_KEY, DL_INTEGRATION_ID. Usage: node export.mjs 2026-10-05 2026-10-13
import { writeFile } from 'node:fs/promises';

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). 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
• Unity Catalog permissions model concepts
• Get connection details for a Databricks compute resource
• Databricks SQL Driver for Node.js
• unix_millis function
• Automate jobs with schedules and triggers
• Get started using COPY INTO to load data

Checked against Databricks’ documentation on 2026-10-13.