빅데이터 분석 환경에서 PySpark SQL로 조회한 결과를 Matplotlib를 활용해 그래프로 시각화하는 것은 데이터 인사이트를 파악하는 데 매우 유용합니다. 이 글에서는 PySpark SQL 쿼리 결과를 Pandas 데이터프레임으로 변환한 뒤 Matplotlib으로 플롯하는 전체 과정을 단계별로 살펴보겠습니다.
처리 순서
- 그림 크기(figure size)를 설정하고 서브플롯 주변 및 사이의 여백(padding)을 조정합니다.
- Spark 기능의 기본 진입점(Entry Point) 역할을 하는 SparkContext 인스턴스를 생성합니다.
- Hive에 저장된 데이터와 통합되는 Spark SQL 변형인 HiveContext 인스턴스를 가져옵니다.
- 튜플 형태의 레코드 리스트를 만듭니다.
- 로컬 Python 컬렉션을 분산 처리하여 RDD(Resilient Distributed Dataset)로 변환합니다.
- 리스트의 각 레코드를 DB 스키마 형태(Row 객체)로 매핑합니다.
- 스키마 인스턴스를 생성하여 "my_table" 테이블에 등록합니다.
- SQL 쿼리를 실행해 테이블에서 레코드를 조회합니다.
- 조회된 결과를 Pandas 데이터프레임으로 변환합니다.
- name 컬럼을 인덱스로 지정한 후 데이터를 플롯합니다.
- 그래프를 화면에 표시하려면 show() 메서드를 사용합니다.
예제 코드
from pyspark.sql import Row
from pyspark.sql import HiveContext
import pyspark
import matplotlib.pyplot as plt
plt.rcParams["figure.figsize"] = [7.50, 3.50]
plt.rcParams["figure.autolayout"] = True
sc = pyspark.SparkContext()
sqlContext = HiveContext(sc)
test_list = [(1, 'John'), (2, 'James'), (3, 'Jack'), (4, 'Joe')]
rdd = sc.parallelize(test_list)
people = rdd.map(lambda x: Row(id=int(x[0]), name=x[1]))
schemaPeople = sqlContext.createDataFrame(people)
sqlContext.registerDataFrameAsTable(schemaPeople, "my_table")
df = sqlContext.sql("Select * from my_table")
df = df.toPandas()
df.set_index('name').plot()
plt.show()실행 결과
위 코드를 실행하면 각 이름(id 1~4: John, James, Jack, Joe)에 해당하는 값이 막대그래프로 출력됩니다. x축에는 이름(name)이, y축에는 id 값이 표시됩니다.
핵심 포인트 정리
- toPandas() 변환: PySpark 데이터프레임은 분산 환경에서 동작하기 때문에 Matplotlib으로 바로 그릴 수 없습니다. 반드시
toPandas()메서드를 사용해 로컬 Pandas 데이터프레임으로 변환해야 합니다. - HiveContext 활용: HiveContext를 사용하면 Hive에 저장된 기존 데이터와도 손쉽게 연동할 수 있습니다.
- 임시 테이블 등록:
registerDataFrameAsTable()로 데이터프레임을 임시 테이블로 등록하면 SQL 문법으로 직관적으로 데이터를 조회할 수 있습니다.
이처럼 PySpark SQL과 Matplotlib을 결합하면 대용량 데이터를 SQL로 가공한 뒤, 익숙한 시각화 도구로 결과를 빠르게 확인할 수 있습니다.