Čtení a zápis souborů CSV

CSV (hodnoty oddělené čárkami) je tabulkový formát prostého textu, který se běžně používá pro výměnu dat, kanály ETL a úložiště dat pro obecné účely. Azure Databricks podporuje formát CSV pro čtení i zápis v Apache Sparku, včetně odvozování schématu, komprese, zpracování chybných záznamů a zachráněných dat.

Poznámka:

Databricks doporučuje read_files funkci s hodnotou tabulky , aby uživatelé SQL mohli číst soubory CSV. read_files je k dispozici ve službě Databricks Runtime 13.3 LTS a vyšší.

Můžete také použít dočasné zobrazení. Pokud používáte SQL k přímému čtení dat CSV bez použití dočasných zobrazení nebo read_files, platí následující omezení:

• Nemůžete zadat možnosti zdroje dat.
• Pro data nelze zadat schéma.

Předpoklady

Azure Databricks nevyžaduje další konfiguraci pro použití souborů CSV. Ke streamování souborů CSV ale potřebujete Auto Loader.

Možnosti

Ke konfiguraci zdrojů dat CSV použijte metody .option() a .options() objektů DataFrameReader a DataFrameWriter. Úplný seznam podporovaných možností naleznete v DataFrameReadermožnostech CSV a DataFrameWritermožnostech CSV.

Usage

Následující příklady ukazují čtení a zápis souborů CSV, zadávání schémat a zpracování poškozených záznamů.

Čtení souborů CSV

Následující příklad používá ukázkovou datovou sadu Wanderbricks . Zapisuje data recenzí do souboru CSV a pak je znovu přečte.

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)

Čtení souborů CSV pomocí SQL

Následující příklad SQL načte soubor CSV pomocí read_files.

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

Zadání schématu

Pokud je známo schéma souboru CSV, můžete pro čtenáře CSV zadat požadované schéma s možností schema.

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

Číst podmnožinu sloupců

Chování analyzátoru CSV závisí na tom, které sloupce se čtou. Pokud zadané schéma neodpovídá rozložení souboru, výsledky se můžou výrazně lišit v závislosti na tom, ke kterým sloupcům se přistupuje. CSV neobsahuje metadata názvů sloupců, takže Spark mapuje pole schématu na sloupce podle jejich pozice – nesoulad schématu způsobí, že se hodnoty posunou do nesprávných polí.

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

Zpracování poškozených záznamů CSV

Při čtení souborů CSV se zadaným schématem je možné, že data v souborech neodpovídají schématu. Například pole obsahující název města nebude parsovat jako celé číslo. Důsledky závisí na režimu, ve kterém se analyzátor spouští:

• PERMISSIVE (výchozí): Hodnoty null se vloží pro pole, která nelze správně analyzovat.
• DROPMALFORMED: zahodí řádky obsahující pole, která nelze analyzovat.
• FAILFAST: Přeruší čtení, pokud jsou nalezena nějaká poškozená data.

Pokud chcete nastavit režim, použijte možnost 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 V režimu je možné zkontrolovat řádky, které nelze správně analyzovat pomocí jedné z následujících metod:

• Můžete zadat vlastní cestu k možnosti badRecordsPath, aby se poškozené záznamy ukládaly do souboru.
• Sloupec _corrupt_record můžete přidat do schématu objektu DataFrameReader, abyste zkontrolovali poškozené záznamy ve výsledném dataframe.

Poznámka:

Tato badRecordsPath možnost má přednost před _corrupt_record, což znamená, že chybně zapsané řádky do zadané cesty se ve výsledném datovém rámci nezobrazí.

Výchozí chování chybně formátovaných záznamů se změní při použití záchranného datového sloupce.

Pokud chcete zkontrolovat poškozené řádky pomocí _corrupt_record, přidejte ho do schématu a vyfiltrujte hodnoty, které nejsou null:

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

Povolení sloupce zachráněných dat

Poznámka:

Tato funkce je podporovaná v Databricks Runtime 8.3 a novějších.

Když použijete PERMISSIVE režim, můžete povolit, aby zachycovaný datový sloupec zaznamenával všechna data, která nebyla analyzována, protože jedno nebo více polí v záznamu má jeden z následujících problémů:

• Chybí ze zadaného schématu.
• Neodpovídá datovému typu zadaného schématu.
• Neshoduje se s názvy polí v zadaném schématu.

Uložený datový sloupec se vrátí jako dokument JSON obsahující sloupce, které byly uloženy, a cestu ke zdrojovému souboru záznamu.

Chcete-li povolit sloupec pro zachráněná data, nastavte při čtení možnost rescuedDataColumn na název sloupce:

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

Pokud chcete odebrat cestu ke zdrojovému souboru ze sloupce zachráněných dat, nastavte:

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

Analyzátor CSV podporuje při analýze záznamů tři režimy: PERMISSIVE, DROPMALFORMEDa FAILFAST. Při použití společně s datovým rescuedDataColumntypem neshody nezpůsobí vyřazení záznamů v DROPMALFORMED režimu nebo vyvolání chyby v FAILFAST režimu. Vyřadí se pouze poškozené záznamy, tj. neúplné nebo chybné CSV, anebo vyvolají chyby.

Pokud rescuedDataColumn se používá v PERMISSIVE režimu, platí pro poškozené záznamy následující pravidla:

• První řádek souboru (řádek záhlaví nebo datový řádek) nastaví očekávanou délku řádku.
• Řádek s jiným počtem sloupců se považuje za neúplný.
• Neshody datových typů se nepovažují za poškozené záznamy.
• Pouze neúplné a poškozené záznamy CSV jsou považovány za poškozené a zaznamenané do _corrupt_record sloupce nebo badRecordsPath.

Dodatečné zdroje

• Čtení a zápis souborů Parquet: Pokud vaše úloha vyžaduje lepší výkon dotazů nebo efektivnější úložiště, sloupcové rozložení Parquet nabízí významné výhody oproti formátu prostého textu CSV.