Spark API options reference

This page lists available input and output options for Spark APIs that read and write data.

DataFrameReader options

Use these options with DataFrameReader.option(), DataFrameReader.options(), read_files, COPY INTO, and Auto Loader to control how Azure Databricks reads data files.

Example

The following example sets multiLine to True for reading JSON files:

Python

df = spark.read.format("json").option("multiLine", True).load("/path/to/data")

Scala

val df = spark.read.format("json").option("multiLine", "true").load("/path/to/data")

SQL

SELECT * FROM read_files("/path/to/data", format => "json", multiLine => true)

Common

The following options apply to all file formats.

Avro

The following options apply when reading Avro files.

CSV

The following options apply when reading CSV files.

Excel

The following options apply when reading Excel files.

JSON

The following options apply when reading JSON files.

Kafka

For the full list of Kafka reader options, see DataStreamReader Kafka options. The following options apply only to batch reads using spark.read.format("kafka").

ORC

The following options apply when reading ORC files.

Parquet

The following options apply when reading Parquet files.

State store

Use these options with spark.read.format("statestore") or the read_statestore table-valued function to read Structured Streaming state data. See Read Structured Streaming state information.

Text

The following options apply when reading text files.

XML

The following options apply when reading XML files.

DataStreamReader options

Use these options with DataStreamReader.option() to configure streaming reads from Delta Lake tables and other file-based sources.

For file format options (JSON, CSV, Parquet, and others), see DataFrameReader options.

For Auto Loader (cloudFiles.*) options, see Auto Loader.

Example

The following example sets maxFilesPerTrigger to 10 for a Delta Lake table stream:

Python

df = spark.readStream.format("delta").option("maxFilesPerTrigger", 10).load("/path/to/delta-table")

Scala

val df = spark.readStream.format("delta").option("maxFilesPerTrigger", "10").load("/path/to/delta-table")

Common

The following options apply to Delta Lake tables and other file-based streaming sources.

Auto Loader

Use these options with the cloudFiles source to configure Auto Loader for streaming ingestion from cloud storage. Options specific to the cloudFiles source are prefixed with cloudFiles to keep them in a separate namespace from other Structured Streaming source options.

Common

The following options apply to all Auto Loader configurations.

Directory listing

The following option applies when using directory listing mode.

File notification

For information about configuring file notification mode, including required cloud permissions, setup instructions, and authentication methods, see Configure Auto Loader streams in file notification mode.

Cloud provider resource tags

Auto Loader adds the following key-value tag pairs by default on a best-effort basis:

• vendor: Databricks
• path: The location from where the data is loaded. Unavailable in GCP due to labeling limitations.
• checkpointLocation: The location of the stream's checkpoint. Unavailable in GCP due to labeling limitations.
• streamId: A globally unique identifier for the stream.

Databricks reserves these key names, and you cannot overwrite their values.

For more information on Azure, see Naming Queues and Metadata and the coverage of properties.labels in Event Subscriptions. Auto Loader stores these key-value tag pairs in JSON as labels.

Cloud-specific

Auto Loader has options for configuring cloud infrastructure for file notification mode. For required cloud permissions and setup instructions, see Configure Auto Loader streams in file notification mode.

Azure

You must specify values for all of the following options if you specify cloudFiles.useNotifications = true and you want Auto Loader to set up the notification services for you:

If a Databricks service credential is not available, you can specify the following authentication options instead:

Specify the following option only if you set cloudFiles.useNotifications = true and you want Auto Loader to use an existing queue:

Delta Lake

The following options apply when reading from a Delta Lake table using spark.readStream.

Kafka

Use these options with either spark.readStream.format("kafka") or spark.read.format("kafka"):

The following options apply only to streaming reads with spark.readStream.format("kafka"):

For offset options that apply only to batch reads with spark.read.format("kafka"), see DataFrameReader Kafka options.

Authentication

Databricks recommends using a Unity Catalog service credential to authenticate to cloud-managed Kafka services (AWS MSK, Azure Event Hubs, or Google Cloud Managed Kafka).

When a service credential is not available, use SASL/SSL options (passed through as kafka.* properties). When you use a service credential, you don't need to specify kafka.sasl.mechanism, kafka.sasl.jaas.config, or kafka.security.protocol.

For complete authentication setup instructions, see Authentication.

Pub/Sub

Use these options with spark.readStream.format("pubsub") to subscribe to Google Pub/Sub. The options subscriptionId, topicId, and projectId are required.

For more information on Pub/Sub, see Subscribe to Google Pub/Sub.

Pulsar

Use these options with spark.readStream.format("pulsar") to stream from Apache Pulsar. Available in Databricks Runtime 14.1 and above.

The following options are required. You must specify exactly one of topic, topics, or topicsPattern.

The following options are also supported:

You can specify additional Pulsar client, admin, and reader configurations using the following option patterns:

For more information on Pulsar client and admin authentication options, see Authentication.

Authentication

Azure Databricks supports truststore and keystore authentication to Pulsar. Azure Databricks recommends using secrets to store authentication details. See Secret management.

If the stream uses a PulsarAdmin, you can also set the following options:

For authentication examples, see Authenticate to Pulsar.

DataFrameWriter options

Use these options with DataFrameWriter.option() and DataFrameWriterV2.option() to control how Azure Databricks writes data.

Example

The following example sets mergeSchema to True for writing a Delta Lake table:

Python

df.write.format("delta").option("mergeSchema", True).saveAsTable("my_table")

Scala

df.write.format("delta").option("mergeSchema", "true").saveAsTable("my_table")

Avro

The following options apply when writing Avro files.

Delta Lake and Apache Iceberg

The following options apply when writing Delta Lake and Apache Iceberg tables.

CSV

