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

CSV

Excel

JSON

ORC

Parquet

Text

XML

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

Directory listing

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.

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

Cloud-specific

Auto Loader provides 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.

AWS

Provide the following options only if you choose cloudFiles.useNotifications = true and you want Auto Loader to set up the notification services for you:

Provide the following option only if you choose cloudFiles.useNotifications = true and you want Auto Loader to use a queue that you have already set up:

AWS authentication options

Provide the following authentication option to use a Databricks service credential:

When Databricks service credentials or IAM roles are not available, you can provide the following authentication options instead:

Azure

You must provide 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 provide the following authentication options instead:

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

GCP

Auto Loader can automatically set up notification services for you by leveraging Databricks service credentials. The service account created with the Databricks service credential will require the permissions specified in Configure Auto Loader streams in file notification mode.

If a Databricks service credential is not available, you can use Google Service Accounts directly. You can either configure your cluster to assume a service account by following Google service setup or provide the following authentication options directly:

Provide the following option only if you choose cloudFiles.useNotifications = true and you want Auto Loader to use a queue that you have already set up:

Delta Lake

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

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

Delta Lake and Apache Iceberg

CSV

Excel

JSON

ORC

Parquet

Text

XML

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

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

For a complete list of options for writing streams to Kafka, see Options.

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