Extract, transform, and load data in Databricks

This flow extracts data from any source, optionally transforms it, and loads it into Databricks. Etlworks stages each batch as a file in cloud storage or a Databricks Volume, then loads the file into the destination Delta table using COPY INTO. INSERT, MERGE, and CDC MERGE actions are supported. The destination table is auto-created (as a Delta table) and auto-altered for schema drift.

Which Databricks flow should I use?

How does this flow work?

1. Etlworks extracts rows from the source (database query, file, API call, queue message, …).
2. Rows are written to a staging file in cloud storage or a Databricks Volume. CSV is the default; JSON, Parquet, and Avro are also supported.
3. Etlworks executes COPY INTO against the destination Databricks table. The Delta table is created automatically if it doesn't exist (with USING DELTA).
4. For MERGE, the staging file is first loaded into a temporary Delta table, then merged into the destination using one of three strategies (see Configure MERGE below).
5. On success, staging files are purged.

What do I need before I start?

• A Databricks connection (see Get started with Databricks for setup).
• A stage connection — S3, ADLS Gen2, GCS, server storage, or Databricks Volume.
• The user / service principal must have READ FILES on the stage location's external location or volume, plus SELECT, MODIFY, and (for auto-create) CREATE TABLE on the target schema.

Step-by-step setup

Step 1. Create connections

You need at least two connections:

1. The source connection (database, file, API, queue, …).
2. The Databricks connection (PAT or OAuth Service Principal).
3. A stage connection if you're using cloud storage or a Volume separate from the Databricks workspace.

Step 2. Create a data exchange format

Create a CSV format (the default and most efficient for Databricks COPY INTO). JSON, Parquet, and Avro are also supported — pick based on the source shape and downstream needs.

Step 3. Create the flow

In Flows, click + and pick a flow type matching the source: Database to Databricks, File to Databricks, Web service to Databricks, Well-known API to Databricks, or Queue to Databricks.

Step 4. Set the Databricks connection on the flow

Set the named connection Databricks on the flow to the Databricks connection from step 1.

Step 5. Set the stage connection

