[SparkSQL] CSV DataFrameReader 알아보기

- 목차

 

들어가며.

이번 글에서는 SparkSQL 의 csv DataFrameReader 와 관련된 여러가지 옵션들에 대해서 알아보는 시간을 가지려고 합니다.

CSV 파일은 흔히 사용되는 파일 형식임에도 불구하고 데이터 타입에 대한 정확한 규격이 존재하지 않습니다.

공백으로 표현되는 빈 값, Null or None 데이터의 처리, 숫자 데이터와 날짜 데이터의 변환 등

SparkSQL 의 CSV Reader 옵션들을 살펴보고 차질없이 활용할 수 있도록 정리해보겠습니다.

 

저는 SparkSQL 2.4 버전 이후의 라이브러리를 사용하구요.

Kaggle 의 Insurance 와 관련된 CSV 형식의 데이터를 토대로 CSV DataFrameReader 에 대해서 알아보도록 하겠습니다.

 

https://www.kaggle.com/code/mariapushkareva/medical-insurance-cost-with-linear-regression/input?select=insurance.csv

Medical Insurance Cost with Linear Regression

 

데이터 살펴보기.

먼저 간단하게 Kaggle Insurance 데이터의 스키마와 칼럼의 정보를 살펴보겠습니다.

 

데이터의 내용은 아래와 같습니다.

당뇨 환자들의 메타데이터들로 구성되구요.

성별과 나이 같은 신체정보와 자녀의 수, 지역 등의 정보와 의료비로 사용한 charges 정보가 존재합니다.

해당 데이터는 Regression 을 위한 데이터이기 때문에 환자에 대한 Feature 들과 의료비라는 Label 로 구성됩니다.

+---+------+------+--------+------+---------+-----------+
|age|   sex|   bmi|children|smoker|   region|    charges|
+---+------+------+--------+------+---------+-----------+
| 19|female|  27.9|       0|   yes|southwest|  16884.924|
| 18|  male| 33.77|       1|    no|southeast|  1725.5523|
| 28|  male|    33|       3|    no|southeast|   4449.462|
| 33|  male|22.705|       0|    no|northwest|21984.47061|
| 32|  male| 28.88|       0|    no|northwest|  3866.8552|
+---+------+------+--------+------+---------+-----------+
only showing top 5 rows

 

Schema 는 아래와 같습니다.

root
 |-- age: integer (nullable = true)
 |-- sex: string (nullable = true)
 |-- bmi: double (nullable = true)
 |-- children: integer (nullable = true)
 |-- smoker: string (nullable = true)
 |-- region: string (nullable = true)
 |-- charges: double (nullable = true)

 

inferSchema.

이제 본격적으로 CSV 를 읽어들이기 위한 DataFrameReader 의 옵션에 대해서 알아보겠습니다.

inferSchema 옵션은 CSV 의 데이터 일부분을 조회하여 데이터들의 타입을 분석합니다.

그리고 분석된 데이터를 통해서 Schema 를 추론하고, 생성되는 DataFrame 의 Schema 로써 활용합니다.

 

저의 경우에는 csv 파일이 files/insurance.csv 파일에 위치하며, inferSchema 옵션을 활성화하였습니다.

val insuranceFile = getClass.getClassLoader.getResource("files/insurance.csv").getFile
val df = spark.read.format("csv")
  .option("inferSchema", value = true)
  .option("header", value = true)
  .load(insuranceFile)
df.printSchema()

 

위 프로그램은 아래의 Schema 를 출력합니다.

root
 |-- age: integer (nullable = true)
 |-- sex: string (nullable = true)
 |-- bmi: double (nullable = true)
 |-- children: integer (nullable = true)
 |-- smoker: string (nullable = true)
 |-- region: string (nullable = true)
 |-- charges: double (nullable = true)

 

만약 inferSchema 옵션이 비활성화된다면 아래처럼 모든 데이터의 타입이 string 타입으로 적용됩니다.

root
 |-- age: string (nullable = true)
 |-- sex: string (nullable = true)
 |-- bmi: string (nullable = true)
 |-- children: string (nullable = true)
 |-- smoker: string (nullable = true)
 |-- region: string (nullable = true)
 |-- charges: string (nullable = true)

 

 

header.

csv 파일은 첫번째 라인에 Column 들을 명시합니다.

반면, CSV 파일은 명확한 규격이 존재하지 않기 때문에 첫번째 라인이 Column 정보들로 구성되지 않을 수 있습니다.

예를 들어, Column 정보가 첫번째 라인에 존재하지 않는 CSV 파일을 읽어보겠습니다.

 

< noHeaderInsurance.csv >

19,female,27.9,0,yes,southwest,16884.924
18,male,33.77,1,no,southeast,1725.5523
28,male,33,3,no,southeast,4449.462
33,male,22.705,0,no,northwest,21984.47061
32,male,28.88,0,no,northwest,3866.8552

 

첫번째 라인에 Column 정보가 없는 CSV 파일을 header True 옵션으로 읽어들이게되면,

아래 출력결과와 같이 첫번째 Row 가 Header 가 됩니다.

이러한 경우에는 header 옵션을 비활성화시켜야합니다.

    val insuranceFile = getClass.getClassLoader.getResource("files/noHeaderInsurance.csv").getFile
    val df = spark.read.format("csv")
      .option("header", value = true)
      .load(insuranceFile)
    df.show(5)

