Informazioni di riferimento sulle opzioni dell'API Spark

Questa pagina elenca le opzioni di input e output disponibili per le API Spark che leggono e scrivono dati.

Opzioni DataFrameReader

Usare queste opzioni con DataFrameReader.option(), DataFrameReader.options(), read_files, COPY INTO e Auto Loader per controllare come Azure Databricks legge i file di dati.

Example

L'esempio seguente imposta su multiLineTrue per la lettura di file 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)

Comune

Le opzioni seguenti si applicano a tutti i formati di file.

Avro

CSV

Excel

JSON

Kafka

Per l'elenco completo delle opzioni del lettore Kafka, vedere Opzioni Kafka di DataStreamReader. Le opzioni seguenti si applicano solo alle letture batch tramite spark.read.format("kafka").

ORCO

Parquet

Archivio stati

Usare queste opzioni con spark.read.format("statestore") o la read_statestore funzione con valori di tabella per leggere i dati dello stato structured streaming. Vedi Leggi le informazioni sullo stato di Structured Streaming.

Testo

XML

Opzioni DataStreamReader

Usare queste opzioni con DataStreamReader.option() per configurare le letture di streaming dalle tabelle Delta Lake e da altre origini basate su file.

Per le opzioni di formato di file (JSON, CSV, Parquet e altri), vedere Opzioni DataFrameReader.

Per le opzioni del caricatore automatico (cloudFiles.*), vedere Caricamento automatico.

Example

L'esempio seguente imposta maxFilesPerTrigger su 10 per un flusso di tabelle 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")

Comune

Le opzioni seguenti si applicano alle tabelle Delta Lake e ad altre origini di streaming basate su file.

Caricatore automatico

Usare queste opzioni con l'origine cloudFiles per configurare il caricatore automatico per l'inserimento in streaming dall'archiviazione cloud. Le opzioni specifiche dell'origine cloudFiles sono precedute cloudFiles dal prefisso per mantenerle in uno spazio dei nomi separato da altre opzioni di origine Structured Streaming .

Comune

Directory

Notifica file

Per informazioni sulla configurazione della modalità di notifica file, incluse le autorizzazioni cloud necessarie, le istruzioni di installazione e i metodi di autenticazione, vedere Configurare i flussi del caricatore automatico in modalità di notifica file.

(1) Il caricatore automatico aggiunge per impostazione predefinita le seguenti coppie di tag chiave-valore su base migliorativa:

• vendor: Databricks
• path: percorso da cui vengono caricati i dati. Non disponibile in GCP a causa di limitazioni di etichettatura.
• checkpointLocation: La posizione del checkpoint dello stream. Non disponibile in GCP a causa di limitazioni di etichettatura.
• streamId: un identificatore univoco globale per il flusso.

Databricks riserva questi nomi di chiave e non è possibile sovrascriverne i valori.

Specifico del cloud

Il caricatore automatico offre opzioni per la configurazione dell'infrastruttura cloud per la modalità di notifica file. Per le autorizzazioni cloud necessarie e le istruzioni di configurazione, vedere Configurare flussi del caricatore automatico in modalità di notifica file.

AWS

Specificare le opzioni seguenti solo se si sceglie cloudFiles.useNotifications = true e si vuole che il caricatore automatico configuri automaticamente i servizi di notifica:

Specificare l'opzione seguente solo se si sceglie cloudFiles.useNotifications = true e si vuole che il caricatore automatico usi una coda già configurata:

Opzioni di autenticazione di AWS

Fornire l'opzione di autenticazione seguente per usare una credenziale del servizio Databricks:

Quando le credenziali del servizio Databricks o i ruoli IAM non sono disponibili, è invece possibile fornire le opzioni di autenticazione seguenti:

Azure

È necessario specificare i valori per tutte le opzioni seguenti se si specifica cloudFiles.useNotifications = true e si vuole che il caricatore automatico configuri automaticamente i servizi di notifica:

Se una credenziale del servizio Databricks non è disponibile, è possibile fornire invece le opzioni di autenticazione seguenti:

Specificare l'opzione seguente solo se si imposta cloudFiles.useNotifications = true e si vuole che il caricatore automatico usi una coda esistente:

GCP

Auto Loader può configurare automaticamente i servizi di notifica sfruttando le credenziali del servizio Databricks . L'account del servizio creato con le credenziali del servizio Databricks richiederà le autorizzazioni specificate in Configurare flussi del caricatore automatico in modalità di notifica file.

