CSV dosyalarını okuma ve yazma

CSV (virgülle ayrılmış değerler), veri değişimi, ETL işlem hatları ve genel amaçlı veri depolama için yaygın olarak kullanılan düz metin tablosal biçimidir. Azure Databricks şema çıkarımı, sıkıştırma, hatalı biçimlendirilmiş kayıt işleme ve kurtarılan veriler dahil olmak üzere Apache Spark ile hem okuma hem de yazma için CSV'yi destekler.

Prerequisites

Azure Databricks, CSV dosyalarını kullanmak için ek yapılandırma gerektirmez. Ancak CSV dosyalarının akışını yapmak için Otomatik Yükleyici gerekir.

Seçenekler

CSV veri kaynaklarını yapılandırmak için .option() ve .options() öğelerinin DataFrameReader ve DataFrameWriter yöntemlerini kullanın. Desteklenen seçeneklerin tam listesi için bkz DataFrameReader . CSV seçenekleri ve DataFrameWriter CSV seçenekleri.

Usage

Aşağıdaki örneklerde CSV dosyalarını okuma ve yazma, şema belirtme ve hatalı biçimlendirilmiş kayıtları işleme işlemleri gösterilmektedir.

CSV dosyalarını okuma

Aşağıdaki örnekte Wanderbricks örnek veri kümesi kullanılmıştır. İnceleme verilerini CSV'ye yazar, ardından tekrar okur.

Python

# Write wanderbricks reviews to CSV format
df = spark.read.table("samples.wanderbricks.reviews")
df.write.format("csv").option("header", "true").save("/Volumes/<catalog>/<schema>/<volume>/reviews_csv")

# Read the CSV file into a DataFrame
df = (spark.read
  .format("csv")
  .option("header", "true")
  .option("inferSchema", "true")
  .load("/Volumes/<catalog>/<schema>/<volume>/reviews_csv"))
display(df)
df.printSchema()

Scala

// Write wanderbricks reviews to CSV format
val reviews = spark.read.table("samples.wanderbricks.reviews")
reviews.write.format("csv").option("header", "true").save("/Volumes/<catalog>/<schema>/<volume>/reviews_csv")

// Read the CSV file into a DataFrame
val df = spark.read
  .format("csv")
  .option("header", "true")
  .option("inferSchema", "true")
  .load("/Volumes/<catalog>/<schema>/<volume>/reviews_csv")
df.show()
df.printSchema()

R

df <- read.df("/Volumes/<catalog>/<schema>/<volume>/reviews_csv", source = "csv", header = "true", inferSchema = "true")
display(df)
printSchema(df)

SQL kullanarak CSV dosyalarını okuma

Aşağıdaki SQL örneği, read_files kullanarak bir CSV dosyasını okur.

-- mode "FAILFAST" aborts file parsing with a RuntimeException if malformed lines are encountered
SELECT * FROM read_files(
  'abfss://<bucket>@<storage-account>.dfs.core.windows.net/<path>/<file>.csv',
  format => 'csv',
  header => true,
  mode => 'FAILFAST')

Şema belirtme

CSV dosyasının şeması bilindiğinde, schema seçeneğiyle CSV okuyucusuna istediğiniz şemayı belirtebilirsiniz.

Python

from pyspark.sql.types import StructType, StructField, StringType, IntegerType

schema = StructType([
  StructField("review_id", StringType(), True),
  StructField("rating", IntegerType(), True),
  StructField("comment", StringType(), True)
])

df = spark.read.format("csv").schema(schema).option("header", "true").load("/Volumes/<catalog>/<schema>/<volume>/reviews_csv")
df.printSchema()

Scala

import org.apache.spark.sql.types._

val schema = StructType(Array(
  StructField("review_id", StringType, nullable = true),
  StructField("rating", IntegerType, nullable = true),
  StructField("comment", StringType, nullable = true)
))

val df = spark.read.format("csv").schema(schema).option("header", "true").load("/Volumes/<catalog>/<schema>/<volume>/reviews_csv")
df.printSchema()

SQL

