Referência de opções de API do Spark

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

Opções do DataFrameReader

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

Example

O exemplo a seguir define multiLine para ler True arquivos 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 obter a lista completa de opções de leitor do Kafka, consulte as opções do Kafka do DataStreamReader. As opções a seguir se aplicam somente às leituras em lote usando spark.read.format("kafka").

ORC

Parquet

Repositório de estado

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

Texto

XML

Opções do DataStreamReader

Use essas opções para DataStreamReader.option() configurar leituras de streaming de tabelas delta lake e outras fontes baseadas em arquivo.

Para obter opções de formato de arquivo (JSON, CSV, Parquet e outros), consulte as opções DataFrameReader.

Para opções do Carregador Automático (cloudFiles.*), consulte o Carregador Automático.

Example

O exemplo a seguir define maxFilesPerTrigger para 10 um fluxo de tabela 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")

Comum

As opções a seguir se aplicam a tabelas delta lake e outras fontes de streaming baseadas em arquivo.

Carregador Automático

Use essas opções com a origem para configurar o cloudFilesCarregador Automático para ingestão de streaming do armazenamento em nuvem. As opções específicas à origem cloudFiles são prefixadas cloudFiles para mantê-las em um namespace separado de outras opções de origem de Streaming Estruturado .

Comum

Listagem de diretório

Notificação de arquivo

Para obter informações sobre como configurar o modo de notificação de arquivo, incluindo permissões de nuvem necessárias, instruções de instalação e métodos de autenticação, consulte Configurar fluxos do Carregador Automático no modo de notificação de arquivo.

(1) O Carregador automático adiciona os seguintes pares de marca de chave-valor por padrão com base no melhor esforço:

• vendor: Databricks
• path: Local de onde os dados são carregados. Não disponível em GCP devido à limitações de rotulagem.
• checkpointLocation: a posição do ponto de controle do fluxo. Não disponível em GCP devido à limitações de rotulagem.
• streamId: Identificador exclusivo global do fluxo.

O Databricks reserva esses nomes de chave e você não pode substituir seus valores.

Específico da nuvem

O Carregador Automático fornece opções para configurar a infraestrutura de nuvem para o modo de notificação de arquivo. Para obter as permissões de nuvem necessárias e as instruções de instalação, consulte Configurar fluxos do Carregador Automático no modo de notificação de arquivo.

AWS

Forneça as seguintes opções somente se você escolher cloudFiles.useNotifications = true e quiser que o Carregador Automático configure os serviços de notificação para você:

Forneça a opção a seguir somente se você escolher cloudFiles.useNotifications = true e desejar que o Carregador automático use uma fila já configurada:

Opções de autenticação do AWS

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

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

Azure

Você deverá fornecer valores para todas as opções a seguir se especificar cloudFiles.useNotifications = true e quiser que o Carregador automático configure os serviços de notificação para você:

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

Forneça a seguinte opção somente se você definir cloudFiles.useNotifications = true e quiser que o Carregador Automático use uma fila existente:

GCP

O Carregador Automático pode configurar automaticamente os serviços de notificação para você aproveitando as credenciais de serviço do Databricks . A conta de serviço criada com a credencial de serviço do 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 do Databricks não estiver disponível, você poderá usar as Contas de Serviço do Google diretamente. Você pode configurar seu cluster para assumir uma conta de serviço seguindo a configuração do serviço Google ou fornecer as seguintes opções de autenticação diretamente:

Forneça a opção a seguir somente se você escolher cloudFiles.useNotifications = true e desejar que o Carregador automático use uma fila já configurada:

Lago Delta

As opções a seguir se aplicam ao ler de uma tabela delta lake usando spark.readStream.

Kafka

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

As opções a seguir se aplicam somente às leituras de streaming com spark.readStream.format("kafka"):

Para obter opções de deslocamento que se aplicam somente às leituras em lote com spark.read.format("kafka"), consulte as opções do Kafka do DataFrameReader.

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

Opções de DataFrameWriter

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

Example

O exemplo a seguir define mergeSchema para escrever True uma tabela 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 e Apache Iceberg

CSV

Excel

JSON

ORC

Parquet

Texto

XML

Opções do DataStreamWriter

Use essas opções para DataStreamWriter.option() configurar gravações de streaming.

Example

O exemplo a seguir define o local de ponto de verificação 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

Coletor de console

Lago Delta

As opções a seguir se aplicam ao escrever um fluxo em uma tabela delta lake usando format("delta"). Opções somente substituição, como overwriteSchema, replaceWheree partitionOverwriteMode não têm suporte para gravações de streaming.

Coletor de arquivos

A opção a seguir se aplica ao gravar um fluxo em formatos baseados em arquivo (Parquet, JSON, CSV, ORC, texto). Para obter opções específicas de formato, consulte as opções do DataFrameWriter.

Coletor Kafka

Para obter uma lista completa de opções para gravar fluxos no Kafka, consulte Opções.

Coletor de memória

Opções de função do Spark

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

Example

O exemplo a seguir analisa uma coluna JSON ao descartar registros 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 do Avro aceitam as mesmas opções que as opções de DataFrame correspondentes:

• from_avro e schema_of_avro use as opções do DataFrameReader Avro.
• to_avro usa as opções do DataFrameWriter Avro.

Example

O exemplo a seguir decodifica uma coluna do Avro com a evolução do esquema habilitada:

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 Registro de Esquema e from_avroto_avro as seguintes opções são aceitas:

CSV

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

• from_csv e schema_of_csv use as opções de CSV do DataFrameReader.
• to_csv usa opções de CSV do DataFrameWriter.

Example

O exemplo a seguir lê 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 de DataFrame correspondentes:

• from_json e schema_of_json use as opções JSON do DataFrameReader.
• to_json usa opções JSON do DataFrameWriter.

Example

O exemplo a seguir grava JSON com NULL campos ignorados e formatação bastante habilitada:

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 use um DataSource baseado em arquivo. Os dados do Protobuf são sempre lidos e gravados como colunas binárias usando essas funções. As opções são passadas como um Map[String, String] e diferenciam maiúsculas de minúsculas.

Example

O exemplo a seguir decodifica uma coluna Protobuf usando o modo 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")))

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

XML

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

• from_xml e schema_of_xml use as opções XML do DataFrameReader.
• to_xml usa opções XML do DataFrameWriter.

Example

O exemplo a seguir grava XML com marcas de linha e raiz 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")))