Справочник по параметрам 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

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

CSV

При чтении CSV-файлов применяются следующие параметры.

Excel

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

JSON

При чтении JSON-файлов применяются следующие параметры.

Кафка

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

ОРК

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

Паркетным

При чтении файлов 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")

Общие

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

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

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

Общие

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

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

Следующий параметр применяется при использовании режима перечисления каталогов.

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

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

Теги ресурсов поставщика облачных служб

Автозагрузчик добавляет следующие пары тегов "ключ-значение" по умолчанию на основе наилучших усилий:

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

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

Дополнительные сведения об Azure см. в разделе Именование очередей и метаданных и охват properties.labels в подписках на события. Автозагрузчик сохраняет эти пары тегов "ключ — значение" в формате JSON в виде меток.

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

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

Azure

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

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

Укажите следующий параметр, только если задано 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.

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

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. Параметры subscriptionIdи topicIdprojectId обязательные.

Дополнительные сведения о pub/Sub см. в разделе "Подписка на Google Pub/Sub".

Пульсар

Используйте эти параметры для spark.readStream.format("pulsar") потоковой передачи из Apache Pulsar. Доступно в Databricks Runtime 14.1 и более поздних версиях.

Необходимы следующие параметры. Необходимо указать именно один из topic, topicsили topicsPattern.

Также поддерживаются следующие параметры:

Вы можете указать дополнительные конфигурации клиента, администратора и читателя Pulsar, используя следующие шаблоны параметров:

Дополнительные сведения о параметрах проверки подлинности клиента и администратора Pulsar см. в разделе "Проверка подлинности".

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

Azure Databricks поддерживает аутентификацию для Pulsar с использованием хранилища доверенных сертификатов и хранилища ключей. Azure Databricks рекомендует использовать секреты для хранения сведений о проверке подлинности. Дополнительные сведения см. в разделе Управление секретами.

Если в потоке используется, PulsarAdminможно также задать следующие параметры:

Примеры проверки подлинности см. в разделе "Проверка подлинности в Pulsar".

Параметры 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

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

Delta Lake и Apache Iceberg

Следующие параметры применяются при написании таблиц Delta Lake и Apache Iceberg.

CSV

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

Excel

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

JSON

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

ОРК

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

Паркетным

При написании файлов 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")

Общие

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

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

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

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