SELECT * FROM read_files(
  '/Volumes/<catalog>/<schema>/<volume>/reviews_csv',
  format => 'csv',
  header => true,
  schema => 'review_id string, rating int, comment string'
)

Sütunların alt kümesini okuma

CSV ayrıştırıcısının davranışı hangi sütunların okunduğuna bağlıdır. Belirtilen şema dosya düzeniyle eşleşmiyorsa, hangi sütunlara erişildiğine bağlı olarak sonuçlar önemli ölçüde farklılık gösterebilir. CSV'de sütun adı meta verileri yoktur, bu nedenle Spark şema alanlarını konuma göre sütunlara eşler; eşleşmeyen bir şema değerleri yanlış alanlara kaydırır.

Python

from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# Read only a subset of columns by specifying a partial schema
schema = StructType([
  StructField("review_id", StringType(), True),
  StructField("rating", IntegerType(), True)
])

df = spark.read.format("csv").schema(schema).option("header", "true").load("/Volumes/<catalog>/<schema>/<volume>/reviews_csv")
display(df)

Scala

import org.apache.spark.sql.types._

val schema = StructType(Array(
  StructField("review_id", StringType, nullable = true),
  StructField("rating", IntegerType, nullable = true)
))

val df = spark.read.format("csv").schema(schema).option("header", "true").load("/Volumes/<catalog>/<schema>/<volume>/reviews_csv")
df.show()

SQL

SELECT * FROM read_files(
  '/Volumes/<catalog>/<schema>/<volume>/reviews_csv',
  format => 'csv',
  header => true,
  schema => 'review_id string, rating int'
)

Hatalı biçimlendirilmiş CSV kayıtlarını işleme

Belirtilen şemaya sahip CSV dosyalarını okurken, dosyalardaki verilerin şemayla eşleşmemesi mümkündür. Örneğin, şehir adını içeren bir alan tamsayı olarak ayrıştırılmaz. Bunun sonuçları, ayrıştırıcının çalıştığı moda bağlıdır:

• PERMISSIVE (varsayılan): Doğru ayrıştırılamayan alanlar için nulllar eklenir
• DROPMALFORMED: ayrıştırılamayan alanlar içeren satırları bırakır
• FAILFAST: Hatalı biçimlendirilmiş veriler bulunursa okumayı durdurur

Modu ayarlamak için seçeneğini kullanın mode .

Python

df = (spark.read
  .format("csv")
  .option("header", "true")
  .option("mode", "PERMISSIVE")
  .load("/Volumes/<catalog>/<schema>/<volume>/reviews_csv")
)

Scala

val df = spark.read
  .format("csv")
  .option("header", "true")
  .option("mode", "PERMISSIVE")
  .load("/Volumes/<catalog>/<schema>/<volume>/reviews_csv")

SQL

SELECT * FROM read_files(
  '/Volumes/<catalog>/<schema>/<volume>/reviews_csv',
  format => 'csv',
  header => true,
  mode => 'PERMISSIVE'
)

PERMISSIVE Modda, aşağıdaki yöntemlerden birini kullanarak doğru ayrıştırılamayan satırları incelemek mümkündür:

• Bozuk kayıtları bir dosyaya kaydetme seçeneği badRecordsPath için özel bir yol sağlayabilirsiniz.
• Şemaya sütunu _corrupt_record ekleyerek DataFrameReader'a sağlanan sonuçta elde edilen DataFrame'deki bozuk kayıtları inceleyebilirsiniz.

_corrupt_record kullanarak hatalı biçimlendirilmiş satırları incelemek için bunu şemaya ekleyin ve null olmayan değerlere göre filtreleyin:

Python

from pyspark.sql.types import StructType, StructField, StringType, IntegerType

schema = StructType([
  StructField("review_id", StringType(), True),
  StructField("rating", IntegerType(), True),
  StructField("comment", StringType(), True),
  StructField("_corrupt_record", StringType(), True)
])

df = (spark.read
  .format("csv")
  .option("header", "true")
  .option("mode", "PERMISSIVE")
  .schema(schema)
  .load("/Volumes/<catalog>/<schema>/<volume>/reviews_csv")
)
display(df.filter(df["_corrupt_record"].isNotNull()))

