Dokumentacja opcji interfejsu API platformy Spark

Na tej stronie wymieniono dostępne opcje danych wejściowych i wyjściowych dla interfejsów API platformy Spark, które odczytują i zapisują dane.

Opcje elementu DataFrameReader

Użyj tych opcji z DataFrameReader.option(), DataFrameReader.options(), read_files, COPY INTO i Auto Loader kontrolować, jak Azure Databricks odczytuje pliki danych.

Example

Poniższy przykład ustawia wartość multiLine na True do odczytywania plików 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)

Wspólne

Poniższe opcje mają zastosowanie do wszystkich formatów plików.

Avro

Podczas odczytywania plików Avro mają zastosowanie następujące opcje.

CSV

Podczas odczytywania plików CSV mają zastosowanie następujące opcje.

Excel

Podczas odczytywania plików Excel mają zastosowanie następujące opcje.

JSON

Podczas odczytywania plików JSON mają zastosowanie następujące opcje.

Kafka

Aby uzyskać pełną listę opcji czytnika platformy Kafka, zobacz Opcje dataStreamReader Kafka. Poniższe opcje dotyczą tylko operacji odczytu wsadowego przy użyciu polecenia spark.read.format("kafka").

ORC

Podczas odczytywania plików ORC mają zastosowanie następujące opcje.

Parkiet

Podczas odczytywania plików Parquet mają zastosowanie następujące opcje.

Magazyn stanów

Użyj tych opcji z funkcją lub funkcją spark.read.format("statestore")read_statestore wartości tabeli, aby odczytać dane stanu przesyłania strumieniowego ze strukturą. Zobacz Jak odczytać informacje o stanie strukturalnego strumieniowania danych.

Tekst

Podczas odczytywania plików tekstowych mają zastosowanie następujące opcje.

XML

Podczas odczytywania plików XML mają zastosowanie następujące opcje.

Opcje DataStreamReader

Te opcje umożliwiają DataStreamReader.option() skonfigurowanie odczytów przesyłanych strumieniowo z tabel usługi Delta Lake i innych źródeł opartych na plikach.

Aby uzyskać opcje formatowania plików (JSON, CSV, Parquet i inne), zobacz Opcje elementu DataFrameReader.

Aby zapoznać się z opcjami automatycznego modułu ładującego (), zobacz cloudFiles.* (Automatyczne ładowanie).

Example

W poniższym przykładzie ustawiono wartość maxFilesPerTrigger10 dla strumienia tabeli usługi 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")

Wspólne

Poniższe opcje dotyczą tabel usługi Delta Lake i innych źródeł przesyłania strumieniowego opartego na plikach.

Moduł automatycznego ładowania

Użyj tych opcji ze cloudFiles źródłem, aby skonfigurować moduł automatycznego ładowania na potrzeby pozyskiwania danych przesyłanych strumieniowo z magazynu w chmurze. Opcje specyficzne dla cloudFiles źródła są poprzedzone prefiksem cloudFiles , aby zachować je w oddzielnej przestrzeni nazw od innych opcji źródła przesyłania strumieniowego ze strukturą .

Wspólne

Poniższe opcje dotyczą wszystkich konfiguracji modułu ładującego automatycznego.

Lista katalogów

Poniższa opcja ma zastosowanie w przypadku korzystania z trybu wyświetlania listy katalogów.

Powiadomienie o pliku

Aby uzyskać informacje na temat konfigurowania trybu powiadomień plików, w tym wymaganych uprawnień do chmury, instrukcji konfiguracji i metod uwierzytelniania, zobacz Konfigurowanie strumieni automatycznego modułu ładującego w trybie powiadomień plików.

Tagi zasobów dostawcy usług w chmurze

Automatyczne ładowanie domyślnie dodaje następujące pary tagów klucz-wartość domyślnie w oparciu o najlepsze wysiłki:

• vendor: Databricks
• path: lokalizacja, z której są ładowane dane. Niedostępne w GCP z powodu ograniczeń etykietowania.
• checkpointLocation: lokalizacja punktu kontrolnego strumienia. Niedostępne w GCP z powodu ograniczeń etykietowania.
• streamId: globalnie unikatowy identyfikator strumienia.

Usługa Databricks rezerwuje te nazwy kluczy i nie można zastąpić ich wartości.

