Spark API 选项参考

此页面列出了用于读取和写入数据的 Spark API 的可用输入和输出选项。

DataFrameReader 选项

将这些选项用于 DataFrameReader.option()、DataFrameReader.options()、read_files、COPY INTO 和 Auto Loader 来控制Azure Databricks读取数据文件的方式。

Example

以下示例设置为multiLineTrue读取 JSON 文件：

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

以下选项适用于所有文件格式。

Avro

读取 Avro 文件时，以下选项适用。

CSV

读取 CSV 文件时，以下选项适用。

Excel

读取Excel文件时，以下选项适用。

JSON

读取 JSON 文件时，以下选项适用。

Kafka

有关 Kafka 读取器选项的完整列表，请参阅 DataStreamReader Kafka 选项。 以下选项仅适用于使用 spark.read.format("kafka").

ORC

读取 ORC 文件时，以下选项适用。

Parquet

读取 Parquet 文件时，以下选项适用。

状态存储

将这些选项用于 spark.read.format("statestore") 或 read_statestore 表值函数来读取结构化流式处理状态数据。 请参阅读取结构化流式处理状态信息。

文本

读取文本文件时，以下选项适用。

XML

读取 XML 文件时，以下选项适用。

DataStreamReader 选项

使用这些选项 DataStreamReader.option() 配置 Delta Lake 表和其他基于文件的源的流式读取。

有关文件格式选项（JSON、CSV、Parquet 等），请参阅 DataFrameReader 选项。

有关自动加载程序（cloudFiles.*）选项，请参阅 自动加载程序。

Example

以下示例设置为 maxFilesPerTrigger10 Delta Lake 表流：

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

以下选项适用于 Delta Lake 表和其他基于文件的流式处理源。

自动加载器

将这些选项与源配合使用 cloudFiles ，为从云存储进行流式引入配置 自动加载程序 。 特定于 cloudFiles 源的选项带有 cloudFiles 前缀，以将它们保留在与其他 结构化流 源选项不同的命名空间中。

Common

以下选项适用于所有自动加载程序配置。

目录列表

使用目录列表模式时，适用以下选项。

文件通知

有关配置文件通知模式（包括所需的云权限、设置说明和身份验证方法）的信息，请参阅 在文件通知模式下配置自动加载程序流。

云提供商资源标记

默认情况下，自动加载程序会尽力添加以下键值标记对：

• vendor: Databricks
• path：从中加载数据的位置。 由于标签限制，在 GCP 中不可用。
• checkpointLocation：流检查点的位置。 由于标签限制，在 GCP 中不可用。
• streamId：流的全局唯一标识符。

Databricks 保留这些密钥名称，并且无法覆盖其值。

有关 Azure 的详细信息，请参阅命名队列和元数据以及properties.labels中的覆盖范围。 自动加载程序将这些键值标记对以 JSON 格式存储为标签。

特定于云

自动加载程序提供了用于为文件通知模式配置云基础结构的选项。 有关所需的云权限和设置说明，请参阅 在文件通知模式下配置自动加载程序流。

蔚蓝

如果指定 cloudFiles.useNotifications = true 并且希望自动加载程序为你设置通知服务，则必须为以下所有选项指定值：

如果 Databricks 服务凭据不可用，可以改为指定以下身份验证选项：

仅当设置 cloudFiles.useNotifications = true 并且希望自动加载程序使用现有队列时，才指定以下选项：

Delta Lake

使用以下 spark.readStream选项从 Delta Lake 表读取时适用。

Kafka

将这些选项与以下选项一起使用： spark.readStream.format("kafka") 或 spark.read.format("kafka")：

以下选项仅适用于流式读取：spark.readStream.format("kafka")

有关仅适用于批处理读取的 spark.read.format("kafka")偏移选项，请参阅 DataFrameReader Kafka 选项。

身份验证

Databricks 建议使用 Unity 目录服务凭据向云托管的 Kafka 服务（AWS MSK、Azure 事件中心 或 Google Cloud Managed Kafka）进行身份验证。

当服务凭据不可用时，请使用 SASL/SSL 选项（作为 kafka.* 属性传递）。 使用服务凭据时，无需指定kafka.sasl.mechanism或kafka.sasl.jaas.configkafka.security.protocol指定。

有关完整的身份验证设置说明，请参阅 “身份验证”。

Pub/Sub

