소개
소매업체가 주문 기록을 받아 유용한 정보로 바꾸려고 합니다. 판매 보고서를 계산하기 전에 행을 표현하고 열을 설명하며 처리 도구에 작업 실행을 요청할 방법이 필요합니다. 눈으로 확인할 수 있을 만큼 작은 기록 세 개로 시작합니다.
**빅데이터(big data)**는 크기, 유입 속도 또는 처리 요구가 편리한 단일 컴퓨터 방식의 한계를 넘어설 수 있는 데이터 작업을 말합니다. Apache Spark는 클러스터의 여러 컴퓨터에 작업을 나눌 수 있는 데이터 처리 엔진입니다. PySpark는 그 Python 인터페이스입니다. DataFrame은 이름과 자료형이 있는 열을 가진 행들을 표현하며 필터링과 집계 같은 연산을 지원합니다. Spark가 데이터를 처리하고 별도 파일이나 저장 시스템이 영속적인 입력과 출력을 제공합니다. 이 과정은 규모를 다루기 전에 정확성을 판단할 수 있도록 작은 예제로 시작합니다.
SparkSession을 만들고 DataFrame을 표시하며 스키마와 행 수를 확인한 뒤 기록 하나를 추가하고 다시 실행합니다. 준비된 환경은 local[2]를 사용합니다. Spark가 이 컴퓨터 한 대에서 실행 스레드 두 개로 작동합니다. 실제 Spark API를 배우되 분산 배포는 별도 주제입니다. 기본 Python 변수, 리스트와 함수 호출을 알아야 하며 이러한 선수 지식에는 Python 경로가 도움이 됩니다. pandas, Hadoop, Hive, Java, 설치나 이전 실습 파일은 필요하지 않습니다. 터미널을 사용하세요.
작은 DataFrame 전체 스크립트 실행하기
이 단계에서는 Python 기록 세 개를 Spark 테이블로 바꾸는 완전한 스크립트를 실행합니다. 작업 공간에서 시작합니다.
cd /home/labex/project
SparkSession은 DataFrame을 만들고 Spark의 구조화 데이터 API에 접근하는 진입점입니다. 빌더는 애플리케이션 이름을 설정하고 로컬 실행을 선택한 뒤 세션을 만듭니다. local[2]는 이 컴퓨터에 실행 스레드 두 개를 허용하며 클러스터 컴퓨터 두 대를 만들지 않습니다. setLogLevel은 이후의 정보 로그를 줄여 학습 출력을 쉽게 찾게 합니다.
Python 소스에서 from pyspark.sql import SparkSession은 설치된 클래스를 프로그램에서 사용할 수 있게 합니다. 점은 이름 있는 멤버에 접근하고 괄호는 메서드를 호출합니다. df.show()는 이 DataFrame에 행 표시를 요청합니다. 빌더는 여러 호출을 연결하며, 바깥 괄호 덕분에 표현식을 여러 줄에 쓸 수 있습니다. getOrCreate()가 세션을 얻기 전까지 각 줄을 설정 하나로 읽으면 됩니다.
orders 리스트에는 괄호로 묶은 값의 집합인 튜플 세 개가 있습니다. 각 튜플은 선언한 열과 같은 순서로 행 하나를 제공합니다. 스키마 문자열은 각 열 이름과 자료형을 지정하며 텍스트 식별자와 상품 뒤에 정수 수량이 옵니다. createDataFrame은 이 기록과 스키마를 사용합니다. show는 Spark에 행 표시를 요청하고 stop은 완료 후 세션을 해제합니다.
전체 스크립트를 저장합니다. 셸의 >는 지정 파일에 쓰고 따옴표로 묶은 PYTHON 표시는 안의 Python 텍스트를 그대로 보존하는 히어 도큐먼트를 시작합니다. 마지막 PYTHON 줄까지 복사하세요.
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
spark.stop()
PYTHON
준비된 bigdata-python 명령은 PySpark가 설치된 Python 환경을 선택합니다. 일반 Python 프로그램처럼 파일을 실행하고 일반 출력을 로컬에 저장합니다.
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/first-run.txt
cat /home/labex/project/output/first-run.txt
order_id, product, quantity 열에 s1/notebook/2, s2/pen/1, s3/folder/3이 있는 테이블이 예상됩니다. 시작 경고나 진행 메시지는 여전히 터미널에 나타날 수 있으며 저장된 테이블과 별개입니다. 명시적 정렬 없이는 행 순서가 보장되지 않으므로 주문 ID로 기록을 식별하세요.
Python 프로그램은 **드라이버(driver)**입니다. 세션을 만들고 Spark에 작업을 요청합니다. 클러스터에서는 실행기(executor) 프로세스가 나뉜 데이터에 태스크를 실행합니다. 여기서는 같은 컴퓨터에서 로컬로 실행합니다. 클러스터를 구성하지 않고 작동하는 프로그램을 만들었습니다.
스키마를 확인하고 행 수 세기
이 단계에서는 열에 무엇이 들어갈 수 있는지와 테이블에 기록이 몇 개 있는지를 구분합니다. **스키마(schema)**는 열 이름, 자료형과 NULL 허용 여부를 나열합니다. 정수 열은 수량 계산에 적합한 정수를 나타내고 텍스트에는 다른 연산을 사용합니다. Nullable은 필드에 누락값을 허용한다는 뜻이며 모든 행에서 값이 없다는 뜻은 아닙니다.
이 설명을 확인하는 printSchema와 행 수를 계산하는 count를 추가합니다. 숨겨진 대화형 상태가 없도록 갱신된 파일 전체를 제공합니다.
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
df.printSchema()
print("Row count:", df.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/schema-and-count.txt
cat /home/labex/project/output/schema-and-count.txt
테이블에는 여전히 기록 세 개가 있어야 합니다. 아래에서 다음을 찾으세요.
root
|-- order_id: string (nullable = true)
|-- product: string (nullable = true)
|-- quantity: integer (nullable = true)
Row count: 3
printSchema는 DataFrame을 설명합니다. show와 count는 Spark에 보이는 결과나 값을 요청하는 **액션(action)**입니다. Spark는 액션이 답을 요청할 때까지 많은 처리를 미룰 수 있습니다. 이후 실습에서는 변환과 계획으로 지연 실행을 설명합니다. 개수 3은 주문 기록 세 개라는 뜻입니다. 수량의 합계는 다른 업무 질문에 대한 답입니다.
각 호출은 새 Python 프로세스를 시작하고 자체 SparkSession을 만듭니다. 파일이 df를 명시적으로 생성하므로 대화형 셸에서 이전에 spark나 df가 만들어졌다는 가정에 의존하지 않습니다.
기록 하나 추가를 예상하고 테스트하기
이 단계에서는 구조를 유지하면서 데이터를 바꿉니다. s4/notebook/1을 추가하면 무엇이 바뀔지 예상해 보세요. 주문 행 수는 네 개가 되고 스키마는 같으며 원래 세 기록은 남습니다. notebook이 다른 주문에 이미 있어도 새 행은 주문 한 건입니다.
해당 기록만 추가하는 다음 버전으로 스크립트를 바꿉니다.
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
("s4", "notebook", 1),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
df.printSchema()
print("Row count:", df.count())
spark.stop()
PYTHON
새 프로세스에서 파일을 다시 실행합니다.
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/expanded-run.txt
cat /home/labex/project/output/expanded-run.txt
네 주문 ID 모두와 Row count: 4를 찾으세요. 스키마는 같은 세 열 이름과 자료형을 설명해야 합니다. 상품 이름이 반복되어도 주문을 자동으로 묶거나 중복을 제거하지 않습니다. 이후 수업에서 그 연산을 명시적으로 요청해야 합니다.
이전 first-run.txt와 schema-and-count.txt는 세 행 버전의 증거로 남습니다. 실행 파일은 이제 네 행 버전을 설명하므로 다른 독립 실행에서도 그 결과를 재현할 수 있습니다. 작은 변경, 예상, 실행, 확인 주기를 완료했습니다. 이 방법은 프로그램이 성공적으로 종료되었다고 믿기만 하는 대신 더 큰 데이터 작업 흐름을 판단하는 데 도움이 됩니다.
요약
자체 SparkSession이 있는 실제 PySpark 스크립트를 실행하고 자료형을 지정한 DataFrame을 만들었으며 표시된 기록, 스키마와 행 수를 확인했습니다. 기록 하나를 추가하니 스키마는 유지되고 행 수가 바뀌었습니다. 다음에는 Python 안에 쓴 기록에서 명시적인 스키마로 읽는 CSV 파일로 넘어갑니다.
