Databricks Writer examples

What we're trying to achieve: build an initial-load pipeline that reads the CUSTOMERS and ORDERS tables from Oracle and writes them to Databricks on AWS, using a Personal Access Token for authentication and S3 for staging, with the data loaded to the target as inserts (Append Only mode).

CREATE OR REPLACE APPLICATION databricks_il_pat;
CREATE OR REPLACE SOURCE DBSource USING Global.DatabaseReader ( 
  Tables: 'COMPANY.CUSTOMERS;COMPANY.ORDERS',
  QuiesceOnILCompletion: true, 
  adapterName: 'DatabaseReader', 
  Password_encrypted: 'true', 
  connectionProfileName: 'OracleConnectionProfile', 
  DatabaseProviderType: 'Oracle', 
  ParallelThreads: 1, 
  RestartBehaviourOnILInterruption: 'keepTargetTableData', 
  useConnectionProfile: true, 
  Query: '', 
  FetchSize: 10000, 
  CreateSchema: true)
OUTPUT TO DBOut;

CREATE OR REPLACE TARGET DatabricksTarget USING Global.DeltaLakeWriter ( 
  authenticationType: 'PersonalAccessToken',
  useConnectionProfile: false, 
  connectionUrl: "jdbc:databricks://myworkspace.cloud.databricks.com:443",
  personalAccessToken: "${vault.databricks_pat}" 
  adapterName: 'DeltaLakeWriter', 
  ConnectionRetryPolicy: 'initialRetryDelay=10s, retryDelayMultiplier=2, maxRetryDelay=1m, maxAttempts=5, totalTimeout=10m', 
  Mode: 'APPENDONLY', 
  Tables: 'COMPANY.CUSTOMERS,my_catalog.company.customers;COMPANY.ORDERS,my_catalog.company.orders', 
  optimizedMerge: false, 
  stageLocation: '/', 
  externalStageType: 'S3', 
  s3AccessKey: '${vault.s3accesskey}',
  s3SecretAccessKey: '${vault.s3secretaccesskey}',
  s3BucketName: 'striim-databricks-stage',
  s3Region: 'us-east1',
  uploadPolicy: 'eventcount:100000,interval:60s') 
INPUT FROM DBOut;

END APPLICATION databricks_il_pat;

What we're trying to achieve: continuously capture changes from Postgres and merge them into Databricks on Google Cloud, so the target mirrors current source state (Merge mode), using Service Principal (M2M OAuth) authentication and GCS for staging.

CREATE APPLICATION databricks_cdc_sp_gcs
CREATE SOURCE PostgresCDCSource
  USING PostgresReader(
    ConnectionURL: 'jdbc:postgresql://localhost:5432/mydb',
    Username: 'postgres',
    Password: '${vault.postgres_pwd}',
    Tables: 'POSTGRES_SOURCE.ACCOUNTS;POSTGRES_SOURCE.TRANSACTIONS',
    Mode: 'CDC',
    LogicalDecodingSlotName: 'striim_slot'
  )
  OUTPUT TO PostgresStream;

CREATE OR REPLACE TARGET DatabricksTarget USING Global.DeltaLakeWriter ( 
  authenticationType: 'ServicePrincipal',
  useConnectionProfile: false, 
  connectionUrl: "jdbc:databricks://myworkspace.cloud.databricks.com:443",
  ClientId: "${vault.databricks_client_id}",
  ClientSecret: "${vault.databricks_client_secret}",
  adapterName: 'DeltaLakeWriter', 
  ConnectionRetryPolicy: 'initialRetryDelay=10s, retryDelayMultiplier=2, maxRetryDelay=1m, maxAttempts=5, totalTimeout=10m', 
  Mode: 'MERGE', 
  Tables: 'POSTGRES_SOURCE.ACCOUNTS,my_catalog.my_schema.accounts;POSTGRES_SOURCE.TRANSACTIONS,my_catalog.my_schema.transactions',
  optimizedMerge: false, 
  stageLocation: '/', 
  externalStageType: 'GCS', 
  storageCredentialName: "gcs-striim-staging",
  gcsBucketName: "my-staging-bucket",
  gcsBucketRegion: "us-central1",
  gcsProjectId: "my-gcp-project",
  gcsServiceAccountKey: "/striim/config/gcs-key.json"
  uploadPolicy: 'eventcount:100000,interval:60s') 
INPUT FROM PostgresStream;

END APPLICATION databricks_cdc_sp_gcs;

What we're trying to achieve: write to Azure Databricks using Microsoft Entra ID authentication, with Azure Data Lake Storage Gen2 as the staging area, loading data to the target as inserts (Append Only mode).

CREATE OR REPLACE TARGET db USING Global.DeltaLakeWriter (
  useConnectionProfile: true,
  connectionProfileName: 'admin.DatabricksEntraIDCP',
  stageLocation: '/',
  CDDLAction: 'Process',
  ConnectionRetryPolicy: 'initialRetryDelay=10s, retryDelayMultiplier=2, maxRetryDelay=1m, maxAttempts=5, totalTimeout=10m',
  Mode: 'APPENDONLY',
  externalStageType: 'ADLSGen2',
  Tables: 'public.sample_pk,sample_catalog.sample_db.sample_pk',
  externalStageConnectionProfileName: admin.ADLSGen2CP'',
  uploadPolicy: 'eventcount:10000,interval:60s')
INPUT FROM DBOut;

What we're trying to achieve: continuously capture changes from SQL Server and merge them into Databricks on Google Cloud, so the target mirrors current source state (Merge mode), using saved Connection Profiles for both Databricks authentication and GCS staging — a cleaner approach for repeated deployments than specifying credentials inline.

CREATE APPLICATION databricks_with_profile
CREATE SOURCE SQLServerCDCSource
  USING MSSqlReader(
    useConnectionProfile: true,
    connectionProfileName: 'admin.SQLServerConnectionProfile'
    Tables: 'SQL_SERVER_SOURCE.%',
    Mode: 'CDC',
    AutoDisableTableCDC: false,
    FetchTransactionMetadata: true,
    TransactionSupport: true
  )
  OUTPUT TO SQLServerStream;

CREATE OR REPLACE TARGET DatabricksTarget USING Global.DeltaLakeWriter ( 
  optimizedMerge: false, 
  useConnectionProfile: true, 
  CDDLAction: 'Process', 
  adapterName: 'DeltaLakeWriter', 
  ConnectionRetryPolicy: 'initialRetryDelay=10s, retryDelayMultiplier=2, maxRetryDelay=1m, maxAttempts=5, totalTimeout=10m', 
  Mode: 'MERGE', 
  connectionProfileName: 'admin.Databricks_Connection_Profile', 
  Tables: 'SQL_SERVER_SOURCE.%,my_catalog.my_schema.%', 
  externalStageType: 'GCS', 
  externalStageConnectionProfileName: 'admin.GCS_Connection_Profile',
  storageCredentialName: 'gcsstagecredential', 
  gcsBucketName: 'striim-deltalake-bucket',
  gcsBucketRegion: 'asia-south1',
  uploadPolicy: 'eventcount:100000,interval:60s') 
INPUT FROM SQLServerStream;

END APPLICATION databricks_with_profile;