Connectors

A connector is chosen by the format field of an input, an output, or a join/union source. The same registry serves all three, and it is extensible — see Adding your own.

Format	Read	Write	Notes
`parquet`	✓	✓	Spark native
`delta`	✓	✓	MERGE upserts, time travel
`iceberg`	✓	✓	MERGE INTO
`csv`	✓	✓	`header` and `inferSchema` default to true; RFC 4180 quoting
`txt`	✓	✓	plain text, single `value` column
`view`	✓	✓	Spark temp views, auto-cached
`kafka`	✓	✓	batch read and publish; Amazon MSK
`postgresql`	✓	✓	JDBC
`mysql`	✓	✓	JDBC
`mariadb`	✓	✓	JDBC
`sqlserver`	✓	✓	JDBC
`oracle`	✓	✓	JDBC (service name)
`bigquery`	✓	✓	Google BigQuery
`snowflake`	✓	✓	Snowflake
`redshift`	✓	✓	Amazon Redshift, S3 staging
`mongodb`	✓	✓	MongoDB
`documentdb`	✓	✓	Amazon DocumentDB (Mongo protocol)
`dynamodb`	✓	✓	Amazon DynamoDB, write is upsert
`cassandra`	✓	✓	Cassandra / ScyllaDB, write is upsert
`elasticsearch`	✓	✓	Elasticsearch / OpenSearch

Files and lakehouse

Parquet

{ "format": "parquet", "path": "/data/curated/orders", "mode": "overwrite", "partition_by": ["country"] }

Any Spark Parquet option passes through options. path is a directory of part files, not a single file.

CSV

{ "format": "csv", "path": "/data/landing/orders", "options": { "sep": ";", "encoding": "UTF-8" } }

Defaults applied by the framework: header: true and inferSchema: true on read, header: true on write, encoding: UTF-8 and escape: " on both. inferSchema costs an extra pass over the data — set an explicit cast and turn it off for large, stable feeds.

