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

CSV

Excel

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

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

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

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.

(1) O Auto Loader adiciona por padrão os seguintes pares de etiquetas chave-valor, de acordo com melhores esforços:

• 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.

Específico da cloud

O Auto Loader oferece 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.

AWS

Forneça as seguintes opções apenas se quiser cloudFiles.useNotifications = true e quiser que o Auto Loader configure os serviços de notificação por si:

Forneça a seguinte opção somente se você escolher cloudFiles.useNotifications = true e quiser que o Auto Loader use uma fila que você já configurou:

Opções de autenticação AWS

Forneça a seguinte opção de autenticação para usar uma credencial de serviço Databricks:

Quando as credenciais de serviço Databricks ou funções do IAM não estão disponíveis, você pode fornecer as seguintes opções de autenticação:

Azure

Você deve fornecer 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 para você:

Se uma credencial de serviço Databricks não estiver disponível, você poderá fornecer as seguintes opções de autenticação:

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

GCP

O Auto Loader pode configurar automaticamente os serviços de notificação para si, ao utilizar as credenciais de serviço do Databricks (). A conta de serviço criada com a credencial de serviço Databricks exigirá as permissões especificadas em Configurar fluxos do Carregador Automático no modo de notificação de arquivo.

Se uma credencial de serviço Databricks não estiver disponível, você poderá usar as Contas de serviço do Google diretamente. Você pode configurar o seu cluster para adotar uma conta de serviço seguindo a configuração de serviço do Google ou fornecer as seguintes opções de autenticação diretamente:

Forneça a seguinte opção somente se você escolher cloudFiles.useNotifications = true e quiser que o Auto Loader use uma fila que você já configurou:

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.

Para o cliente Kafka (kafka.*) e opções de autenticação, veja Opções.

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

Lago Delta e Iceberg Apache

CSV

Excel

JSON

ORC

Parquet

Texto

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

Lava-solas

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

Para uma lista completa de opções para escrever streams para Kafka, veja Opções.

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