Aby uzyskać więcej informacji na temat platformy Azure, zobacz Nazewnictwo kolejek i metadanych oraz omówienie properties.labels w subskrypcjach zdarzeń. Moduł automatycznego ładowania przechowuje te pary tagów klucz-wartość w formacie JSON jako etykiety.

Specyficzne dla chmury

Moduł automatycznego ładowania ma opcje konfigurowania infrastruktury chmury dla trybu powiadomień plików. Aby uzyskać wymagane uprawnienia do chmury i instrukcje dotyczące konfiguracji, zobacz Konfigurowanie strumieni automatycznego ładowania w trybie powiadomień plików.

Azure

Należy określić wartości dla wszystkich następujących opcji, jeśli określisz cloudFiles.useNotifications = true i chcesz, aby automatycznie ładujący skonfigurować usługi powiadomień dla Ciebie:

Jeśli poświadczenie usługi Databricks jest niedostępne, możesz określić następujące opcje uwierzytelniania:

Określ następującą opcję tylko w przypadku ustawienia cloudFiles.useNotifications = true i chcesz, aby moduł automatycznego ładowania używał istniejącej kolejki:

Delta Lake

Poniższe opcje mają zastosowanie podczas odczytywania z tabeli usługi Delta Lake przy użyciu polecenia spark.readStream.

Kafka

Użyj tych opcji z opcją spark.readStream.format("kafka") lub spark.read.format("kafka"):

Następujące opcje dotyczą tylko odczytów przesyłanych strumieniowo za pomocą polecenia spark.readStream.format("kafka"):

Aby uzyskać opcje przesunięcia, które mają zastosowanie tylko do operacji odczytu wsadowego za pomocą spark.read.format("kafka")polecenia , zobacz Opcje elementu DataFrameReader platformy Kafka.

Uwierzytelnianie

Usługa Databricks zaleca użycie poświadczeń usługi wykazu aparatu Unity do uwierzytelniania w usługach Kafka zarządzanych przez chmurę (AWS MSK, Azure Event Hubs lub Google Cloud Managed Kafka).

Jeśli poświadczenie usługi jest niedostępne, użyj opcji SASL/SSL (przekazywanych jako kafka.* właściwości). Jeśli używasz poświadczeń usługi, nie musisz określać kafka.sasl.mechanismwartości , kafka.sasl.jaas.configani kafka.security.protocol.

Aby uzyskać pełne instrukcje dotyczące konfigurowania uwierzytelniania, zobacz Uwierzytelnianie.

Pub/Sub

Użyj tych opcji, spark.readStream.format("pubsub") aby zasubskrybować usługę Google Pub/Sub. Opcje subscriptionId, topicIdi projectId są wymagane.

Aby uzyskać więcej informacji na temat pubu/subskrypcji, zobacz Subskrybuj usługę Google Pub/Sub.

Puls

Użyj tych opcji za pomocą polecenia spark.readStream.format("pulsar") , aby przesyłać strumieniowo z usługi Apache Pulsar. Dostępne w środowisku Databricks Runtime 14.1 lub nowszym.

Wymagane są następujące opcje. Musisz określić dokładnie jedną z wartości topic, topicslub topicsPattern.

Obsługiwane są również następujące opcje:

Dodatkowe konfiguracje klienta pulsar, administratora i czytelnika można określić przy użyciu następujących wzorców opcji:

Aby uzyskać więcej informacji na temat opcji uwierzytelniania klienta Pulsar i administratora, zobacz Uwierzytelnianie.

Uwierzytelnianie

Usługa Azure Databricks obsługuje uwierzytelnianie do Pulsara przy użyciu truststore i keystore. Azure Databricks zaleca używanie wpisów tajnych do przechowywania szczegółów uwierzytelniania. Zobacz Zarządzanie tajemnicami.

Jeśli strumień używa PulsarAdminelementu , możesz również ustawić następujące opcje:

Aby zapoznać się z przykładami uwierzytelniania, zobacz Uwierzytelnianie w pulsacie.

Opcje elementu DataFrameWriter

Użyj tych opcji z DataFrameWriter.option() i DataFrameWriterV2.option(), aby kontrolować, jak Azure Databricks zapisuje dane.

Example

W poniższym przykładzie ustawiono wartość mergeSchema na True potrzeby pisania tabeli usługi 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

Podczas zapisywania plików Avro mają zastosowanie następujące opcje.

Delta Lake i Apache Iceberg

