Referência de opções da API Spark

Esta página lista opções de entrada e saída disponíveis para APIs Spark que leem e escrevem dados.

Opções DataFrameReader

Use estas opções com DataFrameReader.option(), DataFrameReader.options(), read_files, COPY INTO e Auto Loader para controlar como Azure Databricks lê ficheiros de dados.

Example

O exemplo seguinte é multiLineTrue para ler ficheiros 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)

Comum

As opções a seguir se aplicam a todos os formatos de arquivo.

Avro

As seguintes opções aplicam-se ao ler ficheiros Avro.

CSV

As seguintes opções aplicam-se ao ler ficheiros CSV.

Excel

As seguintes opções aplicam-se ao ler ficheiros Excel.

JSON

As seguintes opções aplicam-se ao ler ficheiros JSON.

Kafka

Para a lista completa de opções de leitores Kafka, consulte DataStreamReader Opções Kafka. As seguintes opções aplicam-se apenas a leituras em lote que usam spark.read.format("kafka").

ORC

As seguintes opções aplicam-se ao ler ficheiros ORC.

Parquet

As seguintes opções aplicam-se ao ler ficheiros Parquet.

Loja estatal

Use estas opções com spark.read.format("statestore") ou a read_statestore função de valores de tabela para ler dados de estado de Streaming Estruturado. Consulte Ler informações de estado do Structured Streaming.

Texto

As seguintes opções aplicam-se ao ler ficheiros de texto.

XML

As seguintes opções aplicam-se ao ler ficheiros XML.

Opções DataStreamReader

Use estas opções para DataStreamReader.option() configurar leituras em streaming a partir de tabelas Delta Lake e outras fontes baseadas em ficheiros.

Para opções de formatos de ficheiro (JSON, CSV, Parquet e outros), consulte as opções DataFrameReader.

Para opções de Auto Loader (cloudFiles.*), veja Auto Loader.

Example

O exemplo seguinte define maxFilesPerTrigger para 10 um curso de água de mesa do Lago Delta:

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

Comum

As seguintes opções aplicam-se às tabelas Delta Lake e a outras fontes de streaming baseadas em ficheiros.

Carregador Automático

Use estas opções com a cloudFiles fonte para configurar o Auto Loader para streaming de ingestão a partir do armazenamento na cloud. As opções específicas da cloudFiles fonte são prefixadas com cloudFiles para as manter num espaço de nomes separado das outras opções de fonte de Streaming Estruturado .

Comum

As seguintes opções aplicam-se a todas as configurações do Auto Loader.

Listagem de diretórios

A seguinte opção aplica-se ao usar o modo de listagem de diretórios.

Notificação de ficheiro

Para informações sobre a configuração do modo de notificação de ficheiros, incluindo permissões necessárias na cloud, instruções de configuração e métodos de autenticação, consulte Configurar fluxos do Auto Loader em modo de notificação de ficheiros.

Etiquetas de recursos do fornecedor de cloud

O Auto Loader adiciona os seguintes pares de etiquetas-chave-valor por defeito, numa base de melhor esforço:

• vendor: Databricks
• path: O local de onde os dados são carregados. Indisponível no GCP devido a limitações de rotulagem.
• checkpointLocation: A localização do ponto de verificação do fluxo. Indisponível no GCP devido a limitações de rotulagem.
• streamId: Um identificador global exclusivo para o fluxo.

O Databricks reserva estes nomes-chave, e não se pode sobrescrever os seus valores.

Para obter mais informações sobre o Azure, consulte Nomeação de filas e metadados e a cobertura de properties.labels em Assinaturas de Eventos. O Auto Loader armazena esses pares de tags chave-valor em JSON como rótulos.

Específico da cloud

O Auto Loader tem opções para configurar a infraestrutura cloud para o modo de notificação de ficheiros. Para permissões necessárias na cloud e instruções de configuração, consulte Configurar fluxos do Auto Loader em modo de notificação de ficheiros.

Azure

Deve especificar valores para todas as seguintes opções se especificar cloudFiles.useNotifications = true e quiser que o Auto Loader configure os serviços de notificação por si:

Se uma credencial de serviço Databricks não estiver disponível, pode especificar as seguintes opções de autenticação em vez disso:

Especifique a seguinte opção apenas se definir cloudFiles.useNotifications = true e quiser que o Auto Loader use uma fila existente:

Lago Delta

As seguintes opções aplicam-se ao ler a partir de uma tabela Delta Lake usando spark.readStream.

Kafka

Use estas opções com um ou spark.readStream.format("kafka") outro spark.read.format("kafka"):

As seguintes opções aplicam-se apenas a leituras em streaming com spark.readStream.format("kafka"):

Para opções de deslocamento que se aplicam apenas a leituras em lote com spark.read.format("kafka"), veja DataFrameReader opções Kafka.

Autenticação

A Databricks recomenda usar uma credencial de serviço Unity Catalog para autenticar serviços Kafka geridos na cloud (AWS MSK, Hubs de Eventos do Azure ou Google Cloud Managed Kafka).

Quando uma credencial de serviço não estiver disponível, utilize opções SASL/SSL (passadas como kafka.* propriedades). Quando utiliza uma credencial de serviço, não precisa de especificar kafka.sasl.mechanism, kafka.sasl.jaas.config, ou kafka.security.protocol.

Para instruções completas de configuração da autenticação, consulte Autenticação.

