Справочник по параметрам API Spark

На этой странице перечислены доступные параметры входных и выходных данных для API Spark, которые считывают и записывают данные.

Параметры DataFrameReader

Используйте эти параметры с DataFrameReader.option(), DataFrameReader.options(), read_files, COPY INTO и Auto Loader для управления тем, как Azure Databricks считывает файлы данных.

Example

В следующем примере задано multiLine значение True для чтения 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)

Общие

Следующие параметры применяются ко всем форматам файлов.

Avro

CSV

Excel

JSON

Кафка

Полный список параметров чтения Kafka см. в параметрах DataStreamReader Kafka. Следующие параметры применяются только к пакетным считываниям с помощью spark.read.format("kafka").

ОРК

Паркетным

Хранилище состояний

Используйте эти параметры с функцией с read_statestore табличным значением для spark.read.format("statestore") чтения данных о состоянии структурированной потоковой передачи. См. Сведения о состоянии структурированной потоковой передачи.

Текст

XML

Параметры DataStreamReader

Используйте эти параметры для DataStreamReader.option() настройки потоковых операций чтения из таблиц Delta Lake и других источников на основе файлов.

Параметры формата файлов (JSON, CSV, Parquet и другие) см. в параметрах DataFrameReader.

Параметры автозагрузчика (cloudFiles.*) см. в разделе "Автозагрузчик".

Пример

Следующий пример задает значение 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")

Общие

Следующие параметры применяются к таблицам Delta Lake и другим источникам потоковой передачи на основе файлов.

Автозагрузчик

Используйте эти параметры с cloudFiles источником, чтобы настроить автозагрузчик для приема потоковой передачи из облачного хранилища. Параметры, относящиеся к cloudFiles источнику, имеют префикс, cloudFiles чтобы сохранить их в отдельном пространстве имен от других параметров источника структурированной потоковой передачи .

Общие

Список каталогов

Уведомление о файле

Сведения о настройке режима уведомлений файлов, включая необходимые облачные разрешения, инструкции по настройке и методы проверки подлинности, см. в разделе "Настройка потоков автозагрузчика" в режиме уведомлений о файлах.

(1) Автозагрузчик добавляет следующие пары тегов "ключ-значение" по умолчанию, по мере возможности:

• vendor: Databricks
• path: расположение, из которого загружаются данные. Недоступно в GCP из-за ограничений по маркировке.
• checkpointLocation: расположение контрольной точки потока. Недоступно в GCP из-за ограничений по маркировке.
• streamId: глобальный уникальный идентификатор для потока.

Databricks резервирует эти имена ключей, и вы не можете перезаписать их значения.

Облачные решения

Автозагрузчик предоставляет параметры настройки облачной инфраструктуры для режима уведомлений файлов. Необходимые облачные разрешения и инструкции по настройке см. в разделе "Настройка потоков автозагрузчика" в режиме уведомлений о файлах.

AWS

Укажите следующие параметры, только если вы выберете cloudFiles.useNotifications = true и хотите, чтобы автозагрузчик настраивал службы уведомлений для вас:

Укажите следующий параметр, только если вы выбрали cloudFiles.useNotifications = true и хотите, чтобы Автозагрузчик использовал уже настроенную вами очередь.

Параметры проверки подлинности AWS

Укажите следующий параметр проверки подлинности для использования учетных данных службы Databricks:

Если учетные данные службы Databricks или роли IAM недоступны, можно указать следующие параметры проверки подлинности.

Azure

Необходимо указать значения для всех следующих параметров, если вы указали cloudFiles.useNotifications = true и хотите, чтобы Автозагрузчик настроил службы уведомлений:

Если учетные данные службы Databricks недоступны, можно указать следующие параметры проверки подлинности.

Укажите следующий параметр, только если задано cloudFiles.useNotifications = true , и вы хотите, чтобы автозагрузчик использовал существующую очередь:

GCP

Автозагрузчик может автоматически настроить службы уведомлений для вас, используя учетные данные службы Databricks . Для учетной записи службы, созданной с учетными данными службы Databricks, требуются разрешения, указанные в разделе "Настройка потоков автозагрузчика" в режиме уведомлений о файлах.

Если учетные данные службы Databricks недоступны, вы можете напрямую использовать учетные записи службы Google. Вы можете настроить кластер для получения учетной записи службы, выполнив настройку службы Google , или укажите следующие параметры проверки подлинности напрямую:

Укажите следующий параметр, только если вы выбрали cloudFiles.useNotifications = true и хотите, чтобы Автозагрузчик использовал уже настроенную вами очередь.

Delta Lake

Следующие параметры применяются при чтении из таблицы Delta Lake с помощью spark.readStream.

Кафка

Используйте следующие параметры с помощью следующих spark.readStream.format("kafka") вариантов:spark.read.format("kafka")

Следующие параметры применяются только к потоковым чтениям со spark.readStream.format("kafka")следующими параметрами:

Параметры смещения, которые применяются только к пакетным считываниям, spark.read.format("kafka")см. в параметрах Kafka DataFrameReader.

Параметры клиента Kafka (kafka.*) и проверки подлинности см. в разделе "Параметры".

Параметры DataFrameWriter

Используйте эти параметры с DataFrameWriter.option() и DataFrameWriterV2.option() для управления Azure Databricks записи данных.

Example

В следующем примере задано mergeSchema значение True для записи таблицы 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

ОРК

Паркетным

Текст

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

Общие

Приемник консоли

Delta Lake

Следующие параметры применяются при написании потока в таблицу Delta Lake с помощью format("delta"). Параметры, доступные только для перезаписи, например overwriteSchema, replaceWhereи partitionOverwriteMode не поддерживаются для потоковой записи.

Приемник файлов

Следующий параметр применяется при написании потока в форматы на основе файлов (Parquet, JSON, CSV, ORC, текст). Дополнительные сведения о параметрах формата см. в разделе "Параметры 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_avroпараметры DataFrameReader 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_avro реестра схем и to_avro принимают следующие параметры:

CSV

Функции CSV принимают те же параметры, что и соответствующие параметры кадра данных:

• from_csv используйте schema_of_csvпараметры CSV DataFrameReader.
• to_csv использует параметры CSV-файла DataFrameWriter.

Example

Следующий пример считывает CSV-файл с пользовательским разделителем и NULL значением:

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 принимают те же параметры, что и соответствующие параметры кадра данных:

• from_json используйте schema_of_jsonпараметры JSON DataFrameReader.
• to_json использует параметры JSON DataFrameWriter.

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 не используйте файловый источник Данных. Данные Protobuf всегда считываются и записываются как двоичные столбцы с помощью этих функций. Параметры передаются в виде Map[String, String] регистра.

Example

В следующем примере декодирует столбец Protobuf с помощью режима PERMISSIVE:

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 принимают те же параметры, что и соответствующие параметры кадра данных:

• from_xml используйте schema_of_xmlпараметры XML DataFrameReader.
• to_xml использует параметры XML DataFrameWriter.

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