CSV 파일 읽기 및 쓰기

CSV(쉼표로 구분된 값)는 데이터 교환, ETL 파이프라인 및 범용 데이터 스토리지에 널리 사용되는 일반 텍스트 테이블 형식입니다. Azure Databricks는 Apache Spark에서 CSV 읽기와 쓰기를 모두 지원하며, 스키마 추론, 압축, 잘못된 형식의 레코드 처리 및 구조 복구 데이터를 포함합니다.

사전 요구 사항

Azure Databricks CSV 파일을 사용하기 위해 추가 구성이 필요하지 않습니다. 그러나 CSV 파일을 스트리밍하려면 자동 로더가 필요합니다.

옵션

.option() 및 .options()의 DataFrameReader 및 DataFrameWriter 메서드를 사용하여 CSV 데이터 원본을 구성합니다. 지원되는 옵션의 전체 목록은 CSV 옵션 및 CSV 옵션을 참조DataFrameReader하세요.DataFrameWriter

Usage

다음 예제에서는 CSV 파일을 읽고 쓰고, 스키마를 지정하고, 잘못된 형식의 레코드를 처리하는 방법을 보여 줍니다.

CSV 파일 읽기

다음 예제에서는 Wanderbricks 샘플 데이터 세트를 사용합니다. 검토 데이터를 CSV에 쓴 다음 다시 읽습니다.

파이썬

# 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을 사용하여 CSV 파일 읽기

다음 SQL 예제에서는 read_files을 사용하여 CSV 파일을 읽습니다.

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

스키마 지정

CSV 파일의 스키마가 알려진 경우 schema 옵션을 사용하여 CSV 판독기에서 원하는 스키마를 지정할 수 있습니다.

파이썬

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

열 하위 집합 읽기

CSV 파서의 동작은 읽는 열에 따라 달라집니다. 지정된 스키마가 파일 레이아웃과 일치하지 않으면 액세스되는 열에 따라 결과가 크게 다를 수 있습니다. CSV에는 열 이름 메타데이터가 없으므로 Spark는 스키마 필드를 위치별로 열에 매핑합니다. 일치하지 않는 스키마는 값을 잘못된 필드로 이동합니다.

파이썬

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

잘못된 형식의 CSV 레코드 처리

지정된 스키마를 사용하여 CSV 파일을 읽을 때 파일의 데이터가 스키마와 일치하지 않을 수 있습니다. 예를 들어 도시 이름이 포함된 필드는 정수로 변환될 수 없습니다. 결과는 파서가 실행되는 모드에 따라 다릅니다.

• PERMISSIVE (기본값): 올바르게 구문 분석할 수 없는 필드에 null이 삽입됩니다.
• DROPMALFORMED: 구문 분석할 수 없는 필드가 포함된 행 삭제
• FAILFAST: 잘못된 데이터가 발견되면 읽기를 중단합니다.

모드를 설정하려면 mode 옵션을 사용합니다.

파이썬

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 모드에서는 다음 방법 중 하나를 사용하여 올바르게 구문 분석할 수 없는 행을 검사할 수 있습니다.

• 파일에 손상된 레코드를 기록하기 위해 옵션 badRecordsPath에 대한 사용자 지정 경로를 제공할 수 있습니다.
• DataFrameReader에 제공된 스키마에 열 _corrupt_record 추가하여 결과 DataFrame의 손상된 레코드를 검토할 수 있습니다.

_corrupt_record를 사용해 형식이 잘못된 행을 검사하려면, 이를 스키마에 추가하고 null이 아닌 값을 기준으로 필터링합니다.

파이썬

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

복구된 데이터 열 활성화

이 PERMISSIVE 모드를 사용하는 경우, 레코드의 하나 이상의 필드에 다음 문제 중 하나가 있어 구문 분석되지 않은 데이터를 캡처하는 복구 데이터 열을 사용하도록 설정할 수 있습니다.

• 제공된 스키마에 누락되었습니다.
• 제공된 스키마의 데이터 형식과 일치하지 않습니다.
• 제공된 스키마의 필드 이름과 대소문자가 일치하지 않는 부분이 있습니다.

구조된 데이터 열은 구조된 열과 레코드의 원본 파일 경로를 포함하는 JSON 문서로 반환됩니다.

복구 데이터 열을 사용 설정하려면 읽을 때 rescuedDataColumn 옵션을 열 이름으로 설정하세요:

파이썬

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

복구된 데이터 열에서 원본 파일 경로를 제거하려면 다음을 설정합니다.

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

CSV 파서는 레코드를 구문 분석할 때 PERMISSIVE, DROPMALFORMED 및 FAILFAST 등, 세 가지 모드를 지원합니다. rescuedDataColumn과 함께 사용하면 데이터 형식 불일치로 인해 DROPMALFORMED 모드에서 레코드가 삭제되거나 FAILFAST 모드에서 오류가 throw되지 않습니다. 손상된 레코드, 즉 불완전하거나 형식이 잘못된 CSV만 삭제되거나 오류가 발생합니다.

rescuedDataColumn이 PERMISSIVE 모드에서 사용되면 다음 규칙이 손상된 레코드에 적용됩니다.

• 파일의 첫 번째 행(헤더 행 또는 데이터 행)은 예상되는 행 길이를 설정합니다.
• 열 수가 다른 행은 불완전한 것으로 간주됩니다.
• 데이터 형식 불일치는 손상된 레코드로 간주되지 않습니다.
• 불완전하고 형식이 잘못된 CSV 레코드만 손상된 것으로 간주되어 _corrupt_record 열 또는 badRecordsPath기록됩니다.

추가 리소스

• Parquet 파일 읽기 및 쓰기: 워크로드에 더 나은 쿼리 성능 또는 보다 효율적인 스토리지가 필요한 경우 Parquet의 열 형식 레이아웃은 CSV의 일반 텍스트 형식에 비해 상당한 이점을 제공합니다.