[PySpark] Spark의 Dataframe API를 알아보자!(1)

🔊 해당 포스팅은 스파크 완벽 가이드 책과 인프런의 스파크 머신러닝 완벽 가이드 강의로 공부한 후 배운 내용을 저만의 방식으로 재구성한 것임을 알립니다. 특히, 참고한 인프런 강의의 강의 자료를 그대로 사용하지 않았음을 필히 알립니다!

Apache Spark를 Python으로 이용하는 PySpark에 대해 알아보자

---

예전에 아파치 스파크에 대한 개념글과 PySpark를 활용해서 문제 유형에 맞게 간단한 머신러닝 모델을 튜토리얼 형식으로 배웠었다. 그 이후 어느 정도 시간이 지난 후, 아파치 스파크의 버전도 많이 업데이트 되었고 개인적으로 분산 시스템에 대한 역량을 키워보고자 책과 강의를 통해 PySpark를 다시 배우고 있다.

 

필자는 PySpark를 활용하는 목적이 대량의 데이터를 전처리하고 머신러닝 모델을 개발하는 방법을 배워보는 것임을 알린다. 그래서 이번 첫 포스팅에서 알아볼 PySpark API는 바로 DataFrame API이다. Spark의 DataFrame API는 판다스 DataFrame과 비슷하기는 하지만 확연히 차이점은 존재하는 API이다. 마치 "SQL 탈을 쓴 DataFrame" 이라고도 할 수 있다.

 

그러면 이제 대표적인 벤치마크 데이터인 타이타닉 데이터를 활용해서 스파크의 데이터프레임 API를 사용하는 방법에 대해 알아보도록 하자. 참고로 앞으로 사용할 PySpark 버전은 3.2.0 이며 Jupyter Notebook에서 PySpark를 실습하였다.

1. 데이터 로드 및 미리보기

데이터가 현재 로컬에서 csv 데이터 형태로 저장되어 있기 때문에 Spark의 read.csv 메소드를 활용해서 데이터를 로드할 수 있다.

 

path = '/Users/younghun/Desktop/gitrepo/data/titanic/train.csv'

spark_df = spark.read.csv(path, header=True, inferSchema=True)
spark.show()

 

show() 메소드 결과

 

위 결과를 보면 show() 메소드는 로드한 스파크 데이터프레임의 상위 20개 Row를 미리보기 할 수 있다. 물론 show() 메소드 인자에 $n$(숫자)를 넣어주어 원하는 개수만큼 상위 $n$개의 Row를 미리보기 할 수 있다.

 

그리고 show() 메소드 앞에다가 limit() 이라는 메소드로 데이터프레임의 상위 Row 개수를 미리 추출할 수도 있다. limit() 메소드의 인자는 num 으로 아래처럼 줄 수 있다.

 

path = '/Users/younghun/Desktop/gitrepo/data/titanic/train.csv'

spark_df = spark.read.csv(path, header=True, inferSchema=True)
spark_df.limit(num=5).show()

 

또 다른 방벙으로 판다스와 마찬가지로 head() 메소드가 존재한다. 하지만 head() 메소드는 판다스 때와 달리 스파크에서는 Row 객체를 반환한다. 만약 head() 메소드에 아무런 인자도 주지 않으면 아래와 같이 가장 상위의 Row 객체 1개만을 반환한다.

 

path = '/Users/younghun/Desktop/gitrepo/data/titanic/train.csv'
spark_df = spark.read.csv(path, header=True, inferSchema=True)
spark_df.head()

 

head() 메소드 호출 시 반환되는 Row 객체 1개

 

특이한 점은 스키마(칼럼 이름)와 값 모두 반환된다. 그래서 위 Row 객체를 파이썬의 네임드 튜플처럼 Key 값을 참조하여 그에 해당하는 Value 값을 얻어낼 수 있다. 아래처럼 말이다.

 

path = '/Users/younghun/Desktop/gitrepo/data/titanic/train.csv'

spark_df = spark.read.csv(path, header=True, inferSchema=True)
row = spark_df.head()

print(row)
print()
print('row의 PassengerId:', row['PassengerId'], row.PassengerId)

 

그리고 또 특이한 점은 head() 메소드 인자에 1을 포함한 즉, 숫자를 인자로 넣어주게 되면 반드시 반환은 Row 객체(들)가 담긴 리스트를 반환하게 된다. 아래 그림처럼 말이다.

 

path = '/Users/younghun/Desktop/gitrepo/data/titanic/train.csv'
spark_df = spark.read.csv(path, header=True, inferSchema=True)

# 1. head()에 인자로 아무것도 안 줄 때
row = spark_df.head()
print('row:', row)
print()
# 2. head()에 인자로 1을 줄 때
row2 = spark_df.head(1)
print('row2:', row2)
print()
# 3. head()에 인자로 3을 줄 때
row3 = spark_df.head(3)
print('row3:', row3)

 

head() 메소드 인자에 숫자를 넣고 안넣고의 차이점

2. 데이터 명세서 살펴보기