Quoting is RFC 4180: a quote inside a field is written doubled (""), not backslash-escaped (\") as Spark does by default. Spark reads its own dialect back, but Python’s csv, pandas and Excel do not — which used to break the validations.report, whose rule_params column is quote-heavy JSON, in the very tools it is analysed in. To read a file written in the old dialect, ask for it: "options": { "escape": "\\" }.

Delta Lake

// read a table, or a path, or a past version
{ "format": "delta", "path": "catalog.schema.orders" }
{ "format": "delta", "path": "/mnt/raw/orders", "options": { "versionAsOf": "5" } }
{ "format": "delta", "path": "catalog.schema.orders", "options": { "timestampAsOf": "2026-01-01T00:00:00Z" } }

// upsert
{ "format": "delta", "path": "catalog.schema.orders", "mode": "merge",
  "options": { "merge_keys": ["order_id"], "merge_condition": "T.deleted = false" } }

path is treated as a table name when it contains a dot and does not start with / or a storage scheme; otherwise as a path. In merge_condition, T is the target table and S the incoming DataFrame.

Outside Databricks, install the OSS package: pip install "sparquet[delta]".

Iceberg

{ "format": "iceberg", "path": "catalog.db.orders", "mode": "merge", "options": { "merge_keys": ["id"] } }

Text

{ "format": "txt", "path": "/data/exports/lines", "mode": "overwrite" }

Reads into a single value column, and writes a DataFrame that must have exactly one string column — project it first with select.

Temp views

{ "format": "view", "path": "orders_staging", "mode": "overwrite" }

Registers a Spark temp view, cached by default, so a later pipeline in the same session can read it as {"format": "view", "path": "orders_staging"}. This is the staging mechanism behind multi-pipeline jobs — one writes the view, the next consumes it.

options.scope chooses the lifetime: session (default) is a normal temp view, visible only to the current session; global uses createOrReplaceGlobalTempView, visible to every session of the same Spark application (read it as global_temp.<name>, or with scope: "global" on the reader).

Kafka

Batch read and batch publish. path is the topic. Requires spark-sql-kafka-0-10 on the classpath.

// read the whole topic
{
  "format": "kafka",
  "path": "orders-events",
  "options": { "bootstrap_servers": "broker1:9092,broker2:9092", "startingOffsets": "earliest" }
}

// publish
{
  "format": "kafka",
  "path": "orders-events",
  "mode": "append",
  "transformations": [
    { "type": "with_column", "column": "value", "expression": "to_json(payload)" }
  ],
  "options": {
    "bootstrap_servers": "broker1:9092,broker2:9092",
    "value_column": "value",
    "key_column": "order_id"
  }
}

On read, the DataFrame comes back with Kafka’s native schema (binary key/value, topic, partition, offset, timestamp, timestampType) — usually followed by a CAST(value AS STRING). The batch defaults are startingOffsets: earliest and endingOffsets: latest, so a run reads the whole topic; override them, or pass assign / subscribePattern instead of the default subscribe.

On write, the configured value_column (default payload) and key_column (default header, null to omit) are renamed to value and key, and every column outside {key, value, topic, partition, timestamp, headers} is dropped before the write.

bootstrap_servers is a friendly alias for kafka.bootstrap.servers. For Amazon MSK with IAM auth, pass kafka.security.protocol=SASL_SSL and kafka.sasl.mechanism=AWS_MSK_IAM in options (plus the aws-msk-iam-auth JAR on the classpath).

JSON

Native, no extra JAR. path is a directory; the default is one object per line (JSON Lines). Set multiLine: "true" to read one document per file.

{ "format": "json", "path": "/data/landing/events", "options": { "multiLine": "true" } }

ORC

Native columnar format (like Parquet), no extra JAR. Options include compression (default zlib) and mergeSchema.

{ "format": "orc", "path": "/data/curated/orders", "mode": "overwrite", "partition_by": ["dt_ref"] }

Avro

Row-oriented; requires the org.apache.spark:spark-avro package on the classpath. Options: avroSchema, compression (snappy/deflate/bzip2/xz), recordName.

{ "format": "avro", "path": "/data/raw/events" }

XML

Requires the com.databricks:spark-xml package (registers the xml format). rowTag is required on read and write; rootTag (default rows) names the write root.

{ "format": "xml", "path": "/data/raw/catalog", "options": { "rowTag": "book" } }

Binary files

Read-only (binaryFile). Loads whole files into columns path, modificationTime, length, content (binary) — images, PDFs, blobs. There is no binary writer; persist content via parquet/delta.

{ "format": "binary", "path": "/data/raw/docs", "options": { "pathGlobFilter": "*.pdf" } }

Apache Hudi

Lakehouse table format with upserts. Requires the hudi-spark-bundle JAR and the Hudi session extensions. path is the table base path; partitioning and upsert are driven by hoodie.* options (the framework partition_by is not used).

{
  "format": "hudi",
  "path": "s3a://lake/hudi/orders",
  "mode": "append",
  "options": {
    "hoodie.table.name": "orders",
    "hoodie.datasource.write.recordkey.field": "order_id",
    "hoodie.datasource.write.precombine.field": "updated_at",
    "hoodie.datasource.write.operation": "upsert"
  }
}

Relational databases (JDBC)

postgresql, mysql, mariadb, sqlserver and oracle share one JDBC base with per-database dialects that fill in the driver class, default port and URL shape. path is the table name (dbtable).

// read a table
{
  "format": "postgresql",
  "path": "public.orders",
  "options": {
    "host": "db.internal", "port": 5432, "database": "sales",
    "user": "reader", "password": "…"
  }
}

// or give the full URL and a pushdown query
{
  "format": "sqlserver",
  "path": "dbo.orders",
  "options": {
    "url": "jdbc:sqlserver://db:1433;databaseName=sales",
    "query": "SELECT id, total FROM dbo.orders WHERE total > 0",
    "user": "reader", "password": "…"
  }
}

Connection options: url (full JDBC URL, takes precedence) or host + database (+ optional port, defaulted per database) to build it; driver (defaulted per database); user / password. Reads also accept query (a SELECT used instead of dbtable), dbtable (overrides path), partitionColumn / lowerBound / upperBound / numPartitions for parallel reads, and fetchsize. Writes support append and overwrite (truncate: "true" does a TRUNCATE instead of DROP/CREATE), plus batchsize, isolationLevel, createTableColumnTypes / createTableOptions. JDBC has no merge — use append/overwrite.

Pushdown

Spark can hand parts of the query to the database instead of dragging rows across the network to filter them in the JVM. All five levers default to true on the Spark 4 read path and are passed straight through:

Option	What goes to the database
`pushDownPredicate`	the `WHERE` clause
`pushDownAggregate`	`sum`/`count`/`min`/`max`/`avg` and the `GROUP BY`
`pushDownLimit`	the `LIMIT`
`pushDownOffset`	the `OFFSET`
`pushDownTableSample`	`TABLESAMPLE`

{ "format": "postgresql", "path": "public.orders",
  "options": { "host": "db.internal", "database": "shop",
               "pushDownAggregate": "true", "pushDownLimit": "true" } }

The constraint that matters: they only apply when reading a table. With query the SELECT you wrote is already the cut, so Spark wraps it as a subquery and pushes nothing further into it. Aggregate pushdown also needs the aggregation to be the only thing above the scan — a filter on a computed column, a join or a UDF in between and the database sees a plain SELECT. Turning a lever on is a request, not a fact: { "type": "debug", "actions": ["pushdown"] } says what the plan actually pushed. See debug.

MariaDB speaks MySQL

mariadb builds a jdbc:mysql:// URL with the MySQL driver, on purpose. Spark 4.1 has no MariaDB dialect and its MySQLDialect only matches URLs starting with jdbc:mysql; with jdbc:mariadb:// the connector falls back to the default dialect, which quotes identifiers with " — and MariaDB rejects that on reads as much as on writes, since the SELECT Spark builds quotes the columns the same way:

You have an error in your SQL syntax ... near '"id" INTEGER'

MariaDB speaks the MySQL protocol, so Connector/J reaches a MariaDB server and brings the dialect with it: backtick quoting, MySQL type mapping and correct SQL in aggregate and LIMIT pushdown. To use the MariaDB driver anyway, pass url and driver explicitly together with sessionVariables:

{ "format": "mariadb", "path": "orders",
  "options": { "url": "jdbc:mariadb://db.internal:3306/shop?sessionVariables=sql_mode='ANSI_QUOTES'",
               "driver": "org.mariadb.jdbc.Driver", "user": "app", "password": "..." } }

The price is that "..." stops being a string literal in that session, which matters to anyone using query. An explicit jdbc:mariadb:// URL without ANSI_QUOTES warns.

For Oracle, database is the service name (jdbc:oracle:thin:@//host:port/service).

Data warehouses

BigQuery

path is project.dataset.table (or dataset.table with a default project). Uses the spark-bigquery-connector.

{ "format": "bigquery", "path": "my-proj.sales.orders",
  "options": { "parentProject": "billing-proj", "credentialsFile": "/secrets/sa.json" } }

Read options: query (requires viewsEnabled: "true"), parentProject (billing), credentialsFile / credentials, filter, maxParallelism. Write (overwrite / append): temporaryGcsBucket (indirect staging, default), writeMethod (indirect or direct), partitionField / partitionType / clusteredFields.

Snowflake

path is the table (dbtable). Uses spark-snowflake + snowflake-jdbc.

{ "format": "snowflake", "path": "ANALYTICS.PUBLIC.ORDERS",
  "options": {
    "sfUrl": "myorg-acct.snowflakecomputing.com",
    "sfUser": "loader", "sfPassword": "…",
    "sfDatabase": "ANALYTICS", "sfSchema": "PUBLIC", "sfWarehouse": "LOAD_WH"
  } }

Connection is the sfXxx family (sfUrl, sfUser / sfPassword or pem_private_key, sfDatabase, sfSchema, sfWarehouse, sfRole). Reads accept query; writes support overwrite / append.

Redshift

path is the table (dbtable). Uses the spark-redshift community connector, which stages through S3.

{ "format": "redshift", "path": "public.orders",
  "options": {
    "url": "jdbc:redshift://cluster:5439/sales",
    "tempdir": "s3://my-bucket/redshift-staging/",
    "aws_iam_role": "arn:aws:iam::…:role/redshift-copy"
  } }

url and tempdir (an S3 prefix) are required. Auth is user / password or aws_iam_role; forward_spark_s3_credentials: "true" reuses the session’s S3 credentials. Writes support overwrite / append and diststyle / distkey / sortkeyspec / tempformat.

NoSQL and search

MongoDB and DocumentDB

path is the collection. The same Mongo Spark Connector serves Amazon DocumentDB — point connection.uri at the DocumentDB cluster with ?tls=true&retryWrites=false.

{ "format": "mongodb", "path": "orders",
  "options": { "connection.uri": "mongodb://user:pass@host:27017/", "database": "sales" } }

Options: connection.uri (required), database, collection (overrides path), aggregation.pipeline (read pushdown). Writes support overwrite / append, plus operationType (insert / replace / update), idFieldList, ordered.

DynamoDB

path is the table (tableName). Uses spark-dynamodb.

{ "format": "dynamodb", "path": "orders", "options": { "region": "us-east-1" } }

Options: region, roleArn, endpoint (e.g. DynamoDB local), throughput / targetCapacity, readPartitions / stronglyConsistentReads. Writes are always a PutItem per row — an upsert by primary key (append); DynamoDB has no table overwrite.

Cassandra and ScyllaDB

path is keyspace.table (or just the table, with keyspace in options). Uses the spark-cassandra-connector.

{ "format": "cassandra", "path": "sales.orders",
  "options": { "spark.cassandra.connection.host": "node1,node2" } }

Options: keyspace / table (override path), spark.cassandra.connection.host / .port, spark.cassandra.auth.username / .password. Writes are append (INSERT/upsert by key); the table must already exist.

Elasticsearch

path is the index (resource). Uses the elasticsearch-hadoop connector — option prefix es.*.

{ "format": "elasticsearch", "path": "orders",
  "options": { "es.nodes": "es.internal", "es.port": "9200" } }

Options: es.nodes / es.port, es.net.http.auth.user / .pass, es.nodes.wan.only (cloud/proxy), es.query (read DSL), es.mapping.id (use a column as _id), es.write.operation (index / create / update / upsert). Writes support append / overwrite.

elasticsearch-spark-30_2.13 is still compiled against Spark 3.4.3 up to and including 9.5.3, and there is no elasticsearch-spark-40 artifact. On Spark 4 the read may work and the write dies:

java.lang.NoSuchMethodError: 'org.apache.spark.sql.SQLContext org.apache.spark.sql.Dataset.sqlContext()'

Point the OpenSearch connector at the Elasticsearch server instead — same REST API, verified against Elasticsearch 8.16.1. Migrating takes exactly two edits, neither skippable:

format becomes opensearch. The jar does not register es: [DATA_SOURCE_NOT_FOUND] Failed to find the data source: es.
every es.* option becomes opensearch.*. The old prefix is silently ignored and the write dies with OpenSearchHadoopIllegalArgumentException.

If the native connector is a hard requirement, the alternatives are running that pipeline on Spark 3.5, writing to Parquet and indexing with Logstash or a bulk-API job, or the Elasticsearch SQL/JDBC bridge for reads only.

OpenSearch

A separate format with its own connector (opensearch-hadoop) — option prefix opensearch.*, not es.*. Same shape as Elasticsearch otherwise.

{ "format": "opensearch", "path": "orders",
  "options": { "opensearch.nodes": "os.internal", "opensearch.port": "9200" } }

Options mirror the ES ones with the opensearch. prefix: opensearch.nodes / opensearch.port, opensearch.net.http.auth.user / .pass, opensearch.nodes.wan.only, opensearch.query, opensearch.mapping.id, opensearch.write.operation.

On Spark 4 the connector is org.opensearch.client:opensearch-spark-40_2.13:2.0.0; on Spark 3.x it is opensearch-spark-30_2.12. This is also the supported route to an Elasticsearch server — see the caution above.

Driver JARs on the classpath

Every connector above ships as a Spark package that must be on the classpath. Declare it per pipeline so the job carries its own dependency:

{
  "spark": {
    "configs": {
      "spark.jars.packages": "org.postgresql:postgresql:42.7.3,org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.1"
    }
  }
}

On a managed platform you can also add the package to the cluster libraries. The framework never bundles these — a missing JAR surfaces as a ClassNotFoundException / Failed to find data source at read or write time.

Adding your own

fw.register_reader("my_format", MyReader)   # class MyReader(BaseReader)
fw.register_writer("my_format", MyWriter)   # class MyWriter(BaseWriter)

See Extending. Studio preserves unknown formats through import and export, so a custom connector does not break the canvas.