Odczytywanie i zapisywanie plików CSV

CSV (wartości rozdzielane przecinkami) to format tabelaryczny w postaci zwykłego tekstu powszechnie używany do wymiany danych, potoków ETL i magazynu danych ogólnego przeznaczenia. Azure Databricks obsługuje format CSV zarówno do odczytu, jak i zapisu przy użyciu Apache Spark, w tym automatyczne wnioskowanie schematu, kompresję, obsługę nieprawidłowo sformatowanych rekordów oraz dane odzyskane.

Prerequisites

Azure Databricks nie wymaga dodatkowej konfiguracji do korzystania z plików CSV. Jednak w celu przesyłania strumieniowego plików CSV potrzebne jest automatyczne ładowanie.

Opcje

.option() Użyj metod .options() i DataFrameReader , DataFrameWriter aby skonfigurować źródła danych CSV. Aby uzyskać pełną listę obsługiwanych opcji, zobacz DataFrameReader Opcje csv i DataFrameWriter opcje CSV.

Usage

W poniższych przykładach pokazano odczytywanie i zapisywanie plików CSV, określanie schematów i obsługę nieprawidłowo sformułowanych rekordów.

Odczytywanie plików CSV

W poniższym przykładzie użyto przykładowego zestawu danych Wanderbricks. Zapisuje dane przeglądów w pliku CSV, a następnie odczytuje je z powrotem.

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)

Odczytywanie plików CSV przy użyciu języka SQL

Poniższy przykład SQL odczytuje plik CSV przy użyciu polecenia 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')

Określanie schematu

Gdy jest znany schemat pliku CSV, można określić żądany schemat dla czytnika CSV z opcją 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'
)

Odczytywanie podzbioru kolumn

Zachowanie analizatora CSV zależy od tego, które kolumny są odczytywane. Jeśli określony schemat nie jest zgodny z układem pliku, wyniki mogą się znacznie różnić w zależności od tego, do których kolumn uzyskuje się dostęp. Plik CSV nie zawiera metadanych nazwy kolumn, dlatego platforma Spark mapuje pola schematu na kolumny według pozycji — niedopasowany schemat przenosi wartości do nieprawidłowych pól.

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

Obsługa nieprawidłowo sformułowanych rekordów CSV

Podczas odczytywania plików CSV z określonym schematem możliwe jest, że dane w plikach nie są zgodne ze schematem. Na przykład pole zawierające nazwę miasta nie zostanie zinterpretowane jako liczba całkowita. Konsekwencje zależą od trybu, w jaki działa analizator:

• PERMISSIVE (ustawienie domyślne): wartości null są wstawiane dla pól, których nie można poprawnie przeanalizować
• DROPMALFORMED: usuwa wiersze zawierające pola, których nie można przeanalizować
• FAILFAST: przerywa odczytywanie w przypadku znalezienia źle sformułowanych danych

Aby ustawić tryb, użyj mode opcji .

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

W trybie PERMISSIVE można sprawdzić wiersze, których nie można poprawnie przeanalizować przy użyciu jednej z następujących metod:

• Możesz podać niestandardową ścieżkę do opcji badRecordsPath rejestrowania uszkodzonych rekordów w pliku.
• Możesz dodać kolumnę _corrupt_record do schematu dostarczonego do komponentu DataFrameReader, aby przejrzeć uszkodzone rekordy w wynikowej ramce danych.

Aby sprawdzić nieprawidłowo sformatowane wiersze za pomocą _corrupt_record, dodaj je do schematu i przefiltruj według wartości różnych od 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

Włącz odzyskaną kolumnę danych

Podczas korzystania z trybu PERMISSIVE można włączyć zapasową kolumnę danych, aby przechwycić wszystkie niezinterpretowane dane, które nie zostały przeanalizowane, ponieważ przynajmniej jedno z pól w rekordzie ma jeden z następujących problemów:

• Brak podanego schematu.
• Nie jest zgodny z typem danych podanego schematu.
• Ma niezgodność pisowni liter z nazwami pól w podanym schemacie.

Uratowana kolumna danych jest zwracana jako dokument JSON zawierający kolumny, które zostały uratowane, oraz ścieżkę pliku źródłowego rekordu.

Aby włączyć kolumnę odzyskanych danych, ustaw opcję rescuedDataColumn na nazwę kolumny podczas wczytywania:

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

Aby usunąć ścieżkę pliku źródłowego z uratowanej kolumny danych, ustaw:

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

Analizator CSV obsługuje trzy tryby podczas analizowania rekordów: PERMISSIVE, DROPMALFORMEDi FAILFAST. W przypadku użycia razem z elementem rescuedDataColumn niezgodność typów danych nie powoduje odrzucenia rekordów w trybie DROPMALFORMED lub zgłaszania błędu w trybie FAILFAST. Tylko uszkodzone rekordy, czyli niekompletne lub źle sformułowane pliki CSV, zostają odrzucone lub zgłaszają błędy.

Gdy rescuedDataColumn jest używane w trybie PERMISSIVE, stosuje się następujące reguły dotyczące uszkodzonych rekordów:

• Pierwszy wiersz pliku (wiersz nagłówka lub wiersz danych) ustawia oczekiwaną długość wiersza.
• Wiersz, który ma różną liczbę kolumn, jest uznawany za niekompletny.
• Niezgodność typów danych nie jest uznawana za uszkodzone rekordy.
• Tylko niekompletne i źle sformułowane rekordy CSV są uznawane za uszkodzone i rejestrowane w kolumnie _corrupt_record lub badRecordsPath.

Dodatkowe zasoby

• Odczytywanie i zapisywanie plików Parquet: Jeśli Twój proces roboczy wymaga większej wydajności zapytań lub bardziej efektywnego przechowywania danych, kolumnowy układ danych w formacie Parquet oferuje istotne zalety w porównaniu z tekstowym formatem CSV.