PySpark에서 주문 선택, 필터링과 금액 계산하기

소개

올바른 자료형의 주문 테이블은 출발점이며 아직 답은 아닙니다. 동료는 일부 열과 결제 완료 주문만 보고 수량과 단가로 계산한 금액을 원할 수 있습니다. 원래 기록을 유지하면서 이런 결정을 Spark DataFrame 연산으로 표현합니다.

**열 표현식(column expression)**은 Spark가 행에 적용할 계산이나 조건을 설명합니다. 필드를 선택하고 조건을 결합하며 십진수 금액을 추가하고 결과 열에 읽기 쉬운 이름을 줍니다. 이 연산들은 새 DataFrame을 만들고 show 같은 액션이 실제 결과를 요청합니다. 이 차이를 알면 아직 스케줄링 내부를 배우지 않고도 Spark의 지연 실행을 이해할 수 있습니다.

이번 실습은 정상 상품 주문 여섯 건을 독립적으로 제공합니다. 이전 실습의 명시적 CSV 스키마와 기본 Python을 이해하면 됩니다. 실행 가능한 각 버전에 전체 리더와 SparkSession이 포함되므로 이전 프로세스나 파일은 필요하지 않습니다. 터미널과 준비된 bigdata-python 환경을 사용하세요.

질문에 필요한 열 선택하기

이 단계에서는 주문 여섯 기록을 모두 유지하면서 테이블 열을 좁힙니다. 원본부터 확인합니다.

cd /home/labex/project

cat /home/labex/project/data/orders.csv

select는 새 DataFrame이 보여주는 열과 순서를 선택합니다. 행을 필터링하지는 않습니다. 아래 전체 스크립트는 익숙한 명시적 리더를 반복한 뒤 주문 식별자, 수량, 가격과 상태를 선택합니다. selected가 보여주지 않아도 날짜와 고객 필드는 df에 남아 있습니다.

따옴표로 묶은 히어 도큐먼트로 스크립트를 저장하고 실행합니다. >는 파일을 쓰거나 이전 버전을 바꾸며 마지막 PYTHON 표시가 소스 텍스트를 끝냅니다.

cat > /home/labex/project/transform_orders.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("TransformOrders").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),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

selected = df.select("order_id", "quantity", "unit_price", "order_status")
selected.orderBy("order_id").show(truncate=False)

spark.stop()
PYTHON

bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/selected-orders.txt

cat /home/labex/project/output/selected-orders.txt

표시된 네 열에 주문 ID 여섯 개 모두가 있어야 합니다. orderBy는 표시를 예측 가능하게 하고 show는 행 반환에 필요한 작업을 시작합니다. 이것은 투영(projection)이라고도 하는 열 선택입니다. 원본 파일을 바꾸거나 주문을 제거하지 않았습니다.

조건을 결합하여 행 선택하기

이 단계에서는 수량이 적어도 둘인 결제 완료 주문을 유지합니다. col("order_status")는 Spark 열을 참조합니다. 이를 == "paid"와 비교하면 Python의 단일 True 또는 False가 아니라 Spark가 사용할 조건을 만듭니다. >=는 수량 조건을 만듭니다.

&로 Spark 열 조건을 AND 의미로 결합합니다. Python 연산자 우선순위가 중요하므로 각 비교를 괄호로 묶으세요. Python의 일반 and는 Spark Column 객체를 결합하는 연산자가 아닙니다.

조건에 맞는 ID를 예상해 보세요. o1, o4, o6이 둘 다 충족합니다. o2는 결제 완료이지만 수량이 하나입니다. 전체 스크립트를 바꾸고 실행합니다.

cat > /home/labex/project/transform_orders.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("TransformOrders").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),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

eligible = df.filter((col("order_status") == "paid") & (col("quantity") >= 2))
eligible.select("order_id", "quantity", "unit_price", "order_status").orderBy("order_id").show(truncate=False)

spark.stop()
PYTHON

bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/eligible-orders.txt

cat /home/labex/project/output/eligible-orders.txt

결과에는 해당 ID 세 개만 있고 수량은 각각 2여야 합니다. filter는 선택 행을 바꾸고 select는 표시 열을 제어합니다. df가 여전히 전체 여섯 행의 원본을 설명하는 동안 두 연산을 결합할 수 있습니다.

이 단계의 수량 기준은 시연 조건이며 판매의 일반 정의가 아닙니다. 다음 계산에서는 결제 완료 주문마다 매출에 기여하므로 o2를 포함한 네 주문 모두를 포함합니다.

십진수 주문 금액을 계산하고 이름 붙이기

이 단계에서는 수량에 십진수 단가를 곱해 각 결제 완료 주문의 금액을 계산합니다. withColumn은 새 DataFrame에 이름 있는 열을 추가하거나 같은 이름이 있으면 그 열을 바꿉니다. 여기서는 Spark 내장 곱셈 표현식으로 order_amount를 계산합니다.

