데이터 전처리 - PySpark에서 SQL 사용하기

저번 포스트에서 설명했듯이 이번 프로젝트에서는 배구 선수들의 능력치를 통해 팀 별 경기에서 어떤 팀이 승리할지를 예측하는 모델을 만들 것이다.

그러기 위해서는 선수 능력치를 판단할 수 있는 지표가 필요한데, 이전에 모았던 단순 데이터로는 한 눈에 파악하기 어렵다. 

 

현재 데이터의 컬럼에는 이름, 소속 팀, 포지션, 총 점수 외에도 공격 시도, 공격 성공, 리시브 시도, 리시브 정확, 리시브 실패, 서브 성공, 세트 성공, 블로킹 성공, 디그 성공 이 있다.

이 중 공격 지표로 사용할 수 있는 공격 성공률과 수비 지표로 사용할 수 있는 리시브 효율을 구하고, 참여한 세트 수를 통해 세트당 서브, 세트, 블로킹, 디그 개수를 구해서 새롭게 데이터를 만들 것이다.

 

위의 내용으로 데이터 전처리를 해 볼 텐데, 어떤 도구를 사용할까 고민하다가 파이스파크를 사용하니 스파크에서 제공하는 SQL을 사용해보기로 했다.

 

 

 

먼저, 스파크를 사용하기 위해 사용할 라이브러리들을 import 하고 세션을 초기화한다.

# spark 생성

import pyspark
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

크롤링해서 만든 선수 데이터를 하둡에 저장하고 불러온다.

csv 파일을 하둡에 저장하고 쥬피터 노트북을 통해 파이스파크를 사용하는 내용은 이전에 포스트로 작성해두었다.

data = spark.read.csv("hdfs://localhost:9000/data/player.csv", header="true", inferSchema="true")

불러온 데이터를 확인해보았다. show() 메소드를 속성값 지정 없이 사용하면 상위 20줄의 내용을 확인할 수 있다.

선수 데이터

데이터를 테이블로 사용하기 위해 선언해준다.

data.createOrReplaceTempView("player")

 

 

기존 데이터 내용을 토대로 공격 성공률, 리시브 효율, 세트당 서브/세트/블로킹/디그 를 계산해서 구하고 확인해본다.

# 공성, 공효(보류), 리효, 세트당 서브/세트/블로킹/디그 계산

player_rate = spark.sql("""
      select name,team,pos,score,error,setcount,round(at_succ/at_try,2) as at_srate,
          round((rs_corr-rs_fail)/rs_try,2) as rs_rate, round(sv_succ/setcount,2) as sv, 
          round(set_succ/setcount,2) as set,round(bl_succ/setcount,2) as bl,
          round(dg_succ/setcount,2) as dg
      from player
      """)

player_rate.show()

마찬가지로 가공한 데이터를 테이블로 사용하기 위해 선언해주었다.

player_rate.createOrReplaceTempView("player_rate")

SQL의 WHERE 절을 이용해 각 구단별로 데이터를 나눠주었다.

hgun_rate = spark.sql("""
          select *
          from player_rate
          where team like '현대건설'
          """)
hgun_rate.show(5)

 

 

가공한 선수 데이터와 구단 데이터를 하둡에 csv 파일로 저장하고 확인해본다.

# 선수 데이터 저장 - 하둡

player_rate.write.format('csv').save('hdfs://localhost:9000/data/player_r.csv',header='true')

data = spark.read.csv("hdfs://localhost:9000/data/player_r.csv", header='true', inferSchema='true')
data.show(5)

# 구단별 데이터 저장 - 하둡

hgun_rate.write.format('csv').save('hdfs://localhost:9000/data/hgun_r.csv',header='true')
dogong_rate.write.format('csv').save('hdfs://localhost:9000/data/dogong_r.csv',header='true')
gscal_rate.write.format('csv').save('hdfs://localhost:9000/data/gscal_r.csv',header='true')
insam_rate.write.format('csv').save('hdfs://localhost:9000/data/insam_r.csv',header='true')
ibk_rate.write.format('csv').save('hdfs://localhost:9000/data/ibk_r.csv',header='true')
hguk_rate.write.format('csv').save('hdfs://localhost:9000/data/hguk_r.csv',header='true')
pepper_rate.write.format('csv').save('hdfs://localhost:9000/data/pepper_r.csv',header='true')

# 데이터 잘 저장됐는지 확인

data = spark.read.csv("hdfs://localhost:9000/data/pepper_r.csv", header='true', inferSchema="true")
data.show(5)

 

데이터가 잘 저장되었는지 하둡에서 확인해보았다.

 

 

 

이렇게 데이터를 가공해서 하둡에 저장하는 것까지 완료했고, 앞으로는 이 데이터들을 통해 경기 결과를 예측할 수 있는 모델을 만들어 볼 예정이다.