Spark API seçenekleri başvurusu

Bu sayfada, veri okuyan ve yazan Spark API'leri için kullanılabilir giriş ve çıkış seçenekleri listelenmiştir.

DataFrameReader seçenekleri

Azure Databricks veri dosyalarını nasıl okuduğunu denetlemek için DataFrameReader.option(), DataFrameReader.options(), read_files, COPY INTO ve Auto Loader ile bu seçenekleri kullanın.

Example

Aşağıdaki örnek, JSON dosyalarını okumak için olarak multiLine ayarlarTrue:

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)

Ortak

Aşağıdaki seçenekler tüm dosya biçimleri için geçerlidir.

Avro

Avro dosyaları okunurken aşağıdaki seçenekler geçerlidir.

CSV

CSV dosyaları okunurken aşağıdaki seçenekler geçerlidir.

Excel

Excel dosyaları okurken aşağıdaki seçenekler geçerlidir.

JSON

JSON dosyaları okunurken aşağıdaki seçenekler geçerlidir.

Kafka

Kafka okuyucu seçeneklerinin tam listesi için bkz. DataStreamReader Kafka seçenekleri. Aşağıdaki seçenekler yalnızca kullanılarak spark.read.format("kafka")toplu okuma işlemleri için geçerlidir.

ORC

ORC dosyaları okunurken aşağıdaki seçenekler geçerlidir.

Parke

Parquet dosyaları okunurken aşağıdaki seçenekler geçerlidir.

Durum deposu

Yapılandırılmış Akış durumu verilerini okumak için veya spark.read.format("statestore") tablo değerli işleviyle read_statestore bu seçenekleri kullanın. Bkz. Yapılandırılmış Akış durumu bilgilerini okuyun.

Metin

Metin dosyaları okunurken aşağıdaki seçenekler geçerlidir.

XML

XML dosyaları okunurken aşağıdaki seçenekler geçerlidir.

DataStreamReader seçenekleri

Delta Lake tablolarından ve diğer dosya tabanlı kaynaklardan akış okumalarını yapılandırmak için ile DataStreamReader.option() bu seçenekleri kullanın.

Dosya biçimi seçenekleri (JSON, CSV, Parquet ve diğerleri) için bkz. DataFrameReader seçenekleri.

Otomatik Yükleyici (cloudFiles.*) seçenekleri için bkz. Otomatik Yükleyici.

Example

Aşağıdaki örnek, Delta Lake tablo akışı için olarak maxFilesPerTrigger ayarlar10:

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

Ortak

Aşağıdaki seçenekler Delta Lake tabloları ve diğer dosya tabanlı akış kaynakları için geçerlidir.

Otomatik Yükleyici

cloudFiles bulut depolama alanından akış alımı için yapılandırmak üzere kaynakla birlikte bu seçenekleri kullanın. Kaynağa özgü cloudFiles seçenekler, bunları diğer cloudFiles kaynağı seçeneklerinden ayrı bir ad alanında tutmak için ön eke sahiptir.

Ortak

Aşağıdaki seçenekler tüm Otomatik Yükleyici yapılandırmaları için geçerlidir.

Dizin listesi

Dizin listeleme modu kullanılırken aşağıdaki seçenek geçerlidir.

Dosya bildirimi

Gerekli bulut izinleri, kurulum yönergeleri ve kimlik doğrulama yöntemleri dahil olmak üzere dosya bildirim modunu yapılandırma hakkında bilgi için bkz. Dosya bildirim modunda Otomatik Yükleyici akışlarını yapılandırma.

Bulut sağlayıcısı kaynak etiketleri

Otomatik Yükleyici varsayılan olarak en iyi çaba temelinde aşağıdaki anahtar-değer etiketi çiftlerini ekler:

• vendor: Databricks
• path: Verilerin yüklendiği konum. Etiketleme sınırlamaları nedeniyle GCP'de kullanılamıyor.
• checkpointLocation: Akışın denetim noktasının konumu. Etiketleme sınırlamaları nedeniyle GCP'de kullanılamıyor.
• streamId: Akış için genel olarak benzersiz bir tanımlayıcı.

Databricks bu anahtar adlarını ayırır ve değerlerinin üzerine yazamazsınız.

Azure hakkında daha fazla bilgi için, Adlandırma Kuyrukları ve Metadata ile properties.labels kapsamını inceleyin. Otomatik Yükleyici bu anahtar-değer etiket çiftlerini JSON'da etiket olarak depolar.