Podczas pisania tabel Delta Lake i Apache Iceberg mają zastosowanie następujące opcje.

CSV

Podczas zapisywania plików CSV mają zastosowanie następujące opcje.

Excel

Podczas zapisywania plików Excel mają zastosowanie następujące opcje.

JSON

Podczas zapisywania plików JSON mają zastosowanie następujące opcje.

ORC

Podczas zapisywania plików ORC mają zastosowanie następujące opcje.

Parkiet

Podczas zapisywania plików Parquet mają zastosowanie następujące opcje.

Tekst

Podczas pisania plików tekstowych mają zastosowanie następujące opcje.

XML

Podczas pisania plików XML mają zastosowanie następujące opcje.

Opcje DataStreamWriter

Te opcje umożliwiają DataStreamWriter.option() skonfigurowanie zapisów przesyłanych strumieniowo.

Example

W poniższym przykładzie ustawiono lokalizację punktu kontrolnego dla strumienia:

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

Wspólne

Poniższe opcje dotyczą wszystkich operacji zapisu przesyłania strumieniowego.

Ujście konsoli

Podczas zapisywania strumieni do ujścia konsoli mają zastosowanie następujące opcje.

Delta Lake

Poniższe opcje mają zastosowanie podczas zapisywania strumienia do tabeli usługi Delta Lake przy użyciu polecenia format("delta"). Opcje tylko zastępowania, takie jak overwriteSchema, replaceWherei partitionOverwriteMode nie są obsługiwane w przypadku zapisów przesyłanych strumieniowo.

Ujście pliku

Poniższa opcja ma zastosowanie podczas zapisywania strumienia w formatach opartych na plikach (Parquet, JSON, CSV, ORC, text). Aby uzyskać informacje o opcjach specyficznych dla formatu, zobacz Opcje elementu DataFrameWriter.

Ujście platformy Kafka

Poniższe opcje mają zastosowanie podczas zapisywania na platformie Kafka.

Ujście pamięci

Podczas zapisywania strumieni do ujścia pamięci mają zastosowanie następujące opcje.

Opcje funkcji platformy Spark

Niektóre wbudowane funkcje platformy Spark SQL akceptują options mapę, która steruje zachowaniem analizowania lub serializacji. Przekaż opcje jako Python dict lub Scala Map[String, String].

Example

Poniższy przykład analizuje kolumnę JSON podczas upuszczania nieprawidłowo sformułowanych rekordów:

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

Funkcje Avro akceptują te same opcje co odpowiednie opcje ramki danych:

• from_avro i schema_of_avro użyj opcji DataFrameReader Avro.
• to_avro używa opcji DataFrameWriter Avro.

Example

Poniższy przykład dekoduje kolumnę Avro z włączoną ewolucją schematu:

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

Ponadto warianty rejestru schematów from_avro i to_avro akceptują następujące opcje:

CSV

Funkcje CSV akceptują te same opcje co odpowiednie opcje ramki danych:

• from_csv i schema_of_csv użyj opcji CSV Elementu DataFrameReader.
• to_csv używa opcji CSV elementu DataFrameWriter.

Example

Poniższy przykład odczytuje plik CSV z separatorem niestandardowym i NULL wartością:

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

Funkcje JSON akceptują te same opcje co odpowiednie opcje ramki danych:

• from_json i schema_of_json użyj opcji DataFrameReader JSON.
• to_json używa opcji JSON elementu DataFrameWriter.

Example

W poniższym przykładzie jest zapisywany kod JSON z polami NULL ignorowanymi i włączonym dość formatowaniem:

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 nie to_protobuf należy używać źródła danych opartego na plikach. Dane Protobuf są zawsze odczytywane i zapisywane jako kolumny binarne przy użyciu tych funkcji. Opcje są przekazywane jako i Map[String, String] są uwzględniane wielkość liter.

Example

Poniższy przykład dekoduje kolumnę Protobuf przy użyciu trybu 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")))

Funkcje Protobuf korzystają z następujących opcji:

XML

Funkcje XML akceptują te same opcje co odpowiednie opcje ramki danych:

• from_xml i schema_of_xml użyj opcji XML elementu DataFrameReader.
• to_xml używa opcji XML elementu DataFrameWriter.

Example

Poniższy przykład zapisuje kod XML z niestandardowymi tagami katalogu głównego i wiersza:

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