스파크 경험해보기

안녕하세요 크리넥스을라스입니다.

 

스파크를 공부하기 시작하면서 공부한 내용들을 조금씩 정리해 볼려고 합니다.

 

저는 맥북 m1 에어를 사용하고 있으며, 스파크 설치 버전은 3.3.1 입니다.

맥북 m1 에어

파이썬

저는 터미널에서 pyspark 혹은 Spark의 bin 폴더에서 pyspark 명령어로 다음과 같은 화면을 실행시켰습니다.

이 화면에서 저희는 스파크를 이용해 데이터 분석을 수행할 수 있습니다.

 

저도 배워가는 입장이니 간단하게 2022 시즌 K리그 득점 순위 기록을 가지고 데이터를 다뤄보도록 하겠습니다.

득점 순위

저는 그냥 웹페이지의 글을 복사해서 하나의 csv 파일로 만들었습니다.

csv 파일

데이터가 준비되었다면 한 번 스파크에서 데이터를 불러오도록 하겠습니다.

 

스파크에서 데이터를 불러올 때는 다음과 같은 명령문을 사용하는데 자세한 거는 뒤에서 자세히 다뤄보도록 하겠습니다.

 

data = spark.read.format()  # 불러올 데이터의 형식  
                 .option("header", "true") # 파일 내 컬럼 이름 사용
                 .option("inferSchema", "true") # 스키마를 추론
                 .load()  # 불러올 파일 경로

data.show()  # 불러온 데이터를 출력하는 명령어. 괄호 안에 출력하고 싶은 개수를 넣으면 된다.

 

실행 결과

저는 다행히 데이터가 제대로 출력이 되었습니다.

 

컬럼이 많아서 보기 힘드니 원하는 컬러만 골라서 출력해보겠습니다.

 

data.select()  # 괄호 안에 선택한 컬럼명을 적으면 된다.

 

여기서 주의할 점은 select 명령어만 적고 엔터를 누르면 원하는 형태로 출력은 되지 않습니다. 

select만 쓸 때

그 이유는 스파크가 작동하는 원리에 있습니다. 잠깐 설명하면 스파크의 연산은 트랜스포메이션과 액션으로 구분되며,

 

트랜스포메이션 연산은 즉시 결과가 계산되는 것이 아니라 계보(lineage)라 불리는 형태로 기록되다가 액션 연산을 만나면 늦게 실행이 됩니다.

• 트랜스포매이션 : orderBy(), groupBy(), filter(), select(), join()
• 액션 : show(), take(), count(), collect(), save()

스파크가 이런 연산을 하는 이유는 장애가 발생했을 때 기록된 계보를 따라가면 쉽게 복구가 가능하다는 점과 실행의 후반부에 트랜스포메이션을 더 효율적으로 실행할 수 있다는 점입니다.

 

이어서 저는 팀별 득점 정보를 출력했습니다.

 

전북의 조규성 선수가 득점 순위 1위지만 30명의 선수들 중에서는 울산이 가장 많은 득점을 기록했습니다.

 

data.groupyBy()  # 그룹화할 컬럼 이름 설정
    .sum()       # 연산을 진행할 컬럼 설정
    .orderBy()   # 정렬할 컬럼 설정, ascending = True로 하면 오름차순으로 정렬된다.
    .show()

 

 

조금 더 자세히 보기 위해 득점 외에 도움과 슈팅, 오프사이드 컬럼도 추가했습니다. 정렬은 득점 기준입니다.

 

이번에는 슈팅 횟수가 70번 이상인 선수들만 출력하도록 조건을 넣었습니다 .

 

data.where()  # 변수명.컬럼명을 이용해 조건을 설정
    .show()

 

조건을 하나 더 추가해 슈팅을 70번 이상했지만 득점은 15번 이하인 선수들을 출력했습니다.

 

지금까지는 터미널 창에서 코드를 하나하나 수행했다면 이번에는 파이썬 파일로 같은 결과를 출력해보는 방법입니다.

 

먼저 다음과 같은 코드를 k-league.py 파일로 생성합니다.

 

# 필요한 라이브러리
import sys
from pyspark.sql import SparkSession

# 코드 시작
if __name__ == "__main__":
    # spark-submit을 할 때 추가적으로 파일 경로를 입력 안 할 때 출력되는 경고문
    if len(sys.argv) != 2:
        print("다시 시도해 주세요.")
        sys.exit(-1)
    
    # SparkSession 객체를 생성 > pyspark 창이나 spark-shell에서는 필요X
    spark = (SparkSession
             .builder
             .appName("k-league")
             .getOrCreate())
    
    # 터미널 창에서 입력 받은 파일 경로
    file_path = sys.argv[1]
    
    # 데이터 불러오기
    data = (spark.read.format("csv")
                .option("header", "true")
                .option("inferSchema", "true")
                .load(file_path))
    
    data.show(10)
    
    data2 = data.select("순위", "선수명", "팀", "득점", "슈팅", "출장", "경기당 기록")
    data2.show(10)
    
    data3 = data.groupBy("팀").sum("득점").orderBy("sum(득점)", ascending = False)    
    data3.show()
    
    
    data4 = data.groupBy("팀").sum("득점", "도움", "슈팅", "오프사이드").orderBy("sum(득점)", ascending = False)    
    data4.show()
    
    
    data5 = data.where(data.슈팅 >= 70)
    data5.show()

    data6 = data.where(data.슈팅 >= 70).where(data.득점 < 15)
    data6.show()
    
    # SparkSession 종료
    spark.stop()

 

Spark의 bin 폴더에서 spark-submit 을 이용해 파이썬 파일을 실행시킵니다.

(저는 미리 환경변수에 $SPARK_HOME을 지정해줘서 $SPARK_HOME/bin/spark-submit을 사용할 수 있습니다.)

spark경로/bin/spark-submit 파이썬 파일 경로

 

만약 파이썬 파일 경로 뒤에 csv 파일 경로를 넣지 않으면 파이썬 파일에서 설정한 대로 다음과 같이 출력됩니다.

csv 파일 경로까지 넣어서 실행시키면 다음과 같이 차례대로 출력됩니다.

 

 

출력되는 와중에 저처럼 깔끔하게 출력되지 않고 INFO 로그들이 엄청 많이 출력되시는 분들이 있다면

 

이 블로그를 참조하셔서 INFO 로그가 출력되지 않도록 설정해 주시면 됩니다.

 

또한, 제대로 한 거 같은데 실행이 안 되신다면 저처럼 pyspark 라이브러리를 설치하지 않으신건지

 

확인해보시면 될 것 같습니다.

 

여기까지는 파이썬을 이용해 스파크를 사용해보는 방법이었습니다.

 

스칼라를 사용하고 싶으신 분들은 pyspark대신 spark-shell을 터미널에 입력하시면 됩니다.

 

스칼라를 이용할 때 파이썬과 다른 가장 큰 부분은 변수를 생성할 때 val 변수명이 된다는 것입니다. 

스칼라를 이용한 부분은 조금 더 공부해서 다음 글부터는 같이 다룰 수 있도록 하겠습니다.

 

감사합니다.