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

Quando si leggono i file Avro, si applicano le opzioni seguenti.

CSV

Quando si leggono i file CSV, si applicano le opzioni seguenti.

Excel

Quando si legge Excel file, si applicano le opzioni seguenti.

JSON

Quando si leggono i file JSON, si applicano le opzioni seguenti.

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

Durante la lettura dei file ORC si applicano le opzioni seguenti.

Parquet

Quando si leggono i file Parquet, si applicano le opzioni seguenti.

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

Durante la lettura dei file di testo si applicano le opzioni seguenti.

XML (Extensible Markup Language)

Durante la lettura dei file XML si applicano le opzioni seguenti.

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

Le opzioni seguenti si applicano a tutte le configurazioni del caricatore automatico.

Directory

L'opzione seguente si applica quando si usa la modalità elenco 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.

Tag delle risorse del provider di servizi cloud

Il caricatore automatico aggiunge per impostazione predefinita le coppie di tag chiave-valore seguenti per impostazione predefinita:

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

Per altre informazioni su Azure, vedere Denominazione di code e metadati e la trattazione di properties.labels nelle sottoscrizioni di eventi. Il caricatore automatico archivia queste coppie di tag chiave-valore in JSON come etichette.

Specifico del cloud

Il caricatore automatico include 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.

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 specificare 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:

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.

Autenticazione

Databricks consiglia di usare una credenziale del servizio Catalogo Unity per l'autenticazione ai servizi Kafka gestiti dal cloud (AWS MSK, Hub eventi di Azure o Google Cloud Managed Kafka).

Quando una credenziale del servizio non è disponibile, usare le opzioni SASL/SSL (passate come kafka.* proprietà). Quando si usano credenziali del servizio, non è necessario specificare kafka.sasl.mechanism, kafka.sasl.jaas.configo kafka.security.protocol.

Per istruzioni complete sull'installazione dell'autenticazione, vedere Autenticazione.

Pub/Sub

Usa queste opzioni con spark.readStream.format("pubsub") per sottoscrivere Google Pub/Sub. Sono necessarie le opzioni subscriptionId, topicIde projectId .

Per altre informazioni su Pub/Sub, vedi Iscriviti a Google Pub/Sub.

Pulsar

Usare queste opzioni con spark.readStream.format("pulsar") per eseguire lo streaming da Apache Pulsar. Disponibile in Databricks Runtime 14.1 e versioni successive.

Sono necessarie le opzioni seguenti. È necessario specificare esattamente uno di topic, topicso topicsPattern.

Sono supportate anche le opzioni seguenti:

È possibile specificare configurazioni aggiuntive di client, amministratore e lettore Pulsar usando i modelli di opzione seguenti:

Per altre informazioni sulle opzioni di autenticazione del client Pulsar e dell'amministratore, vedere Autenticazione.

Autenticazione

Azure Databricks supporta l'autenticazione tramite truststore e keystore per Pulsar. Azure Databricks consiglia di usare i segreti per archiviare i dettagli di autenticazione. Vedere Gestione dei segreti.

Se il flusso usa un PulsarAdmin, è anche possibile impostare le opzioni seguenti:

Per esempi di autenticazione, vedere Eseguire l'autenticazione a Pulsar.

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

Quando si scrivono file Avro, si applicano le opzioni seguenti.

Delta Lake e Apache Iceberg

Quando si scrivono tabelle Delta Lake e Apache Iceberg, si applicano le opzioni seguenti.

CSV

Quando si scrivono file CSV, si applicano le opzioni seguenti.

Excel

Quando si scrivono file di Excel, si applicano le opzioni seguenti.

JSON

Quando si scrivono file JSON, si applicano le opzioni seguenti.

ORCO

Quando si scrivono file ORC, si applicano le opzioni seguenti.

Parquet

Quando si scrivono file Parquet, si applicano le opzioni seguenti.

Testo

Quando si scrivono file di testo, si applicano le opzioni seguenti.

XML (Extensible Markup Language)

Quando si scrivono file XML, si applicano le opzioni seguenti.

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

Le opzioni seguenti si applicano a tutte le operazioni di scrittura di streaming.

Sink della console

Quando si scrivono flussi nel sink della console, si applicano le opzioni seguenti.

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

Quando si scrive in Kafka, si applicano le opzioni seguenti.

Sink di memoria

Quando si scrivono flussi nel sink di memoria, si applicano le opzioni seguenti.

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 (Extensible Markup Language)

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