Naslaginformatie over Spark-API-opties

Deze pagina bevat beschikbare invoer- en uitvoeropties voor Spark-API's die gegevens lezen en schrijven.

Opties voor DataFrameReader

Gebruik deze opties met DataFrameReader.option(), DataFrameReader.options(), read_files, COPY INTO en Auto Loader om te bepalen hoe Azure Databricks gegevensbestanden leest.

Example

In het volgende voorbeeld wordt het lezen van JSON-bestanden ingesteld multiLineTrue :

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)

Gemeenschappelijk

De volgende opties zijn van toepassing op alle bestandsindelingen.

Avro

CSV

Excel

JSON

Kafka

Zie DataStreamReader Kafka-opties voor de volledige lijst met opties voor Kafka-lezers. De volgende opties zijn alleen van toepassing op batchleesbewerkingen met behulp van spark.read.format("kafka").

ORC

Parket

Statusarchief

Gebruik deze opties met spark.read.format("statestore") of de read_statestore functie met de tabelwaarde om statusgegevens van structured streaming te lezen. Zie Lees informatie over de status van Structured Streaming.

Tekst

XML

Opties voor DataStreamReader

Gebruik deze opties voor DataStreamReader.option() het configureren van streaming-leesbewerkingen uit Delta Lake-tabellen en andere bronnen op basis van bestanden.

Zie DataFrameReader-opties voor opties voor bestandsindelingen (JSON, CSV, Parquet en andere).

Zie AutoLoader (cloudFiles.*) voor opties voor automatisch laden.

Example

In het volgende voorbeeld wordt ingesteld maxFilesPerTrigger op 10 een Delta Lake-tabelstroom:

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

Gemeenschappelijk

De volgende opties zijn van toepassing op Delta Lake-tabellen en andere streamingbronnen op basis van bestanden.

Automatisch laadprogramma

Gebruik deze opties met de cloudFiles bron om automatisch laden te configureren voor streamingopname vanuit cloudopslag. Opties die specifiek zijn voor de cloudFiles bron, worden voorafgegaan door cloudFiles ze in een afzonderlijke naamruimte te houden van andere opties voor gestructureerd streamen .

Gemeenschappelijk

Adreslijstvermelding

Bestandsmelding

Zie Voor informatie over het configureren van de modus voor bestandsmeldingen, inclusief vereiste cloudmachtigingen, installatie-instructies en verificatiemethoden, autoladerstreams configureren in de modus voor bestandsmeldingen.

(1) Automatisch laadprogramma voegt standaard de volgende sleutel-waarde tagparen toe op basis van best effort:

• vendor: Databricks
• path: de locatie waar de gegevens worden geladen. Niet beschikbaar in GCP vanwege labelbeperkingen.
• checkpointLocation: De locatie van het controlepunt van de stream. Niet beschikbaar in GCP vanwege labelbeperkingen.
• streamId: Een wereldwijd unieke id voor de stream.

Databricks behoudt deze sleutelnamen voor en u kunt hun waarden niet overschrijven.

Cloudspecifiek

Auto Loader biedt opties voor het configureren van de cloudinfrastructuur voor bestandsmeldingsmodus. Zie Voor de vereiste cloudmachtigingen en installatie-instructies autolaadprogramma's configureren in de modus voor bestandsmeldingen.

AWS

Geef alleen de volgende opties op als u kiest cloudFiles.useNotifications = true en u wilt dat automatisch laadprogramma de meldingsservices voor u instelt:

Geef alleen de volgende optie op als u kiest cloudFiles.useNotifications = true en u wilt dat automatisch laadprogramma een wachtrij gebruikt die u al hebt ingesteld:

AWS-verificatieopties

Geef de volgende verificatieoptie op om een Databricks-servicereferentie te gebruiken:

Wanneer databricks-servicereferenties of IAM-rollen niet beschikbaar zijn, kunt u in plaats daarvan de volgende verificatieopties opgeven:

Azure

U moet waarden opgeven voor alle volgende opties als u opgeeft cloudFiles.useNotifications = true en u wilt dat automatisch laadprogramma de meldingsservices voor u instelt:

Als er geen Databricks-servicereferentie beschikbaar is, kunt u in plaats daarvan de volgende verificatieopties opgeven:

Geef alleen de volgende optie op als u instelt cloudFiles.useNotifications = true en u wilt dat automatisch laden een bestaande wachtrij gebruikt:

GCP

Auto Loader kan meldingsservices automatisch voor u instellen door gebruik te maken van Databricks-servicereferenties. Voor het serviceaccount dat is gemaakt met de Databricks-servicereferentie, zijn de machtigingen vereist die zijn gespecificeerd in Configureer Auto Loader streams in de bestandsmeldingsmodus.