Se le credenziali di un servizio Databricks non sono disponibili, è possibile usare direttamente gli account del servizio Google. È possibile configurare il cluster per presupporre un account del servizio seguendo la configurazione del servizio Google oppure specificare direttamente le opzioni di autenticazione seguenti:

Specificare l'opzione seguente solo se si sceglie cloudFiles.useNotifications = true e si vuole che il caricatore automatico usi una coda già configurata:

Delta Lake

Quando si esegue la lettura da una tabella Delta Lake tramite , si applicano le opzioni seguenti.spark.readStream

Kafka

Usare queste opzioni con spark.readStream.format("kafka") o spark.read.format("kafka"):

Le opzioni seguenti si applicano solo alle letture in streaming con spark.readStream.format("kafka"):

Per le opzioni di offset applicabili solo alle letture batch con spark.read.format("kafka"), vedere Opzioni Kafka di DataFrameReader.

Per le opzioni del client Kafka (kafka.*) e dell'autenticazione, vedere Opzioni.

Opzioni DataFrameWriter

Usare queste opzioni con DataFrameWriter.option() e DataFrameWriterV2.option() per controllare come Azure Databricks scrive i dati.

Example

Nell'esempio seguente viene impostato su mergeSchemaTrue per la scrittura di una tabella 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

ORCO

Parquet

Testo

XML

Opzioni dataStreamWriter

Usare queste opzioni con DataStreamWriter.option() per configurare le scritture di streaming.

Example

Nell'esempio seguente viene impostata la posizione del checkpoint per un flusso:

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

Comune

Sink della console

Delta Lake

Quando si scrive un flusso in una tabella Delta Lake tramite , si applicano le opzioni seguenti.format("delta") Le opzioni di sola sovrascrittura, overwriteSchemaad esempio , replaceWheree partitionOverwriteMode non sono supportate per le scritture in streaming.

Sink di file

L'opzione seguente si applica quando si scrive un flusso in formati basati su file (Parquet, JSON, CSV, ORC, text). Per le opzioni specifiche del formato, vedere Opzioni DataFrameWriter.

Sink Kafka

Per un elenco completo delle opzioni per la scrittura di flussi in Kafka, vedere Opzioni.

Sink di memoria

Opzioni della funzione Spark

Alcune funzioni predefinite di Spark SQL accettano una options mappa che controlla il comportamento di analisi o serializzazione. Passare le opzioni come Python dict o scala Map[String, String].

Example

L'esempio seguente analizza una colonna JSON durante l'eliminazione di record in formato non valido:

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

Le funzioni Avro accettano le stesse opzioni delle opzioni del dataframe corrispondenti:

• from_avro e schema_of_avro usare le opzioni DataFrameReader Avro.
• to_avro usa le opzioni Avro dataFrameWriter.

Example

L'esempio seguente decodifica una colonna Avro con l'evoluzione dello schema abilitata:

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

Inoltre, le varianti del Registro schemi di from_avro e to_avro accettano le opzioni seguenti:

CSV

Le funzioni CSV accettano le stesse opzioni delle opzioni del dataframe corrispondenti:

• from_csv e schema_of_csv usare le opzioni CSV DataFrameReader.
• to_csv usa le opzioni CSV DataFrameWriter.

Example

L'esempio seguente legge CSV con un separatore personalizzato e NULL un valore:

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

Le funzioni JSON accettano le stesse opzioni delle opzioni del dataframe corrispondenti:

• from_json e schema_of_json usare le opzioni JSON di DataFrameReader.
• to_json usa le opzioni JSON DataFrameWriter.

Example

L'esempio seguente scrive JSON con NULL campi ignorati e piuttosto formattati abilitati:

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 non usano un'origine dati basata su file. I dati protobuf vengono sempre letti e scritti come colonne binarie usando queste funzioni. Le opzioni vengono passate come e Map[String, String] fanno distinzione tra maiuscole e minuscole.

Example

L'esempio seguente decodifica una colonna Protobuf usando la modalità 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")))

Le funzioni Protobuf usano le opzioni seguenti:

XML

Le funzioni XML accettano le stesse opzioni delle opzioni del dataframe corrispondenti:

• from_xml e schema_of_xml usare le opzioni XML dataframereader.
• to_xml utilizza le opzioni XML DataFrameWriter.

Example

Nell'esempio seguente viene scritto xml con tag radice e riga personalizzati:

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