Pub/Sub

Use estas opções para spark.readStream.format("pubsub") subscrever o Google Pub/Sub. As opções subscriptionId, topicId, e projectId são obrigatórias.

Para mais informações sobre Pub/Sub, consulte Subscrever o Google Pub/Sub.

Pulsar

Use estas opções para spark.readStream.format("pulsar") transmitir a partir do Apache Pulsar. Disponível no Databricks Runtime 14.1 e superiores.

São necessárias as seguintes opções. Deve especificar exatamente um de topic, topics, ou topicsPattern.

As seguintes opções também são suportadas:

Pode especificar configurações adicionais de cliente, administrador e leitor Pulsar usando os seguintes padrões de opções:

Para mais informações sobre as opções de autenticação do cliente e administrador Pulsar, consulte Autenticação.

Autenticação

O Azure Databricks suporta autenticação com truststore e keystore no Pulsar. O Azure Databricks recomenda usar segredos para armazenar detalhes de autenticação. Consulte Gestão secreta.

Se o stream usar um PulsarAdmin, também pode definir as seguintes opções:

Para exemplos de autenticação, veja Autenticar ao Pulsar.

Opções DataFrameWriter

Use estas opções com DataFrameWriter.option() e DataFrameWriterV2.option() para controlar como Azure Databricks escreve dados.

Example

O exemplo seguinte define mergeSchema para True escrever uma tabela de 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

As seguintes opções aplicam-se ao escrever ficheiros Avro.

Lago Delta e Iceberg Apache

As seguintes opções aplicam-se ao escrever tabelas Delta Lake e Apache Iceberg.

CSV

As seguintes opções aplicam-se ao escrever ficheiros CSV.

Excel

As seguintes opções aplicam-se ao escrever ficheiros Excel.

JSON

As seguintes opções aplicam-se ao escrever ficheiros JSON.

ORC

As seguintes opções aplicam-se ao escrever ficheiros ORC.

Parquet

As seguintes opções aplicam-se ao escrever ficheiros Parquet.

Texto

As seguintes opções aplicam-se ao escrever ficheiros de texto.

XML

As seguintes opções aplicam-se ao escrever ficheiros XML.

Opções DataStreamWriter

Use estas opções para DataStreamWriter.option() configurar escritas em streaming.

Example

O exemplo seguinte define a localização do ponto de controlo para um fluxo:

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

Comum

As seguintes opções aplicam-se a todas as operações de escrita em streaming.

Lava-solas

As seguintes opções aplicam-se ao escrever streams para o sink da consola.

Lago Delta

As seguintes opções aplicam-se ao escrever um fluxo para uma tabela de Delta Lake usando format("delta"). Opções apenas de sobrescrever, como overwriteSchema, replaceWhere, e partitionOverwriteMode não são suportadas para escritas em streaming.

Sumidouro de ficheiros

A seguinte opção aplica-se ao escrever um fluxo para formatos baseados em ficheiros (Parquet, JSON, CSV, ORC, texto). Para opções específicas de formato, consulte opções DataFrameWriter.

Afundamento de Kafka

As seguintes opções aplicam-se ao escrever para Kafka.

Sumidouro de memória

As seguintes opções aplicam-se ao escrever fluxos para o dissipador de memória.

Opções de função de faísca

Algumas funções integradas no SQL do Spark aceitam um options mapa que controla o comportamento de análise ou serialização. Passa opções como Python dict ou Scala Map[String, String].

Example

O exemplo seguinte analisa uma coluna JSON ao eliminar registos malformados:

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

As funções Avro aceitam as mesmas opções que as opções DataFrame correspondentes:

• from_avro e schema_of_avro utilizar as opções Avro do DataFrameReader.
• to_avro utiliza as opções Avro do DataFrameWriter.

Example

O exemplo seguinte decodifica uma coluna Avro com a evolução do esquema ativada:

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

Além disso, as variantes do Registo de Esquemas de from_avro e to_avro aceitam as seguintes opções:

CSV

As funções CSV aceitam as mesmas opções que as opções DataFrame correspondentes:

• from_csv e schema_of_csv utilizar opções CSV DataFrameReader.
• to_csv utiliza opções CSV do DataFrameWriter.

Example

O exemplo seguinte lê-se CSV com um separador e NULL valor personalizados:

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

As funções JSON aceitam as mesmas opções que as opções correspondentes de DataFrame:

• from_json e schema_of_json utilizar opções JSON do DataFrameReader.
• to_json utiliza opções JSON do DataFrameWriter.

Example

O exemplo seguinte escreve JSON com NULL campos ignorados e formatação bonita ativada:

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 e to_protobuf não utilize uma fonte de dados baseada em ficheiros. Os dados protobuf são sempre lidos e escritos como colunas binárias usando estas funções. As opções são aprovadas como a Map[String, String] e são sensíveis a maiúsculas e minúsculas.

Example

O exemplo seguinte decodifica uma coluna Protobuf usando o modo PERMISSIVO:

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

As funções protobuf utilizam as seguintes opções:

XML

As funções XML aceitam as mesmas opções que as opções DataFrame correspondentes:

• from_xml e schema_of_xml utilizar as opções XML do DataFrameReader.
• to_xml utiliza as opções XML do DataFrameWriter.

Example

O exemplo seguinte escreve XML com etiquetas raiz e linhas personalizadas:

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