+---+------+------+---+---+---------+-----------+
| 19|female|  27.9|  0|yes|southwest|  16884.924|
+---+------+------+---+---+---------+-----------+
| 18|  male| 33.77|  1| no|southeast|  1725.5523|
| 28|  male|    33|  3| no|southeast|   4449.462|
| 33|  male|22.705|  0| no|northwest|21984.47061|
| 32|  male| 28.88|  0| no|northwest|  3866.8552|
+---+------+------+---+---+---------+-----------+

 

이러한 경우에는 아래와 같이 header 를 비활성화시키고, Column 이름을 재할당해 줄 필요가 있습니다.

 

    val insuranceFile = getClass.getClassLoader.getResource("files/noHeaderInsurance.csv").getFile
    val df = spark.read.format("csv")
      .load(insuranceFile)
    val renamed_df = df.withColumnRenamed("_c0", "age")
      .withColumnRenamed("_c1", "sex")
      .withColumnRenamed("_c2", "bmi")
      .withColumnRenamed("_c3", "children")
      .withColumnRenamed("_c4", "smoker")
      .withColumnRenamed("_c5", "region")
      .withColumnRenamed("_c6", "charges")

    renamed_df.show(5)

 

 

emptyValue.

Empty Value 는 빈 문자열입니다.

비어있다라는 표현은 참 애매합니다.

"" 이것도 비어있고, '' 이것도 비어있고, 아예 값이 없는 경우도 있습니다.

그래서 Empty Value 를 어떻게 처리할지 결정하기 위해서 빈 문자열에 대한 정의가 필요합니다.

 

먼저 아래의 경우를 보겠습니다.

3개의 Record 들이 있구요. 5개의 Column 으로 구성됩니다.

그리고 모두 세번째 자리의 값은 비어두었습니다.

아예 비워두거나, Single & Double Quote 를 사용하여 빈 문자열을 만들어보았습니다.

1,2,,4,5   // 세번째 값이 없음.

1,2,'',4,5 // 세번째 값이 ''.

1,2,"",4,5 // 세번째 값이 "".

 

이 경우에 Single & Double Quote 로 감싸진 빈 문자열이 Empty Value 의 대상이 됩니다.

그리고 emptyValue 옵션을 통해서 Empty Value 를 어떻게 처리할지 결정할 수 있죠.

 

아래의 CSV 파일은 sex Column 의 값으로 Empty Value 의 세가지 케이스를 지정하였습니다.

single quote, double quote, 그리고 값이 없는 상태입니다.

age,sex,bmi,children,smoker,region,charges
46,'',33.44,1,no,southeast,8240.5896
37,"",27.74,3,no,northwest,7281.5056
'',,29.83,2,no,northeast,6406.4107
None,female,25.84,0,no,null,28923.13692
25,male,26.22,0,no,northeast,2721.3208

 

quote 가 ' 인 경우.

아래의 코드는 emptyValue 옵션을 통해서 Empty Value 를 "EMPTY" 라는 문자로 변경함을 설정합니다.

그리고 quote 라는 옵션을 사용하여 Empty Value 의 구체적인 조건을 명시합니다.

single quote 로 감싸진 문자를 Empty Value 로 간주하겠다는 의미입니다.

val schema = "age INTEGER, sex STRING, bmi DOUBLE, children INTEGER, smoker STRING, region STRING, charges DOUBLE"
val df = spark.read.format("csv")
  .schema(schema)
  .option("header", value = true)
  .option("quote", value = "'")
  .option("emptyValue", value="EMPTY")
  .load(insuranceFile)
df.show(5)

+----+------+-----+--------+------+---------+-----------+
| age|   sex|  bmi|children|smoker|   region|    charges|
+----+------+-----+--------+------+---------+-----------+
|  46| EMPTY|33.44|       1|    no|southeast|  8240.5896|
|  37|    ""|27.74|       3|    no|northwest|  7281.5056|
|null|  null|29.83|       2|    no|northeast|  6406.4107|
|null|female|25.84|       0|    no|     null|28923.13692|
|  25|  male|26.22|       0|    no|northeast|  2721.3208|
+----+------+-----+--------+------+---------+-----------+

 

결과적으로 '' (Single Quote) 로 감싸진 빈 문자열이 EMPTY 로 치환됩니다.

 

quote 가 " 인 경우.

quote 옵션의 기본값은 " 입니다.

그래서 기본적으로 "" 만을 Empty Value 라고 인식합니다.

아래의 코드를 실행시키게 되면, "" 인 문자열만이 EMPTY 로 치환됨을 확인할 수 있습니다.

val schema = "age INTEGER, sex STRING, bmi DOUBLE, children INTEGER, smoker STRING, region STRING, charges DOUBLE"
val df = spark.read.format("csv")
  .schema(schema)
  .option("header", value = true)
  .option("quote", value = "\"")
  .option("emptyValue", value="EMPTY")
  .load(insuranceFile)
df.show(5)

+----+------+-----+--------+------+---------+-----------+
| age|   sex|  bmi|children|smoker|   region|    charges|
+----+------+-----+--------+------+---------+-----------+
|  46|    ''|33.44|       1|    no|southeast|  8240.5896|
|  37| EMPTY|27.74|       3|    no|northwest|  7281.5056|
|null|  null|29.83|       2|    no|northeast|  6406.4107|
|null|female|25.84|       0|    no|     null|28923.13692|
|  25|  male|26.22|       0|    no|northeast|  2721.3208|
+----+------+-----+--------+------+---------+-----------+