If the destination cluster reads from cloud storage or a Volume, set the stage connection on the flow. For S3, ADLS Gen2, and GCS, Etlworks emits the stage path (s3://, abfss://, gs://) into the COPY INTO command. For Databricks Volumes, point a server-storage connection at /Volumes/<catalog>/<schema>/<volume>/<path>.

Step 6. Set source (FROM) and destination (TO)

In the transformation, set FROM to the source connection and object (table, file, endpoint, …). Set TO to the destination Databricks table. Use fully qualified catalog.schema.table if your workspace uses Unity Catalog.

Step 7. Set parameters

The Parameters tab includes Databricks-specific options:

Override the auto-generated COPY INTO SQL — optional

By default, Etlworks generates the COPY INTO statement from the destination table, the stage connection, the source file format, and the settings above. Set Override COPY INTO SQL when you need something the generator does not produce — a projection or a SELECT list on the source, COPY_OPTIONS (for example force or mergeSchema), a Databricks feature the generator does not know about, or a hand-tuned statement for a specific pipeline.

When this field is set, Etlworks uses your statement verbatim (after resolving the tokens below) instead of the generated one. The Format options setting in the table above and the auto-injected WITH (CREDENTIAL ...) clause are both bypassed — include everything the load needs in your statement.

Available tokens

Reference these tokens in your statement. The runtime resolves them per file, per batch, per destination table.

Structure of the auto-generated statement

Useful as a starting template for the override. The generator produces this shape:

COPY INTO <target-table>
  FROM '<stage-uri>/{FILE_TO_LOAD}[.{EXT} | *.*]'
  [WITH (CREDENTIAL <name-or-inline>)]
  FILEFORMAT = <CSV | JSON | PARQUET | AVRO>
  [FORMAT_OPTIONS (...)]

Details:

• Target: {TEMP_TABLE} for MERGE strategies, {TABLE} otherwise.
• Stage URI: composed from the stage connection's transport (S3, Azure Data Lake Storage Gen2, GCS, or server storage / Databricks Volume), bucket / container, and folder.
• File pattern: {FILE_TO_LOAD}.{EXT} for a single file, or {FILE_TO_LOAD}*.* when the flow is loading a wildcard match, split files, or archive-wrapped files (ZIP / GZIP).
• Credential: injected automatically when the Storage Credential setting is set or the stage connection has embedded AWS access / secret keys or an Azure SAS token. Skipped when the workspace has a Unity Catalog External Location or a managed Volume covering the stage path.
• FILEFORMAT: derived from the stage file extension.
• FORMAT_OPTIONS: emitted only for CSV, composed from header, delimiter, quote, line separator, date format, and timestamp format. The Format options setting above overrides just this clause without replacing the whole statement — use it in place of Override COPY INTO SQL when only parser options need to change.

Examples

MERGE flow — load into the temp Delta table with a custom escape and date/time formats:

COPY INTO {TEMP_TABLE}
FROM 's3://etlworks-stage/loads/{FILE_TO_LOAD}'
FILEFORMAT = CSV
FORMAT_OPTIONS ('header' = 'true', 'delimiter' = ',', 'quote' = '"',
                'dateFormat' = 'yyyy-MM-dd',
                'timestampFormat' = 'yyyy-MM-dd HH:mm:ss',
                'escape' = '"')

INSERT flow — load Parquet directly into the destination:

COPY INTO {TABLE}
FROM 's3://etlworks-stage/loads/{FILE_TO_LOAD}.{EXT}'
FILEFORMAT = PARQUET

Advanced — load selected columns with a SELECT list and force re-loads:

COPY INTO {TABLE} (id, name, updated_at)
FROM (SELECT id, name, current_timestamp() AS updated_at
      FROM 's3://etlworks-stage/loads/{FILE_TO_LOAD}.{EXT}')
FILEFORMAT = PARQUET
COPY_OPTIONS ('force' = 'true')

Interaction with other settings

• Format options is bypassed. Include your custom FORMAT_OPTIONS in the override statement itself.
• Storage Credential is bypassed. Include a WITH (CREDENTIAL ...) clause in your statement if the load needs one; the generator's automatic credential injection does not run when the SQL is overridden.
• Before COPY SQL and After COPY SQL still run — they are separate statements executed against the Databricks connection around your COPY INTO, not part of it.
• MERGE strategies (DELETE / INSERT, Native MERGE, Native MERGE with separate DELETE) still run on top of the loaded temp table. Your override handles only the load-into-temp step, not the MERGE step itself. Use {TEMP_TABLE} as the target when the flow uses MERGE, and let Etlworks generate the MERGE statement that follows.

When to use Override COPY INTO SQL vs Format options

• Only parser options need to change (escape, multiLine, quote handling, header, delimiter, date / timestamp format) — use Format options. Safer and narrower; keeps the generated target, stage URI, and credential handling intact.
• The load needs a projection or SELECT list on the source — use Override COPY INTO SQL. The generator emits no SELECT list.
• The load needs COPY_OPTIONS (for example force, mergeSchema, or a custom copy-options map) — use Override COPY INTO SQL. The generator does not emit COPY_OPTIONS.
• A hand-tuned statement is required for a specific pipeline, or a Databricks feature the generator does not know about needs to appear in the SQL — use Override COPY INTO SQL.

Step 8. Configure MERGE (UPSERT) — optional

Set Action to MERGE when you need UPSERT semantics. Enter the lookup fields in Lookup Fields (comma-separated) or enable Predict Lookup Fields to have Etlworks infer them from the destination's primary or unique key.

Pick a MERGE strategy in How to MERGE:

For custom MERGE logic, use User-defined MERGE SQL with these tokens: {TABLE}, {TEMP_TABLE}, {KEY_FIELDS}, {FIELDS}, {INSERT_FIELDS}, {UPDATE_FIELDS}, {UPDATE_CONDITIONS}.

Step 9. Handle source / destination schema differences

Two Databricks-specific options affect how the engine handles schema drift — both default to true and are unusual relative to other warehouses:

• Create temp tables with only TEXT columns (default true). The staging Delta table is created with every column as STRING, then the engine casts during COPY INTO / INSERT INTO. This protects the load against minor source schema changes (extra column on day 2, type widening, …) and is the main schema-drift resilience knob.
• Alter target table if source has new columns (default false). Enable to have Etlworks ALTER TABLE ADD COLUMNS when the source has columns the destination doesn't.

Also available: Recreate target table, Insert null into fields that don't exist in source.

Step 10. Configure mapping — optional

If source and destination columns don't match by name, configure per-field mapping.

Incremental change replication (high watermark)

To extract only changed rows on each run, set the High Watermark Field on the transformation and pick a watermark type (timestamp or monotonic sequence). For details, see Change replication using High Watermark.

For real-time replication from a transaction log, use Stream CDC events into Databricks instead of HWM.

Load multiple tables with a wildcard

One transformation can load many tables when the source is a database.

FROM

Enter a wildcard source name — for example public.* for all tables in the source schema public.

TO

Use the asterisk to map source names to destination names. Example: analytics.public.* maps source test.dbo.patient to destination analytics.public.patient.

Include / exclude objects

Use Include objects and Exclude objects to filter the wildcard match. See Source-to-destination transformation for the full syntax.

MERGE with wildcards

For per-table lookup fields, enter table=fields pairs separated by ;:

analytics.public.inventory = inventory_id, database_name;
analytics.public.payment   = payment_id, database_name;
analytics.public.rental    = rental_id, database_name;

Or enable Predict Lookup Fields to let the flow infer keys per table.

HWM with wildcards

HWM works alongside wildcard sources; the watermark value is tracked per table.

Troubleshooting

For common load issues, see Common issues when loading data into cloud data warehouses.