Referenz zu Spark-API-Optionen

Auf dieser Seite sind die verfügbaren Eingabe- und Ausgabeoptionen für Spark-APIs aufgeführt, die Daten lesen und schreiben.

DataFrameReader-Optionen

Verwenden Sie diese Optionen mit DataFrameReader.option(), DataFrameReader.options(), read_files, COPY INTO und Auto Loader, um zu steuern, wie Azure Databricks Datendateien liest.

Example

Im folgenden Beispiel wird festgelegt multiLine , dass True JSON-Dateien gelesen werden:

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)

Gemeinsam

Die folgenden Optionen gelten für alle Dateiformate.

Avro

CSV

Excel

JSON

Kafka

Eine vollständige Liste der Kafka-Leseoptionen finden Sie unter DataStreamReader Kafka-Optionen. Die folgenden Optionen gelten nur für Batchlesevorgänge.spark.read.format("kafka")

ORC

Parkett

Statusspeicher

Verwenden Sie diese Optionen mit spark.read.format("statestore") oder der read_statestore Tabellenwertfunktion, um Strukturierte Streaming-Statusdaten zu lesen. Weitere Informationen finden Sie unter Lesen von strukturierten Streamingstatusinformationen.

Text

XML

DataStreamReader-Optionen

Verwenden Sie diese Optionen zum DataStreamReader.option() Konfigurieren von Streaming-Lesevorgängen aus Delta Lake-Tabellen und anderen dateibasierten Quellen.

Informationen zu Dateiformatoptionen (JSON, CSV, Parkett und andere) finden Sie unter DataFrameReader-Optionen.

Optionen für das automatische Laden (cloudFiles.*) finden Sie unter "Auto Loader".

Example

Im folgenden Beispiel wird für einen Delta Lake-Tabellendatenstrom festgelegt maxFilesPerTrigger10 :

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

Gemeinsam

Die folgenden Optionen gelten für Delta Lake-Tabellen und andere dateibasierte Streamingquellen.

AutoLadeprogramm

Verwenden Sie diese Optionen mit der cloudFiles Quelle, um das automatische Ladeprogramm für die Streamingaufnahme aus dem Cloudspeicher zu konfigurieren. Optionen, die für die cloudFiles Quelle spezifisch sind, werden präfixiert cloudFiles , um sie in einem separaten Namespace von anderen Optionen für strukturiertes Streaming zu behalten.

Gemeinsam

Verzeichnisauflistung

Dateibenachrichtigung

Informationen zum Konfigurieren des Dateibenachrichtigungsmodus, einschließlich der erforderlichen Cloudberechtigungen, Setupanweisungen und Authentifizierungsmethoden, finden Sie unter Konfigurieren von Datenströmen für das automatische Laden im Dateibenachrichtigungsmodus.

(1) Der Autoloader fügt standardmäßig die folgenden Schlüssel-Wert-Tagpaare nach bestem Bemühen hinzu:

• vendor: Databricks
• path: Der Speicherort, von dem aus die Daten geladen werden. In GCP aufgrund von Bezeichnungseinschränkungen nicht verfügbar.
• checkpointLocation: Die Position des Stream-Checkpoints. In GCP aufgrund von Bezeichnungseinschränkungen nicht verfügbar.
• streamId: Ein global eindeutiger Bezeichner für den Stream.

Databricks behält sich diese Schlüsselnamen vor, und Sie können deren Werte nicht überschreiben.

Cloudspezifisch

Auto Loader bietet Optionen zum Konfigurieren der Cloudinfrastruktur für den Dateibenachrichtigungsmodus. Erforderliche Cloudberechtigungen und Einrichtungsanweisungen finden Sie unter Konfigurieren von Datenströmen für das automatische Laden im Dateibenachrichtigungsmodus.

AWS

Geben Sie die folgenden Optionen nur an, wenn Sie das automatische Laden auswählen cloudFiles.useNotifications = true und möchten, dass die Benachrichtigungsdienste für Sie eingerichtet werden:

Geben Sie die folgende Option nur an, wenn Sie cloudFiles.useNotifications = true auswählen und möchten, dass der Autoloader eine Warteschlange verwendet, die Sie bereits eingerichtet haben:

AWS-Authentifizierungsoptionen

Geben Sie die folgende Authentifizierungsoption für die Verwendung von Databricks-Dienstanmeldeinformationen an:

Wenn Databricks-Dienstanmeldeinformationen oder IAM-Rollen nicht verfügbar sind, können Sie stattdessen die folgenden Authentifizierungsoptionen bereitstellen:

Azure

Sie müssen Werte für alle folgenden Optionen angeben, wenn Sie cloudFiles.useNotifications = true festlegen und möchten, dass der Autoloader die Benachrichtigungsdienste für Sie einrichtet:

Wenn keine Databricks-Dienstanmeldeinformationen verfügbar sind, können Sie stattdessen die folgenden Authentifizierungsoptionen bereitstellen:

Geben Sie die folgende Option nur an, wenn Sie das automatische Laden festlegen cloudFiles.useNotifications = true und möchten, dass eine vorhandene Warteschlange verwendet wird:

GCP

Auto Loader kann Benachrichtigungsdienste automatisch für Sie einrichten, indem Sie Databricks-Dienstanmeldeinformationen nutzen. Das mit den Databricks-Dienstanmeldeinformationen erstellte Dienstkonto erfordert die Berechtigungen, die im Konfigurationsmodus für Auto Loader Streams im Dateibenachrichtigungsmodus angegeben sind.