다음은 PySpark로 로드한 데이터의 명세서를 보기 위한 메소드들이다. 명세서란, 데이터에 어떤 칼럼이 있고 그 칼럼 값의 타입은 무엇인지에 대해 정보를 제공하는 것이다. 판다스로 치면 dataframe.info() 또는 dataframe.describe() 정도가 되겠다.

 

그런데, PySpark는 판다스와 달리 문자열 형태의 칼럼에도 평균값, 표준편차 등과 같은 요약통계량값을 불필요하게 제공한다. 물론 Null 값으로 표현된다. 그래서 PySpark에서는 데이터프레임에서 수치형 타입만 고른 후 describe() 메소드를 활용해 아래와 같이 데이터 명세서 정보를 제공받을 수 있다. 참고로 PySpark에서의 describe() 메소드는 스파크의 데이터프레임 객체를 반환하므로 반드시 내용물을 보기 위해서는 show() 와 같이 미리보기 메소드를 사용해야 한다는 것도 잊지말자.

 

num_cols = [name for name, dtype in spark_df.dtypes if dtype != 'string']

spark_df.select(*num_cols).describe().show()  # 그냥 리스트 형태로 num_cols만 넣어도 동작하긴 함

3. 데이터프레임의 shape 얻기

다음은 PySpark의 데이터프레임의 행, 열 개수를 구하는 즉, shape를 구하는 방법이다. 판다스는 dataframe.shape 코드만 치면 튜플 형태로 (행의 개수, 열의 개수) 나왔지만, 스파크에서는 그렇지 못한다. 이에 대한 이유를 알아보면 다음과 같다. 우선 판다스는 하나의 서버 메모리에 로드한 데이터프레임을 올리게 된다. 메모리에 데이터프레임 모두가 저장되어 있기 때문에 shape 라는 속성 값을 이용해서 행과 열의 개수를 간단하게 구할 수 있다.

 

반면에, 스파크는 알다시피 서버에 여러개의 파티션으로 분산하여 데이터프레임을 나누어 저장하는 형태이다. 따라서, 판다스와는 달리 하나의 서버 메모리에 데이터프레임 전체를 저장하는 방식이 아니기 때문에 스파크에서는 shape라는 속성 값을 이용해 간단하게 행과 열의 개수를 구할 수 없다.

 

그래서 스파크는 약간 귀찮을지라도 아래와 같은 방식을 활용해서 데이터프레임의 행, 열 개수를 각각 구해준다.

 

# 행 개수 구하기
print('행 개수:', spark_df.count())

# 열 개수 구하기
print('열 개수:', len(spark_df.columns))

4. 데이터프레임에서 특정 칼럼들 추출하기

이제 PySpark에서 데이터프레임의 원하는 칼럼을 추출하는 방법에 대해 알아보도록 하자. 이번 목차에서 알아볼 칼럼 추출 방법은 select() 메소드를 활용하는 방법이다. 참고로 대부분의 다른 메소드들도 마찬가지지만 select() 메소드도 dataframe 객체 자체를 반환하기 때문에 내용물을 보기 위해서는 반드시 show() 메소드와 같은 것들을 추가로 작성해주어야 한다.

 

아래 코드를 보면 알겠지만, 특정 칼럼을 추출할 때, 다양한 방법으로 작성할 수 있다. 아래 코드를 면밀히 살펴보면서 5가지 방법이 모두 적용될 수 있다는 것도 알아두자. 예전 스파크 버젼에서는 원래 칼럼 이름이 담긴 리스트를 unpacking 해서 넣어주어야 했지만 최근 버전 업데이트가 되면서 리스트를 그냥 넣어주어도 정상적으로 동작한다.

 

from pyspark.sql.functions import col

# 모든 칼럼 추출하기
spark_df.select('*').limit(3).show()

# Age라는 칼럼 하나만 추출하기
spark_df.select('Age').limit(3).show()
spark_df.select(['Age']).limit(3).show()
spark_df.select(spark_df['Age']).limit(3).show()
spark_df.select(spark_df.Age).limit(3).show()
spark_df.select(col('Age')).limit(3).show()       # 임포트한 col 메소드 사용

# 여러개의 칼럼을 동시에 추출하기
spark_df.select('Name', 'Age').limit(3).show()
spark_df.select(['Name', 'Age']).limit(3).show()

 

다음으로는 pyspark.sql.functions 에서 제공하는 몇 가지 함수들을 임포트해서 특정 칼럼을 변경할 수도 있다.

 

from pyspark.sql.functions import max, min, avg
from pyspark.sql.functions import upper, lower


spark_df.select(max('Age').alias('Max_Age')).show()
spark_df.select(min('Pclass').alias('Min_Pclass')).show()
spark_df.select(avg('Age').alias('Avg_Age')).show()
spark_df.select(upper('Name').alias('Upper_Name')).limit(3).show()
# 모든 칼럼과 변환을 준 칼럼 동시에 추출
spark_df.select('*', lower(col('Name')).alias('Lower_Name')).limit(3).show()