Buluta özgü

Otomatik Yükleyici,dosya bildirim modu için bulut altyapısını yapılandırma seçeneklerine sahiptir. Gerekli bulut izinleri ve kurulum yönergeleri için bkz. Dosya bildirim modunda Otomatik Yükleyici akışlarını yapılandırma.

Azure

Otomatik Yükleyici'nin bildirim hizmetlerini sizin için ayarlamasını istiyorsanız, aşağıdaki seçeneklerin tümü için değerleri belirtmeniz cloudFiles.useNotifications = true gerekir:

Databricks hizmeti kimlik bilgileri kullanılamıyorsa, bunun yerine aşağıdaki kimlik doğrulama seçeneklerini belirtebilirsiniz:

Aşağıdaki seçeneği yalnızca Otomatik cloudFiles.useNotifications = true Yükleyici'nin var olan bir kuyruğu kullanmasını istiyorsanız belirtin:

Delta Gölü

aşağıdaki seçenekler kullanarak spark.readStreamdelta lake tablosundan okurken geçerlidir.

Kafka

Veya spark.readStream.format("kafka")ile spark.read.format("kafka") bu seçenekleri kullanın:

Aşağıdaki seçenekler yalnızca ile spark.readStream.format("kafka")akış okumaları için geçerlidir:

ile spark.read.format("kafka")yalnızca toplu okuma işlemlerine uygulanan uzaklık seçenekleri için bkz. DataFrameReader Kafka seçenekleri.

Kimlik doğrulama

Databricks, bulut tarafından yönetilen Kafka hizmetlerinde (AWS MSK, Azure Event Hubs veya Google Cloud Managed Kafka) kimlik doğrulaması yapmak için Unity Kataloğu hizmeti kimlik bilgilerinin kullanılmasını önerir.

Hizmet kimlik bilgileri olmadığında SASL/SSL seçeneklerini kullanın (özellik olarak kafka.* geçirilir). Hizmet kimlik bilgilerini kullandığınızda, , kafka.sasl.mechanismveya kafka.sasl.jaas.configbelirtmeniz kafka.security.protocolgerekmez.

Tam kimlik doğrulaması kurulum yönergeleri için bkz. Kimlik doğrulaması.

Pub/Sub

Google Pub/Sub'a abone olmak için ile bu seçenekleri spark.readStream.format("pubsub") kullanın. , subscriptionIdve topicId seçenekleri projectIdgereklidir.

Pub/Sub hakkında daha fazla bilgi için bkz. Google Pub/Sub'a abone olma.

Pulsar

Apache Pulsar'dan akış yapmak için ile spark.readStream.format("pulsar") bu seçenekleri kullanın. Databricks Runtime 14.1 ve üzerinde kullanılabilir.

Aşağıdaki seçenekler gereklidir. , veya topictam olarak birini topicstopicsPatternbelirtmelisiniz.

Aşağıdaki seçenekler de desteklenir:

Aşağıdaki seçenek desenlerini kullanarak ek Pulsar istemci, yönetici ve okuyucu yapılandırmaları belirtebilirsiniz:

Pulsar istemcisi ve yönetici kimlik doğrulaması seçenekleri hakkında daha fazla bilgi için bkz. Kimlik doğrulaması.

Kimlik doğrulama

Azure Databricks, Pulsar'da truststore ve keystore kimlik doğrulamasını destekler. Azure Databricks kimlik doğrulama ayrıntılarını depolamak için gizli dizileri kullanmanızı önerir. Bkz. Gizli yönetim.

Akış bir PulsarAdminkullanıyorsa, aşağıdaki seçenekleri de ayarlayabilirsiniz:

Kimlik doğrulama örnekleri için bkz. Pulsar'da kimlik doğrulaması.

DataFrameWriter seçenekleri

DataFrameWriter.option() ve DataFrameWriterV2.option() ile bu seçenekleri kullanarak Azure Databricks verileri nasıl yazacaklarını denetleyin.

Example

Aşağıdaki örnek, Delta Lake tablosu yazmak için olarak mergeSchema ayarlırTrue:

Python

df.write.format("delta").option("mergeSchema", True).saveAsTable("my_table")

Scala

df.write.format("delta").option("mergeSchema", "true").saveAsTable("my_table")

Avro

Avro dosyaları yazılırken aşağıdaki seçenekler geçerlidir.

Delta Gölü ve Apache Iceberg

Delta Lake ve Apache Iceberg tabloları yazarken aşağıdaki seçenekler geçerlidir.