使用这些选项订阅 spark.readStream.format("pubsub") Google Pub/Sub。 选项subscriptionIdtopicId和projectId必需选项。

Pulsar

使用这些选项 spark.readStream.format("pulsar") 从 Apache Pulsar 流式传输。 在 Databricks Runtime 14.1 及更高版本中可用。

需要以下选项。 必须准确指定其中一个topic或 topicstopicsPattern。

还支持以下选项：

可以使用以下选项模式指定其他 Pulsar 客户端、管理员和读取器配置：

有关 Pulsar 客户端和管理员身份验证选项的详细信息，请参阅 身份验证。

身份验证

Azure Databricks 支持向 Pulsar 进行信任存储和密钥存储身份验证。 Azure Databricks建议使用机密来存储身份验证详细信息。 请参阅机密管理。

如果流使用 a PulsarAdmin，还可以设置以下选项：

有关身份验证示例，请参阅 “向 Pulsar 进行身份验证”。

DataFrameWriter 选项

将这些选项用于 DataFrameWriter.option() 和 DataFrameWriterV2.option()来控制Azure Databricks写入数据的方式。

Example

以下示例设置为mergeSchemaTrue写入 Delta Lake 表：

Python

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

Scala

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

Avro

编写 Avro 文件时，以下选项适用。

Delta Lake 和 Apache Iceberg

编写 Delta Lake 和 Apache Iceberg 表时，以下选项适用。

CSV

编写 CSV 文件时，以下选项适用。

Excel

编写Excel文件时，以下选项适用。

JSON

编写 JSON 文件时，以下选项适用。

ORC

编写 ORC 文件时，以下选项适用。

Parquet

编写 Parquet 文件时，以下选项适用。

文本

编写文本文件时，以下选项适用。

XML

编写 XML 文件时，以下选项适用。

DataStreamWriter 选项

使用这些选项 DataStreamWriter.option() 来配置流式写入。

Example

以下示例设置流的检查点位置：

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

以下选项适用于所有流式写入操作。

控制台接收器

将流写入控制台接收器时，以下选项适用。

Delta Lake

使用 format("delta") 将流写入 Delta Lake 表时，以下选项适用。 仅覆盖选项，例如overwriteSchemareplaceWhere，不支持partitionOverwriteMode流式写入。

文件存储端

将流写入基于文件的格式（Parquet、JSON、CSV、ORC、text）时，适用以下选项。 有关特定于格式的选项，请参阅 DataFrameWriter 选项。

Kafka 接收器

写入 Kafka 时，以下选项适用。

内存接收器

将流写入内存接收器时，以下选项适用。

Spark 函数选项

某些 Spark SQL 内置函数接受控制 options 分析或序列化行为的映射。 将选项作为 Python dict 或 Scala Map[String, String] 传递。

Example

以下示例在删除格式不正确的记录时分析 JSON 列：

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 函数接受与相应的数据帧选项相同的选项：

• from_avro 并使用 schema_of_avroDataFrameReader Avro 选项。
• to_avro 使用 DataFrameWriter Avro 选项。

Example

以下示例解码启用了架构演变的 Avro 列：

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

此外，架构注册表变体和from_avroto_avro接受以下选项：

CSV

CSV 函数接受与相应的数据帧选项相同的选项：

• from_csv 并使用 schema_of_csvDataFrameReader CSV 选项。
• to_csv 使用 DataFrameWriter CSV 选项。

Example

以下示例使用自定义分隔符和 NULL 值读取 CSV：

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 函数接受与相应的 DataFrame 选项相同的选项：

• from_json 并使用 schema_of_jsonDataFrameReader JSON 选项。
• to_json 使用 DataFrameWriter JSON 选项。

Example

以下示例编写 JSON，其中 NULL 字段被忽略并启用了漂亮的格式设置：

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 不使用 to_protobuf 基于文件的 DataSource。 Protobuf 数据始终使用这些函数读取和写入为二进制列。 选项以区分 Map[String, String] 大小写的形式传递。

Example

以下示例使用 PERMISSIVE 模式解码 Protobuf 列：

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 函数使用以下选项：

XML

XML 函数接受与相应的 DataFrame 选项相同的选项：

• from_xml 并使用 schema_of_xmlDataFrameReader XML 选项。
• to_xml 使用 DataFrameWriter XML 选项。

Example

以下示例使用自定义根标记和行标记编写 XML：

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