The following options apply when writing CSV files.

Excel

The following options apply when writing Excel files.

JSON

The following options apply when writing JSON files.

ORC

The following options apply when writing ORC files.

Parquet

The following options apply when writing Parquet files.

Text

The following options apply when writing text files.

XML

The following options apply when writing XML files.

DataStreamWriter options

Use these options with DataStreamWriter.option() to configure streaming writes.

Example

The following example sets the checkpoint location for a stream:

Python

(df.writeStream
  .format("delta")
  .option("checkpointLocation", "/path/to/checkpoint")
  .start("/path/to/table"))

Scala

df.writeStream
  .format("delta")
  .option("checkpointLocation", "/path/to/checkpoint")
  .start("/path/to/table")

Common

The following options apply to all streaming write operations.

Console sink

The following options apply when writing streams to the console sink.

Delta Lake

The following options apply when writing a stream to a Delta Lake table using format("delta"). Overwrite-only options such as overwriteSchema, replaceWhere, and partitionOverwriteMode are not supported for streaming writes.

File sink

The following option applies when writing a stream to file-based formats (Parquet, JSON, CSV, ORC, text). For format-specific options, see DataFrameWriter options.

Kafka sink

The following options apply when writing to Kafka.

Memory sink

The following options apply when writing streams to the memory sink.

Spark function options

Some Spark SQL built-in functions accept an options map that controls parsing or serialization behavior. Pass options as a Python dict or a Scala Map[String, String].

Example

The following example parses a JSON column while dropping malformed records:

Python

from pyspark.sql.functions import from_json
from pyspark.sql.types import StructType, StructField, StringType

schema = StructType([StructField("name", StringType())])
df = df.withColumn("parsed", from_json("json_col", schema, {"mode": "DROPMALFORMED"}))

Scala

import org.apache.spark.sql.functions.from_json
import org.apache.spark.sql.types._

val schema = StructType(Seq(StructField("name", StringType)))
val df = df.withColumn("parsed", from_json(col("json_col"), schema, Map("mode" -> "DROPMALFORMED")))

Avro

Avro functions accept the same options as the corresponding DataFrame options:

• from_avro and schema_of_avro use DataFrameReader Avro options.
• to_avro uses DataFrameWriter Avro options.

Example

The following example decodes an Avro column with schema evolution enabled:

Python

from pyspark.sql.functions import from_avro

df = df.withColumn("decoded", from_avro("avro_col", json_schema, {"avroSchemaEvolutionMode": "restart"}))

Scala

import org.apache.spark.sql.avro.functions.from_avro

val df = df.withColumn("decoded", from_avro(col("avro_col"), jsonSchema, Map("avroSchemaEvolutionMode" -> "restart")))

In addition, the Schema Registry variants of from_avro and to_avro accept the following options:

CSV

CSV functions accept the same options as the corresponding DataFrame options:

• from_csv and schema_of_csv use DataFrameReader CSV options.
• to_csv uses DataFrameWriter CSV options.

Example

The following example reads CSV with a custom separator and NULL value:

Python

from pyspark.sql.functions import from_csv
from pyspark.sql.types import StructType, StructField, IntegerType, StringType

schema = StructType([StructField("id", IntegerType()), StructField("name", StringType())])
df = df.withColumn("parsed", from_csv("csv_col", schema, {"sep": "|", "nullValue": "N/A"}))

Scala

import org.apache.spark.sql.functions.from_csv
import org.apache.spark.sql.types._

val schema = StructType(Seq(StructField("id", IntegerType), StructField("name", StringType)))
val df = df.withColumn("parsed", from_csv(col("csv_col"), schema, Map("sep" -> "|", "nullValue" -> "N/A")))

JSON

JSON functions accept the same options as the corresponding DataFrame options:

• from_json and schema_of_json use DataFrameReader JSON options.
• to_json uses DataFrameWriter JSON options.

Example

The following example writes JSON with NULL fields ignored and pretty formatting enabled:

Python

from pyspark.sql.functions import to_json

df = df.withColumn("json_str", to_json("struct_col", {"pretty": "true", "ignoreNullFields": "true"}))

Scala

import org.apache.spark.sql.functions.to_json

val df = df.withColumn("json_str", to_json(col("struct_col"), Map("pretty" -> "true", "ignoreNullFields" -> "true")))

Protobuf

from_protobuf and to_protobuf do not use a file-based DataSource. Protobuf data is always read and written as binary columns using these functions. Options are passed as a Map[String, String] and are case-sensitive.

Example

The following example decodes a Protobuf column using PERMISSIVE mode:

Python

from pyspark.sql.functions import from_protobuf

df = df.withColumn("decoded", from_protobuf("proto_col", "MyMessage", "/path/to/descriptor.desc",
    {"mode": "PERMISSIVE", "enums.as.ints": "true"}))

Scala

import org.apache.spark.sql.protobuf.functions.from_protobuf

val df = df.withColumn("decoded", from_protobuf(col("proto_col"), "MyMessage", "/path/to/descriptor.desc",
    Map("mode" -> "PERMISSIVE", "enums.as.ints" -> "true")))

Protobuf functions use the following options:

XML

XML functions accept the same options as the corresponding DataFrame options:

• from_xml and schema_of_xml use DataFrameReader XML options.
• to_xml uses DataFrameWriter XML options.

Example

The following example writes XML with custom root and row tags:

Python

from pyspark.sql.functions import to_xml

df = df.withColumn("xml_str", to_xml("struct_col", {"rootTag": "records", "rowTag": "record"}))

Scala

import org.apache.spark.sql.functions.to_xml

val df = df.withColumn("xml_str", to_xml(col("struct_col"), Map("rootTag" -> "records", "rowTag" -> "record")))