CSV

CSV dosyaları yazılırken aşağıdaki seçenekler geçerlidir.

Excel

Excel dosyaları yazarken aşağıdaki seçenekler geçerlidir.

JSON

JSON dosyaları yazılırken aşağıdaki seçenekler geçerlidir.

ORC

ORC dosyaları yazılırken aşağıdaki seçenekler geçerlidir.

Parke

Parquet dosyaları yazılırken aşağıdaki seçenekler geçerlidir.

Metin

Metin dosyaları yazılırken aşağıdaki seçenekler geçerlidir.

XML

XML dosyaları yazılırken aşağıdaki seçenekler geçerlidir.

DataStreamWriter seçenekleri

Akış yazmalarını yapılandırmak için ile DataStreamWriter.option() bu seçenekleri kullanın.

Example

Aşağıdaki örnek bir akışın denetim noktası konumunu ayarlar:

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

Ortak

Aşağıdaki seçenekler tüm akış yazma işlemleri için geçerlidir.

Konsol havuzu

Konsol havuzuna akış yazarken aşağıdaki seçenekler geçerlidir.

Delta Gölü

kullanarak format("delta")delta lake tablosuna akış yazarken aşağıdaki seçenekler geçerlidir. , ve overwriteSchema gibi replaceWherepartitionOverwriteModeyalnızca üzerine yazma seçenekleri akış yazma işlemleri için desteklenmez.

Dosya havuzu

Dosya tabanlı biçimlere (Parquet, JSON, CSV, ORC, metin) akış yazarken aşağıdaki seçenek geçerlidir. Biçime özgü seçenekler için bkz. DataFrameWriter seçenekleri.

Kafka havuzu

Kafka'ya yazarken aşağıdaki seçenekler geçerlidir.

Bellek havuzu

Bellek havuzuna akış yazarken aşağıdaki seçenekler geçerlidir.

Spark işlevi seçenekleri

Bazı Spark SQL yerleşik işlevleri ayrıştırma veya serileştirme davranışını denetleyan bir options haritayı kabul eder. Seçenekleri Python dict veya Scala Map[String, String] olarak geçirin.

Example

Aşağıdaki örnek, hatalı biçimlendirilmiş kayıtları bırakırken JSON sütununu ayrıştırmaktadır:

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 işlevleri, ilgili DataFrame seçenekleriyle aynı seçenekleri kabul edin:

• from_avro ve schema_of_avroDataFrameReader Avro seçeneklerini kullanın.
• to_avro DataFrameWriter Avro seçeneklerini kullanır.

Example

Aşağıdaki örnek şema evrimi etkin bir Avro sütununun kodunu çözer:

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

Buna ek olarak, aşağıdaki seçeneklerin from_avro Schema Registry varyantları ve to_avro kabul eder:

CSV

CSV işlevleri, ilgili DataFrame seçenekleriyle aynı seçenekleri kabul edin:

• from_csv ve schema_of_csvDataFrameReader CSV seçeneklerini kullanın.
• to_csv DataFrameWriter CSV seçeneklerini kullanır.

Example

Aşağıdaki örnek, özel ayırıcı ve NULL değerle CSV'yi okur:

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 işlevleri, ilgili DataFrame seçenekleriyle aynı seçenekleri kabul edin:

• from_json ve schema_of_jsonDataFrameReader JSON seçeneklerini kullanın.
• to_json DataFrameWriter JSON seçeneklerini kullanır.

Example

Aşağıdaki örnek, alanları yoksayılan ve oldukça biçimlendirme etkin olan NULL JSON yazar:

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 ve to_protobuf dosya tabanlı datasource kullanmayın. Protobuf verileri her zaman bu işlevler kullanılarak ikili sütunlar olarak okunur ve yazılır. Seçenekler olarak Map[String, String] geçirilir ve büyük/küçük harfe duyarlıdır.

Example

Aşağıdaki örnek PERMISSIVE modunu kullanarak bir Protobuf sütununun kodunu çözer:

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 işlevleri aşağıdaki seçenekleri kullanır:

XML

XML işlevleri, karşılık gelen DataFrame seçenekleriyle aynı seçenekleri kabul edin:

• from_xml ve schema_of_xmlDataFrameReader XML seçeneklerini kullanın.
• to_xml DataFrameWriter XML seçeneklerini kullanır.

Example

Aşağıdaki örnek özel kök ve satır etiketleriyle XML yazar:

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