Scala

import org.apache.spark.sql.types._

val schema = StructType(Array(
  StructField("review_id", StringType, nullable = true),
  StructField("rating", IntegerType, nullable = true),
  StructField("comment", StringType, nullable = true),
  StructField("_corrupt_record", StringType, nullable = true)
))

val df = spark.read
  .format("csv")
  .option("header", "true")
  .option("mode", "PERMISSIVE")
  .schema(schema)
  .load("/Volumes/<catalog>/<schema>/<volume>/reviews_csv")

df.filter(df("_corrupt_record").isNotNull).show()

SQL

SELECT * FROM read_files(
  '/Volumes/<catalog>/<schema>/<volume>/reviews_csv',
  format => 'csv',
  header => true,
  mode => 'PERMISSIVE',
  schema => 'review_id string, rating int, comment string, _corrupt_record string'
)
WHERE _corrupt_record IS NOT NULL

Kurtarılan veri sütununu etkinleştirme

Modu kullanırken, bir kayıttaki bir veya daha fazla alanda aşağıdaki sorunlardan biri sebebiyle ayrıştırılamayan verileri yakalamak için kurtarılan veri sütununu etkinleştirebilirsiniz.

• Sağlanan şemada yok.
• Sağlanan şemanın veri türüyle eşleşmiyor.
• Sağlanan şemadaki alan adlarıyla büyük/küçük harf uyuşmazlığı var.

Kurtarılan veri sütunu, kurtarılan sütunları ve kaydın kaynak dosya yolunu içeren bir JSON belgesi olarak döndürülür.

Kurtarılan veri sütununu etkinleştirmek için, okuma sırasında rescuedDataColumn seçeneğini bir sütun adına ayarlayın:

Python

df = spark.read.option("rescuedDataColumn", "_rescued_data").format("csv").load("/Volumes/<catalog>/<schema>/<volume>/reviews_csv")

Scala

val df = spark.read.option("rescuedDataColumn", "_rescued_data").format("csv").load("/Volumes/<catalog>/<schema>/<volume>/reviews_csv")

SQL

SELECT * FROM read_files(
  '/Volumes/<catalog>/<schema>/<volume>/reviews_csv',
  format => 'csv',
  header => true,
  rescuedDataColumn => '_rescued_data'
)

Kurtarılan veri sütunundan kaynak dosya yolunu kaldırmak için şunu ayarlayın:

spark.conf.set("spark.databricks.sql.rescuedDataColumn.filePath.enabled", "false")

CSV ayrıştırıcısı kayıtları ayrıştırırken üç modu destekler: PERMISSIVE, DROPMALFORMEDve FAILFAST. rescuedDataColumn ile birlikte kullanıldığında, veri türü uyuşmazlıkları kayıtların DROPMALFORMED modunda bırakılmasına veya FAILFAST modunda hata oluşturmasına neden olmaz. Yalnızca bozuk kayıtlar (eksik veya hatalı biçimlendirilmiş CSV) bırakılır veya hata oluşturur.

rescuedDataColumn Modda PERMISSIVE kullanıldığında, bozuk kayıtlara aşağıdaki kurallar uygulanır:

• Dosyanın ilk satırı (üst bilgi satırı veya veri satırı) beklenen satır uzunluğunu ayarlar.
• Farklı sayıda sütuna sahip bir satır tamamlanmamış olarak kabul edilir.
• Veri türü uyuşmazlıkları bozuk kayıtlar olarak kabul edilmez.
• Yalnızca eksik ve hatalı biçimlendirilmiş CSV kayıtları bozuk olarak kabul edilir ve veya _corrupt_recordsütununa badRecordsPath kaydedilir.

Ek kaynaklar

• Parquet dosyalarını okuma ve yazma: İş yükünüz daha iyi sorgu performansı veya daha verimli depolama gerektiriyorsa, Parquet'in sütun düzeni CSV'nin düz metin biçimine göre önemli avantajlar sunar.