이어지는 select는 주문 ID를 유지하고 alias로 계산 열에 표시 이름 amount를 줍니다. 별칭은 결과 열 이름을 바꾸며 CSV 헤더를 편집하지 않습니다. 결제 완료 주문 네 건 모두 계산에 포함됩니다. 대기와 환불 행은 제외하며 이 단순화된 규칙에서 환불을 음수 매출로 다루지 않습니다.

전체 버전을 실행합니다.

cat > /home/labex/project/transform_orders.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("TransformOrders").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),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

paid_result = (
    df.filter(col("order_status") == "paid")
    .withColumn("order_amount", col("quantity") * col("unit_price"))
    .select("order_id", col("order_amount").alias("amount"))
)
print("Paid orders:")
paid_result.orderBy("order_id").show(truncate=False)
paid_result.printSchema()

spark.stop()
PYTHON

bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/paid-amounts.txt

cat /home/labex/project/output/paid-amounts.txt

o1/20.00, o2/25.00, o4/50.00, o6/30.00이 예상됩니다. 뒤에 0이 더 붙을 수 있습니다. 스키마는 amount를 십진수 자료형으로 식별해야 합니다. Spark는 곱셈 중 십진수 정밀도를 넓힐 수 있습니다. 중요한 점은 가격을 이진 부동 소수점으로 바꾸지 않고 십진수 연산을 유지한다는 것입니다.

filter, withColumn, select, orderBy는 **변환(transformation)**을 설명하고 새 DataFrame을 반환합니다. 이 연결을 정의한다고 모든 중간 테이블이 즉시 실제로 만들어지지는 않습니다. show는 필요한 작업을 실행하여 답을 표시하도록 Spark에 요청하는 액션입니다. printSchema는 열 구조를 설명합니다. 이후 이런 연결 뒤의 실제 실행 계획을 확인합니다.

상태 하나를 바꾸고 영향 예상하기

이 단계에서는 결제 완료 결과를 유지하고 별도의 대기 주문 연습 보기를 추가합니다. 실행 전에 출력을 예상해 보세요. pending은 o3뿐이고 수량 하나에 가격 10.00이므로 금액은 10.00입니다.

새 pending_result는 상태 값만 바꾸어 같은 표현식 연결을 사용합니다. 이미 필터링한 paid_result가 아니라 df에서 시작하므로 대기 기록이 아직 있습니다.

cat > /home/labex/project/transform_orders.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("TransformOrders").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),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

paid_result = (
    df.filter(col("order_status") == "paid")
    .withColumn("order_amount", col("quantity") * col("unit_price"))
    .select("order_id", col("order_amount").alias("amount"))
)
print("Paid orders:")
paid_result.orderBy("order_id").show(truncate=False)
paid_result.printSchema()

pending_result = (
    df.filter(col("order_status") == "pending")
    .withColumn("order_amount", col("quantity") * col("unit_price"))
    .select("order_id", col("order_amount").alias("amount"))
)
print("Pending practice:")
pending_result.orderBy("order_id").show(truncate=False)

spark.stop()
PYTHON

bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/status-comparison.txt

cat /home/labex/project/output/status-comparison.txt

Paid orders 아래에는 같은 네 결제 완료 금액이 남아 있어야 합니다. Pending practice 아래에는 o3/10.00만 예상됩니다. 결제 완료 보기를 pending 상태로 필터링하면 그 기록은 이미 제외되었으므로 행이 나오지 않습니다. 적절한 원본에서 시작하는 것이 중요합니다.

두 연결은 Spark 내장 표현식으로 평가됩니다. 일반 데이터셋을 Python 리스트로 수집하여 반복문으로 처리하지 않았습니다. 따라서 처리 설명을 Spark 실행 엔진에 맞게 유지합니다. 최종 스크립트는 결제 완료 계산과 작은 연습 비교를 유지하고 원래 CSV에는 여전히 모든 주문 여섯 건이 있습니다.

요약

열을 선택하고 Spark 표현식으로 행 조건을 결합했으며 십진수 주문 금액을 계산하고 출력 열에 이름을 붙였습니다. 결제 완료와 대기 보기를 비교하여 변환을 적절한 원본에서 시작해야 하는 이유와 액션이 결과를 요청하는 방식을 확인했습니다. 다음에는 익숙한 읽기와 선택 개념을 JSON 입력에 적용합니다.

Spark 주문 데이터 선택, 필터링과 변환하기

소개

질문에 필요한 열 선택하기

조건을 결합하여 행 선택하기

십진수 주문 금액을 계산하고 이름 붙이기

상태 하나를 바꾸고 영향 예상하기

요약