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

CSV

Excel

JSON

ORC

Parquet

文本

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

目录列表

文件通知

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

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

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

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

特定于云

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

AWS

仅当选择 cloudFiles.useNotifications = true 并且希望自动加载程序为你设置通知服务时，才提供以下选项：

仅当你选择 cloudFiles.useNotifications = true 并希望自动加载程序使用已设置的队列时，才提供以下选项：

AWS 身份验证选项

提供以下身份验证选项以使用 Databricks 服务凭据：

当 Databricks 服务凭据或 IAM 角色不可用时，可以改为提供以下身份验证选项：

蔚蓝

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

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

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

GCP

自动加载程序可以利用 Databricks 服务凭据自动为你设置通知服务。 使用 Databricks 服务凭据创建的服务帐户将需要 在文件通知模式下配置自动加载程序流中指定的权限。

如果 Databricks 服务凭据不可用，可以直接使用 Google 服务帐户。 您可以通过按照 Google 服务设置 的步骤将群集配置为使用服务帐户，或者直接提供以下身份验证选项：

仅当你选择 cloudFiles.useNotifications = true 并希望自动加载程序使用已设置的队列时，才提供以下选项：

Delta Lake

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

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

Delta Lake 和 Apache Iceberg

CSV

Excel

JSON

ORC

Parquet

文本

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