소개
이제 소매업체의 주문 기록은 Python 리스트에 직접 쓰는 대신 CSV 파일로 도착합니다. 파일에는 2, 10.00 같은 텍스트가 저장되지만 계산하려면 그것이 텍스트인지, 정수 수량인지, 십진수 가격인지 알아야 합니다. 올바른 문자를 읽는 것은 데이터를 올바르게 읽는 일의 일부일 뿐입니다.
Spark의 기본 문자열 읽기, 자동 자료형 추론과 명시적 스키마를 비교합니다. 그다음 필드 순서가 같은 두 번째 파일에 같은 스키마를 적용합니다. 열과 값 사이의 약속에 집중하도록 이번 업무 데이터는 의도적으로 올바르게 준비했습니다. 누락되거나 잘못된 입력은 이후 실습에서 명시적으로 처리합니다.
환경에는 PySpark와 독립적인 로컬 CSV 예제 두 개가 제공됩니다. 첫 실습의 SparkSession, DataFrame 표시, 스키마와 행 수 및 기본 Python을 이해하면 됩니다. pandas, Hadoop, Hive, 설치나 이전 VM 상태는 필요하지 않습니다. 터미널을 사용하세요.
헤더를 읽고 문자열 열 관찰하기
이 단계에서는 CSV 필드 위치를 Spark가 만드는 테이블과 연결합니다. 실제 파일부터 확인합니다.
cd /home/labex/project
cat /home/labex/project/data/orders.csv
첫 줄은 일곱 열의 이름이고 다음 여섯 줄은 주문입니다. 각 행은 상품 유형 하나를 포함하는 주문 한 건입니다. 파일에는 결제 완료, 대기와 환불 상태가 있습니다. 이번 실습에서는 판매 규칙을 적용하지 않고 모두 읽습니다.
spark.read는 DataFrame 리더를 만듭니다. option("header", True)는 첫 CSV 줄을 주문 기록 대신 열 이름으로 처리하도록 리더에 알립니다. csv(input_path)는 로컬 파일을 지정합니다. 자료형 추론이나 제공된 스키마가 없으면 CSV 리더는 문자열 열을 만듭니다.
전체 프로그램을 저장하고 실행합니다. 따옴표로 묶인 PYTHON 히어 도큐먼트는 안의 소스를 보존하고 >는 로컬 파일을 씁니다. bigdata-python은 준비된 PySpark 환경을 사용합니다.
cat > /home/labex/project/read_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("ReadOrderCSV").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
input_path = "/home/labex/project/data/orders.csv"
df = spark.read.option("header", True).csv(input_path)
df.printSchema()
df.orderBy("order_id").show(truncate=False)
print("Rows:", df.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_orders.py > /home/labex/project/output/string-orders.txt
cat /home/labex/project/output/string-orders.txt
quantity와 unit_price를 포함해 문자열 열 일곱 개와 여섯 행이 예상됩니다. 표시된 2가 숫자처럼 보여도 printSchema는 실제 문자열 자료형을 식별합니다. orderBy("order_id")는 예측 가능하게 살펴보도록 표시 행을 식별자로 정렬합니다. 정렬과 자료형 지정은 별개입니다.
header=True는 헤더가 일곱 번째 기록이 되는 것을 막았습니다. 가격이 정확한 십진수여야 한다거나 날짜 연산을 지원해야 한다고 Spark에 알려주지는 않았습니다.
자료형 추론과 명시적 데이터 약속 비교하기
이 단계에서는 각 CSV 필드의 의미를 선언합니다. inferSchema=True는 Spark에 값을 살펴보고 자료형을 추론하도록 요청합니다. 편리할 수 있지만 결과는 관찰한 데이터에 따라 달라지며 가격이나 향후 파일에 대한 지속적인 약속은 아닙니다. 이 예제에서 추론은 unit_price를 부동 소수점 수인 double로 식별합니다.
의도한 약속에는 명시적 스키마를 사용합니다. StructType은 순서 있는 필드 모음입니다. 각 StructField는 이름, 자료형과 누락값 허용을 뜻하는 True를 포함합니다. StringType은 식별자와 상태 텍스트, IntegerType은 수량, DecimalType(10, 2)는 전체 열 자리와 소수점 아래 두 자리의 정확한 가격, DateType은 달력 날짜를 담습니다. NULL 허용 필드라고 해서 누락된 업무 정보가 자동으로 허용되는 것은 아닙니다.
프로그램을 다음 버전으로 바꿉니다. 비교용 추론 자료형 설명을 출력한 뒤 같은 파일을 명시적 스키마로 읽습니다.
cat > /home/labex/project/read_orders.py <<'PYTHON'
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("ReadOrderCSV").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
StructField("order_id", StringType(), True),
StructField("customer_id", StringType(), True),
StructField("product_id", StringType(), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_status", StringType(), True),
StructField("order_date", DateType(), True),
])
input_path = "/home/labex/project/data/orders.csv"
inferred = spark.read.option("header", True).option("inferSchema", True).csv(input_path)
print("Inferred schema:", inferred.schema.simpleString())
df = (
spark.read
.option("header", True)
.option("dateFormat", "yyyy-MM-dd")
.schema(order_schema)
.csv(input_path)
)
print("Explicit schema:")
df.printSchema()
df.orderBy("order_id").show(truncate=False)
print("Rows:", df.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_orders.py > /home/labex/project/output/typed-orders.txt
cat /home/labex/project/output/typed-orders.txt
추론 요약 아래에서 명시적 스키마를 찾으세요. quantity는 integer, unit_price는 decimal(10,2), order_date는 date여야 합니다. 여섯 기록은 그대로 있어야 합니다. 명시적인 dateFormat 옵션은 입력 날짜를 연-월-일로 설명하며 2026-10-14 같은 값과 일치합니다.
스키마의 필드 순서는 이 CSV의 필드 순서와 같습니다. 스키마 제공은 임의의 입력 파일을 헤더 이름에 따라 재배열하라는 일반 지시가 아닙니다. 순서를 아는 이번 데이터에서는 열과 자료형의 의도를 명확히 하는 약속입니다. 또한 스키마가 데이터 품질 정책을 대신하지는 않습니다. 이후 잘못된 값을 안전하게 처리하는 방법을 배웁니다.
같은 스키마를 새 데이터에 적용하기
이 단계에서는 두 번째 정상 파일로 약속을 테스트합니다. 먼저 살펴봅니다.
cat /home/labex/project/data/new-orders.csv
같은 순서의 일곱 필드와 새 주문 두 건이 있습니다. 무엇이 바뀔지 예상해 보세요. 업무 행은 여섯 대신 둘이 되고 수량, 가격과 날짜는 선언한 자료형을 유지합니다.
다음 전체 버전은 명시적 스키마를 유지하고 입력 경로를 바꿉니다. 일반적인 읽기에는 추론 비교가 더 이상 필요하지 않습니다.
cat > /home/labex/project/read_orders.py <<'PYTHON'
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("ReadOrderCSV").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
StructField("order_id", StringType(), True),
StructField("customer_id", StringType(), True),
StructField("product_id", StringType(), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_status", StringType(), True),
StructField("order_date", DateType(), True),
])
input_path = "/home/labex/project/data/new-orders.csv"
df = (
spark.read
.option("header", True)
.option("dateFormat", "yyyy-MM-dd")
.schema(order_schema)
.csv(input_path)
)
print("Explicit schema:")
df.printSchema()
df.orderBy("order_id").show(truncate=False)
print("Rows:", df.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_orders.py > /home/labex/project/output/new-typed-orders.txt
cat /home/labex/project/output/new-typed-orders.txt
o7은 수량 2, 가격 12.50, 상태 paid, 날짜 2026-10-16이고 o8은 수량 3, 가격 4.00, 상태 pending, 날짜 2026-10-17이어야 합니다. 개수는 2여야 합니다. printSchema에서 같은 integer, decimal과 date 자료형을 확인하세요.
원래 여섯 주문 파일은 바뀌지 않았습니다. 재사용 가능한 스키마를 사용하면 수량이나 가격의 의미를 다시 정의하지 않고 새 데이터를 같은 처리 로직에 넣을 수 있습니다. 다음에는 자료형이 있는 열로 주문을 선택하고 각 주문 금액을 계산합니다.
요약
CSV 헤더를 올바르게 읽고 문자열 읽기와 추론을 비교했으며 식별자, 수량, 십진수 가격과 날짜에 대한 명시적 약속을 선언했습니다. 같은 스키마로 두 번째 데이터도 올바르게 해석했습니다. 다음에는 자료형이 있는 열 표현식으로 행을 필터링하고 주문 금액을 계산합니다.
