소개
Spark 변환을 작성하고 액션을 실행하며 데이터를 저장했습니다. 이제 익숙한 연산을 실행 증거와 연결합니다. 수행할 일을 설명하는 계획과 작업, 스테이지, 태스크를 기록하는 실제 이벤트 로그입니다. 목표는 내부 태스크 수를 외우는 대신 동작을 설명하고 유용한 오류를 읽는 것입니다.
**드라이버(driver)**는 애플리케이션을 실행하고 작업을 조정합니다. 클러스터에서 실행기는 할당된 자원으로 태스크를 실행합니다. local[2]는 컴퓨터 한 대에서 실행 스레드 두 개로 Spark 작업을 실행합니다. 작업(job)은 스테이지(stage)로 나뉠 수 있으며 각 스테이지는 계산 파티션에 태스크(task)를 실행합니다. 특히 쿼리 계획과 적응형 동작 때문에 액션 하나가 여러 작업을 만들 수도 있습니다. 정확한 개수는 달라질 수 있습니다.
이 독립 실습은 정상 주문, 고객, 의도적으로 잘못 만든 표현식 두 개와 읽기 전용 이벤트 요약 도구를 제공합니다. 필터링, 그룹화, 조인과 Parquet 출력을 알아야 합니다. 자원을 튜닝하거나 일반 데이터셋을 Python 반복문으로 수집하거나 클러스터를 배포하지 않습니다. 터미널을 사용하세요.
익숙한 쿼리를 실행 계획과 연결하기
이 단계에서는 결과를 요청하기 전에 날짜별·지역별 결제 완료 판매 쿼리를 확인합니다. 리더, 필터, 금액, 고객 조인과 집계는 앞서 배운 내용입니다.
여기서 groupBy("order_date", "region")은 두 키를 함께 사용합니다. 서로 다른 날짜와 지역 조합마다 그룹 하나가 생깁니다. 10월 14일 East는 10월 15일 East 및 10월 14일 West와 별개입니다. 각 조합 안에서 개수와 합계를 계산합니다. 앞선 일일 질문과 지역 질문을 하나의 보고서로 결합한 것입니다. explain(mode="formatted")는 보고서 행을 반환하는 대신 계획한 연산자를 출력합니다.
SparkSession은 output/spark-events에 실제 이벤트 로그도 활성화합니다. 애플리케이션 종료 후 학습 로그를 읽을 수 있도록 압축을 끕니다. 이 설정은 실행을 기록하며 가짜 작업 기록을 대신 넣지 않습니다. 대상 디렉터리는 준비되어 있습니다.
전체 확인 스크립트를 시작하고 실행합니다.
cd /home/labex/project
cat > /home/labex/project/inspect_execution.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, count, sum as spark_sum, spark_partition_id
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = (
SparkSession.builder.appName("InspectRetailExecution").master("local[2]")
.config("spark.eventLog.enabled", "true")
.config("spark.eventLog.dir", "file:///home/labex/project/output/spark-events")
.config("spark.eventLog.compress", "false")
.config("spark.eventLog.rolling.enabled", "false")
.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),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
matched = paid.alias("o").join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner").select(
col("o.order_date").alias("order_date"), col("c.region").alias("region"), col("o.order_amount").alias("order_amount")
)
report = matched.groupBy("order_date", "region").agg(
count("*").alias("paid_order_count"), spark_sum("order_amount").alias("revenue")
)
print("Report execution plan:")
report.explain(mode="formatted")
print("Report action result:")
report.orderBy("order_date", "region").show(truncate=False)
report.write.mode("overwrite").parquet("/home/labex/project/output/daily-regional-sales")
print("Computation partition IDs represented by source rows:")
orders.select(spark_partition_id().alias("compute_partition")).dropDuplicates().orderBy("compute_partition").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/inspect_execution.py > /home/labex/project/output/execution-plan-and-result.txt
cat /home/labex/project/output/execution-plan-and-result.txt
Report execution plan에서 필터링, 열 선택, 조인, 집계와 계산 파티션 사이의 기록을 옮기는 Exchange의 역할을 찾으세요. 최적화에 따라 이름과 계획 구조는 달라질 수 있습니다. Exchange는 실행 경계이며 새 출력 디렉터리가 아닙니다.
Report action result 아래에는 10월 14일 / East / 1 / 20.00, 10월 14일 / West / 1 / 25.00, 10월 15일 / East / 1 / 50.00, 10월 15일 / West / 1 / 30.00이 예상됩니다. 전체는 여전히 125.00입니다. show와 쓰기는 설명한 작업의 실제 실행을 요청합니다. 계획과 입력 메타데이터 처리에도 준비 작업이 있을 수 있습니다.
마지막 표시는 spark_partition_id로 원본 기록이 있는 계산 파티션 ID를 보여줍니다. dropDuplicates는 관찰한 ID를 한 번씩 나열합니다. ID와 관찰 개수는 달라질 수 있고 빈 파티션은 나타나지 않을 수 있습니다. 실행 중의 이 그룹은 이전 실습에서 partitionBy로 쓴 날짜 이름 디렉터리와 다릅니다. 하나는 작업 구성이고 다른 하나는 저장 데이터 구조입니다.
실제 작업, 스테이지와 태스크 이벤트 읽기
이 단계에서는 방금 실행한 애플리케이션이 만든 이벤트 로그를 확인합니다. 제공된 도구는 해당 디렉터리의 JSON 이벤트 기록을 읽고 실제 이벤트를 요약합니다. 로그 파서를 작성하거나 버전별 파일 이름을 선택할 필요는 없습니다.
이벤트 디렉터리를 입력 인수로 제공하여 요약 도구를 실행합니다. 셸의 >는 검토할 출력을 보관합니다.
bigdata-python /home/labex/project/tools/summarize-events.py /home/labex/project/output/spark-events > /home/labex/project/output/event-summary.txt
cat /home/labex/project/output/event-summary.txt
실제 애플리케이션 식별자, 양수인 Jobs started, Stages submitted, Tasks ended 개수와 작업에서 스테이지로의 참조가 예상됩니다. 숫자는 이번 실행에 따라 정해지며 최적화, 파일 분할과 Spark 버전에 따라 달라질 수 있습니다. 정확히 재현할 목표가 아니라 관찰된 증거입니다.
액션은 답을 요청하며 Spark는 필요한 일을 여러 작업으로 구성할 수 있습니다. 작업은 스테이지를 참조하고 스테이지는 작업 파티션용 태스크를 포함합니다. 셔플 경계가 스테이지를 나눌 수 있고 여러 변환이 스테이지 하나로 결합될 수도 있습니다. local[2]는 동시 실행 스레드를 제한하며 애플리케이션마다 작업, 스테이지나 태스크가 정확히 둘이라는 뜻은 아닙니다.
작은 데이터에서는 계산 자체에 비해 JVM 시작, 스케줄링과 파일 처리가 두드러질 수 있습니다. 작은 시간 차이로 신뢰할 만한 튜닝 결론을 낼 수는 없습니다. 먼저 계획과 이벤트 증거로 작업을 이해하세요.
알 수 없는 열을 진단하고 수정하기
이 단계에서는 원인을 수정하기 전에 의도적인 이름 해석 오류를 읽습니다. 준비된 잘못된 예제를 확인합니다.
cat /home/labex/project/tools/broken-column.py
스키마는 quantity를 선언하지만 마지막 select는 quanity를 요청합니다. 실제 실패 출력을 저장합니다. >는 일반 출력을 로그 파일에 보내고 2>&1은 표준 오류를 같은 곳으로 보내므로 traceback도 저장됩니다. 다음 명령은 실패하는 것이 정상입니다.
bigdata-python /home/labex/project/tools/broken-column.py > /home/labex/project/output/column-error.log 2>&1
tail -n 20 /home/labex/project/output/column-error.log
해석하지 못한 열 이름과 제안된 유효 이름을 찾으세요. 긴 traceback에는 구현 세부 사항이 있지만 열 참조가 수정할 부분을 가리킵니다. 입력을 재설치하거나 Python 반복문으로 바꿀 필요는 없습니다.
비교할 수 있게 잘못된 학습 예제를 두고 선언된 열 이름으로 수정 프로그램을 만듭니다.
cat > /home/labex/project/fixed_column.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("OrderExpressionExample").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),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
fixed = orders.select("order_id", "quantity")
fixed.orderBy("order_id").show(truncate=False)
fixed.write.mode("overwrite").parquet("/home/labex/project/output/fixed-quantities")
spark.stop()
PYTHON
bigdata-python /home/labex/project/fixed_column.py > /home/labex/project/output/fixed-column-result.txt
cat /home/labex/project/output/fixed-column-result.txt
수량이 2, 1, 1, 2, 1, 2인 ID 여섯 개가 예상됩니다. 이제 열 선택이 실제 스키마에서 해석되고 실제 결과가 저장됩니다. 철자를 포함한 열 이름은 데이터 약속의 일부입니다.
불리언 조건이 아닌 필터 수정하기
이 단계에서는 유효한 정수 열과 유효한 필터 조건을 구분합니다. 두 번째 준비된 예제를 확인합니다.
cat /home/labex/project/tools/broken-filter.py
filter는 각 행의 불리언 표현식을 기대합니다. col("quantity")만 있으면 참·거짓 검사가 아니라 정수입니다. 의도적인 자료형 불일치를 저장합니다. 다음 명령은 실패하는 것이 정상입니다.
bigdata-python /home/labex/project/tools/broken-filter.py > /home/labex/project/output/filter-error.log 2>&1
tail -n 20 /home/labex/project/output/filter-error.log
필터 조건과 불리언이 아닌 정수 자료형을 찾으세요. 이미 배운 열 연산자로 의도한 비교 quantity > 0을 표현하면 됩니다. 수정한 전체 프로그램을 실행합니다.
cat > /home/labex/project/fixed_filter.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("OrderExpressionExample").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),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
fixed = orders.filter(col("quantity") > 0)
fixed.orderBy("order_id").show(truncate=False)
fixed.write.mode("overwrite").parquet("/home/labex/project/output/fixed-orders")
spark.stop()
PYTHON
bigdata-python /home/labex/project/fixed_filter.py > /home/labex/project/output/fixed-filter-result.txt
cat /home/labex/project/output/fixed-filter-result.txt
정상 예제 여섯 행 모두 수량이 양수이므로 모두 남습니다. 음수나 0 수량은 이 조건을 통과하지 못합니다. 이번 수정은 별도 결제 완료 판매 규칙을 적용하지 않습니다.
이제 세 수준의 증거가 있습니다. 의도한 표현식과 스키마, 실제 계획 및 실행 작업, 집중적인 수정 후 올바르게 저장된 결과입니다. 실행을 설명할 때 저장된 날짜 디렉터리와 계산 파티션을 계속 구분하세요. 다음에는 명시적 입력 및 출력 매개변수로 독립 배치 스크립트를 제출합니다.
요약
익숙한 변환을 실제 계획 및 이벤트 로그와 연결하고 고정된 개수를 요구하지 않고 애플리케이션·작업·스테이지·태스크 증거를 관찰했으며 계산 파티션과 날짜 디렉터리를 구분했습니다. 스키마와 오류 증거로 미해결 열과 불리언이 아닌 필터를 수정했습니다. 다음에는 명시적 제출 매개변수로 배치 애플리케이션을 이전 프로세스와 독립적으로 만듭니다.