Als er geen Databricks-servicereferentie beschikbaar is, kunt u Google-serviceaccounts rechtstreeks gebruiken. U kunt uw cluster configureren om aan te nemen dat een service-account wordt gebruikt door de installatie van de Google-service te volgen, of door de volgende authenticatieopties direct op te geven:

Geef alleen de volgende optie op als u kiest cloudFiles.useNotifications = true en u wilt dat automatisch laadprogramma een wachtrij gebruikt die u al hebt ingesteld:

Delta Lake

De volgende opties zijn van toepassing bij het lezen vanuit een Delta Lake-tabel met behulp van spark.readStream.

Kafka

Gebruik deze opties met een spark.readStream.format("kafka") van de volgende opties:spark.read.format("kafka")

De volgende opties zijn alleen van toepassing op streaming-leesbewerkingen met spark.readStream.format("kafka"):

Zie De opties van DataFrameReader Kafka voor offsetopties die alleen van toepassing zijn op batchleesbewerkingen.spark.read.format("kafka")

Zie Opties voor kafka-client (kafka.*) en verificatieopties.

Opties voor DataFrameWriter

Gebruik deze opties met DataFrameWriter.option() en DataFrameWriterV2.option() om te bepalen hoe Azure Databricks gegevens schrijft.

Example

In het volgende voorbeeld wordt ingesteld mergeSchema op True het schrijven van een Delta Lake-tabel:

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

CSV

Excel

JSON

ORC

Parket

Tekst

XML

Opties voor DataStreamWriter

Gebruik deze opties om DataStreamWriter.option() streaming-schrijfbewerkingen te configureren.

Example

In het volgende voorbeeld wordt de locatie van het controlepunt voor een stream ingesteld:

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

Gemeenschappelijk

Console-sink

Delta Lake

De volgende opties zijn van toepassing bij het schrijven van een stroom naar een Delta Lake-tabel met behulp van format("delta"). Opties voor alleen overschrijven, zoals overwriteSchema, replaceWhereen partitionOverwriteMode worden niet ondersteund voor streaming-schrijfbewerkingen.

Bestandssink

De volgende optie is van toepassing bij het schrijven van een stream naar bestandsindelingen (Parquet, JSON, CSV, ORC, tekst). Zie DataFrameWriter-opties voor specifieke indelingen.

Kafka-sink

Zie Opties voor een volledige lijst met opties voor het schrijven van streams naar Kafka.

Geheugensink

Spark-functieopties

Sommige ingebouwde Spark SQL-functies accepteren een options kaart waarmee het parserings- of serialisatiegedrag wordt beheerd. Geef opties door als een Python dict of een Scala-Map[String, String].

Example

In het volgende voorbeeld wordt een JSON-kolom geparseerd terwijl onjuiste records worden verwijderd:

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-functies accepteren dezelfde opties als de bijbehorende DataFrame-opties:

• from_avro en schema_of_avro gebruik Avro-opties voor DataFrameReader.
• to_avro maakt gebruik van Avro-opties voor DataFrameWriter.

Example

In het volgende voorbeeld wordt een Avro-kolom gedecodeert waarvoor schemaontwikkeling is ingeschakeld:

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

Daarnaast accepteren de schemaregistervarianten from_avro de volgende opties:to_avro

CSV

CSV-functies accepteren dezelfde opties als de bijbehorende DataFrame-opties:

• from_csv en schema_of_csv gebruik CSV-opties voor DataFrameReader.
• to_csv maakt gebruik van CSV-opties voor DataFrameWriter.

Example

In het volgende voorbeeld wordt CSV gelezen met een aangepast scheidingsteken en NULL een aangepaste waarde:

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-functies accepteren dezelfde opties als de bijbehorende DataFrame-opties:

• from_json en schema_of_json gebruik JSON-opties voor DataFrameReader.
• to_json maakt gebruik van DataFrameWriter JSON-opties.

Example

In het volgende voorbeeld wordt JSON geschreven met NULL velden genegeerd en mooie opmaak ingeschakeld:

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 en to_protobuf gebruik geen gegevensbron op basis van bestanden. Protobuf-gegevens worden altijd gelezen en geschreven als binaire kolommen met behulp van deze functies. Opties worden doorgegeven als hoofdlettergevoelig Map[String, String] .

Example

In het volgende voorbeeld wordt een Protobuf-kolom gedecodeert met behulp van de PERMISSIVE-modus:

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-functies gebruiken de volgende opties:

XML

XML-functies accepteren dezelfde opties als de bijbehorende DataFrame-opties:

• from_xml en schema_of_xml gebruik XML-opties voor DataFrameReader.
• to_xml maakt gebruik van XML-opties voor DataFrameWriter.

Example

In het volgende voorbeeld wordt XML geschreven met aangepaste hoofd- en rijtags:

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