Wenn keine Databricks-Dienstanmeldeinformationen verfügbar sind, können Sie Google Service-Konten direkt verwenden. Sie können Entweder Ihren Cluster so konfigurieren, dass ein Dienstkonto angenommen wird, indem Sie die Einrichtung des Google-Diensts ausführen oder die folgenden Authentifizierungsoptionen direkt bereitstellen:

Geben Sie die folgende Option nur an, wenn Sie cloudFiles.useNotifications = true auswählen und möchten, dass der Autoloader eine Warteschlange verwendet, die Sie bereits eingerichtet haben:

Delta Lake

Die folgenden Optionen gelten beim Lesen aus einer Delta Lake-Tabelle mit spark.readStream.

Kafka

Verwenden Sie diese Optionen entweder oder spark.readStream.format("kafka")spark.read.format("kafka"):

Die folgenden Optionen gelten nur für Streaminglesevorgänge mit spark.readStream.format("kafka"):

Offsetoptionen, die nur für Batchlesevorgänge spark.read.format("kafka")gelten, finden Sie unter DataFrameReader Kafka-Optionen.

Informationen zu Kafka-Client (kafka.*) und Authentifizierungsoptionen finden Sie unter "Optionen".

DataFrameWriter-Optionen

Verwenden Sie diese Optionen mit DataFrameWriter.option() und DataFrameWriterV2.option(), um zu steuern, wie Azure Databricks Daten schreibt.

Example

Im folgenden Beispiel wird festgelegt mergeSchema , dass True eine Delta Lake-Tabelle geschrieben wird:

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 und Apache Iceberg

CSV

Excel

JSON

ORC

Parkett

Text

XML

DataStreamWriter-Optionen

Verwenden Sie diese Optionen zum DataStreamWriter.option() Konfigurieren von Streaming-Schreibvorgängen.

Example

Im folgenden Beispiel wird die Prüfpunktposition für einen Datenstrom festgelegt:

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

Gemeinsam

Konsolenspüle

Delta Lake

Die folgenden Optionen gelten beim Schreiben eines Datenstroms in eine Delta Lake-Tabelle mit format("delta"). Überschreiboptionen wie overwriteSchema, replaceWhereund partitionOverwriteMode werden für Streaming-Schreibvorgänge nicht unterstützt.

Dateispüle

Die folgende Option gilt beim Schreiben eines Datenstroms in dateibasierte Formate (Parkett, JSON, CSV, ORC, Text). Formatspezifische Optionen finden Sie unter DataFrameWriter-Optionen.

Kafka Spüle

Eine vollständige Liste der Optionen zum Schreiben von Datenströmen in Kafka finden Sie unter "Optionen".

Speichersenke

Spark-Funktionsoptionen

Einige integrierte Spark SQL-Funktionen akzeptieren eine options Zuordnung, die das Analyse- oder Serialisierungsverhalten steuert. Übergeben Sie Optionen als Python dict oder scala Map[String, String].

Example

Im folgenden Beispiel wird eine JSON-Spalte analysiert, während falsch formatierte Datensätze abgelegt werden:

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

Avro-Funktionen akzeptieren dieselben Optionen wie die entsprechenden DataFrame-Optionen:

• from_avro und schema_of_avro verwenden Sie DataFrameReader Avro-Optionen.
• to_avro verwendet DataFrameWriter Avro-Optionen.

Example

Im folgenden Beispiel wird eine Avro-Spalte mit aktivierter Schemaentwicklung decodiert:

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

Darüber hinaus akzeptieren die Schemaregistrierungsvarianten die from_avroto_avro folgenden Optionen:

CSV

CSV-Funktionen akzeptieren dieselben Optionen wie die entsprechenden DataFrame-Optionen:

• from_csv und schema_of_csv verwenden Sie DataFrameReader-CSV-Optionen.
• to_csv verwendet CSV-Optionen für DataFrameWriter.

Example

Im folgenden Beispiel wird CSV mit einem benutzerdefinierten Trennzeichen und NULL -Wert gelesen:

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

JSON-Funktionen akzeptieren dieselben Optionen wie die entsprechenden DataFrame-Optionen:

• from_json und schema_of_json verwenden Sie DIE JSON-Optionen von DataFrameReader.
• to_json verwendet JSON-Optionen für DataFrameWriter.

Example

Das folgende Beispiel schreibt JSON mit NULL ignorierten Feldern und aktivierter ziemlicher Formatierung:

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 und to_protobuf verwenden Sie keine dateibasierte Datenquelle. Protobuf-Daten werden mit diesen Funktionen immer als binäre Spalten gelesen und geschrieben. Optionen werden als Groß Map[String, String] - und Kleinschreibung übergeben.

Example

Im folgenden Beispiel wird eine Protobuf-Spalte im PERMISSIVE-Modus decodiert:

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

Protobuf-Funktionen verwenden die folgenden Optionen:

XML

XML-Funktionen akzeptieren dieselben Optionen wie die entsprechenden DataFrame-Optionen:

• from_xml und schema_of_xml verwenden Sie DataFrameReader-XML-Optionen.
• to_xml verwendet DataFrameWriter-XML-Optionen.

Example

Im folgenden Beispiel wird XML mit benutzerdefinierten Stamm- und Zeilentags geschrieben:

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