Apache Spark 기초실습 step 2)
2019-03-26
Data_Engineering_TIL_(20190323)
study program : https://www.fastcampus.co.kr/extension_des
[학습목표]
- Apache Spark 기초실습
[학습기록]
-
SQL은 쿼리를 수정하거나 개선할때 어렵고, 쓸때없이 길어질 수 있지만 반면에 spark dataframe는 프로그래밍 코드로 되어 있기 때문에 이 코드들을 재활용하거나 조합하는등 조금 더 유연하다고 할 수 있음
-
데이터 분석 시 액셀을 많이 활용하므로 자유롭게 다룰 수 있어야 한다. 데이터의 용량이 크지 않은경우는 사실상 액셀로 모든 분석을 다 할 수 있다고 말할 수 있다. 20 ~ 30MB, 로우는 1만개 ~ 10만개까지는 액셀이 커버 가능하기 때문에 액셀을 쓰고, 수백메가바이트 정도의 데이터는 로컬에서 파이썬 판다스나 스파크 데이터 프레임으로 데이터를 돌리게되고, 그 이상이만 여러대로 처리한다.
-
예를 들어서 왕좌의 게임 데이터에서 어느 가문이 가장 많이 등장하는가를 산출하고자 할 때는 이런게 결국에는 집계하는 것이다. 어떠한 기준으로 데이터를 뭉쳐서 산출하자는 것인데 SQL로 치면 group by sum 등을 하는 것이다. 집계기능은 액셀에서는 피벗테이블이라는 강력한 도구를 제공한다.
[ 왕좌의 게임 등장인물 데이터를 활용한 아파치 스파크 실습 ]
- 데이터 관련 URL : https://www.kaggle.com/mylesoneill/game-of-thrones/data
# 최초 spark 구동
import findspark
findspark.init(r"C:\Users\minman\Desktop\css\spark\spark-2.4.0-bin-hadoop2.7")
# 리눅스에서는 절대경로를 넣어주면 된다
import pyspark
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
data_path = r"C:\Users\minman\Downloads\game-of-thrones\character-deaths.csv"
char_death_df = spark.read.option("header", "true").csv(data_path)
char_death_df.show()
+--------------------+---------------+----------+-------------+-------------+------------------+------+--------+---+---+---+---+---+
| Name| Allegiances|Death Year|Book of Death|Death Chapter|Book Intro Chapter|Gender|Nobility|GoT|CoK|SoS|FfC|DwD|
+--------------------+---------------+----------+-------------+-------------+------------------+------+--------+---+---+---+---+---+
| Addam Marbrand| Lannister| null| null| null| 56| 1| 1| 1| 1| 1| 1| 0|
|Aegon Frey (Jingl...| None| 299| 3| 51| 49| 1| 1| 0| 0| 1| 0| 0|
| Aegon Targaryen|House Targaryen| null| null| null| 5| 1| 1| 0| 0| 0| 0| 1|
| Adrack Humble| House Greyjoy| 300| 5| 20| 20| 1| 1| 0| 0| 0| 0| 1|
| Aemon Costayne| Lannister| null| null| null| null| 1| 1| 0| 0| 1| 0| 0|
| Aemon Estermont| Baratheon| null| null| null| null| 1| 1| 0| 1| 1| 0| 0|
|Aemon Targaryen (...| Night's Watch| 300| 4| 35| 21| 1| 1| 1| 0| 1| 1| 0|
| Aenys Frey| None| 300| 5| null| 59| 0| 1| 1| 1| 1| 0| 1|
| Aeron Greyjoy| House Greyjoy| null| null| null| 11| 1| 1| 0| 1| 0| 1| 0|
| Aethan| Night's Watch| null| null| null| 0| 1| 0| 0| 0| 1| 0| 0|
| Aggar| House Greyjoy| 299| 2| 56| 50| 1| 0| 0| 1| 0| 0| 0|
| Aggo|House Targaryen| null| null| null| 54| 1| 0| 1| 1| 1| 0| 1|
| Alan of Rosby| Night's Watch| 300| 5| 4| 18| 1| 1| 0| 1| 1| 0| 1|
| Alayaya| None| null| null| null| 15| 0| 0| 0| 1| 0| 0| 0|
| Albar Royce| Arryn| null| null| null| 38| 1| 1| 1| 0| 0| 1| 0|
| Albett| Night's Watch| null| null| null| 26| 1| 0| 1| 0| 0| 0| 0|
| Alebelly| House Stark| 299| 2| 46| 4| 1| 0| 0| 1| 0| 0| 0|
| Alerie Hightower| House Tyrell| null| null| null| 6| 0| 1| 0| 0| 1| 1| 0|
| Alesander Staedmon| Baratheon| null| null| null| 65| 1| 1| 0| 1| 0| 0| 0|
| Alester Florent| Baratheon| 300| 4| null| 36| 1| 1| 0| 1| 1| 0| 0|
+--------------------+---------------+----------+-------------+-------------+------------------+------+--------+---+---+---+---+---+
only showing top 20 rows
1) 가문별 사람 수를 group by 함수를 이용해서 출력해보기
char_death_df.groupBy("Allegiances").count().show()
+---------------+-----+
| Allegiances|count|
+---------------+-----+
| House Stark| 35|
| None| 253|
|House Baratheon| 8|
| House Tully| 8|
| House Martell| 12|
| Greyjoy| 51|
| Arryn| 23|
| House Tyrell| 11|
| House Greyjoy| 24|
|House Targaryen| 19|
| Baratheon| 56|
| Lannister| 81|
| Targaryen| 17|
| Tully| 22|
| Martell| 25|
| Tyrell| 15|
| House Arryn| 7|
| Night's Watch| 116|
| Stark| 73|
| Wildling| 40|
+---------------+-----+
only showing top 20 rows
from pyspark.sql.functions import col
# 아래 세줄의 코드는 모두 서로같은 표현이다.
# char_death_df.groupBy(char_death_df.Allegiances).count().show()
# char_death_df.groupBy(col("Allegiances")).count().show()
char_death_df.groupBy(char_death_df["Allegiances"]).count().show()
+---------------+-----+
| Allegiances|count|
+---------------+-----+
| House Stark| 35|
| None| 253|
|House Baratheon| 8|
| House Tully| 8|
| House Martell| 12|
| Greyjoy| 51|
| Arryn| 23|
| House Tyrell| 11|
| House Greyjoy| 24|
|House Targaryen| 19|
| Baratheon| 56|
| Lannister| 81|
| Targaryen| 17|
| Tully| 22|
| Martell| 25|
| Tyrell| 15|
| House Arryn| 7|
| Night's Watch| 116|
| Stark| 73|
| Wildling| 40|
+---------------+-----+
only showing top 20 rows
# 정렬까지해서 가문별 사람 수 출력, 변수에 할당
df1 = char_death_df.groupBy("Allegiances").count()
df1.orderBy(col("count").desc()).show()
+---------------+-----+
| Allegiances|count|
+---------------+-----+
| None| 253|
| Night's Watch| 116|
| Lannister| 81|
| Stark| 73|
| Baratheon| 56|
| Greyjoy| 51|
| Wildling| 40|
| House Stark| 35|
| Martell| 25|
| House Greyjoy| 24|
| Arryn| 23|
| Tully| 22|
|House Lannister| 21|
|House Targaryen| 19|
| Targaryen| 17|
| Tyrell| 15|
| House Martell| 12|
| House Tyrell| 11|
| House Tully| 8|
|House Baratheon| 8|
+---------------+-----+
only showing top 20 rows
2) join 함수를 이용한 가문별 사망률 구하기
# withColumn과 isNull함수를 이용한 is_alive 컬럼 추가하기
char_death_df2 = char_death_df.withColumn("is_alive", col("Death Year").isNull())
char_death_df2.show()
+--------------------+---------------+----------+-------------+-------------+------------------+------+--------+---+---+---+---+---+--------+
| Name| Allegiances|Death Year|Book of Death|Death Chapter|Book Intro Chapter|Gender|Nobility|GoT|CoK|SoS|FfC|DwD|is_alive|
+--------------------+---------------+----------+-------------+-------------+------------------+------+--------+---+---+---+---+---+--------+
| Addam Marbrand| Lannister| null| null| null| 56| 1| 1| 1| 1| 1| 1| 0| true|
|Aegon Frey (Jingl...| None| 299| 3| 51| 49| 1| 1| 0| 0| 1| 0| 0| false|
| Aegon Targaryen|House Targaryen| null| null| null| 5| 1| 1| 0| 0| 0| 0| 1| true|
| Adrack Humble| House Greyjoy| 300| 5| 20| 20| 1| 1| 0| 0| 0| 0| 1| false|
| Aemon Costayne| Lannister| null| null| null| null| 1| 1| 0| 0| 1| 0| 0| true|
| Aemon Estermont| Baratheon| null| null| null| null| 1| 1| 0| 1| 1| 0| 0| true|
|Aemon Targaryen (...| Night's Watch| 300| 4| 35| 21| 1| 1| 1| 0| 1| 1| 0| false|
| Aenys Frey| None| 300| 5| null| 59| 0| 1| 1| 1| 1| 0| 1| false|
| Aeron Greyjoy| House Greyjoy| null| null| null| 11| 1| 1| 0| 1| 0| 1| 0| true|
| Aethan| Night's Watch| null| null| null| 0| 1| 0| 0| 0| 1| 0| 0| true|
| Aggar| House Greyjoy| 299| 2| 56| 50| 1| 0| 0| 1| 0| 0| 0| false|
| Aggo|House Targaryen| null| null| null| 54| 1| 0| 1| 1| 1| 0| 1| true|
| Alan of Rosby| Night's Watch| 300| 5| 4| 18| 1| 1| 0| 1| 1| 0| 1| false|
| Alayaya| None| null| null| null| 15| 0| 0| 0| 1| 0| 0| 0| true|
| Albar Royce| Arryn| null| null| null| 38| 1| 1| 1| 0| 0| 1| 0| true|
| Albett| Night's Watch| null| null| null| 26| 1| 0| 1| 0| 0| 0| 0| true|
| Alebelly| House Stark| 299| 2| 46| 4| 1| 0| 0| 1| 0| 0| 0| false|
| Alerie Hightower| House Tyrell| null| null| null| 6| 0| 1| 0| 0| 1| 1| 0| true|
| Alesander Staedmon| Baratheon| null| null| null| 65| 1| 1| 0| 1| 0| 0| 0| true|
| Alester Florent| Baratheon| 300| 4| null| 36| 1| 1| 0| 1| 1| 0| 0| false|
+--------------------+---------------+----------+-------------+-------------+------------------+------+--------+---+---+---+---+---+--------+
only showing top 20 rows
# "Allegiances", "is_alive" 컬럼을 이용한 result_df 피벗테이블 만들기
result_df = char_death_df2.groupBy("Allegiances", "is_alive") \
.count().orderBy("Allegiances", "is_alive")
result_df.show()
+---------------+--------+-----+
| Allegiances|is_alive|count|
+---------------+--------+-----+
| Arryn| false| 3|
| Arryn| true| 20|
| Baratheon| false| 20|
| Baratheon| true| 36|
| Greyjoy| false| 8|
| Greyjoy| true| 43|
| House Arryn| false| 2|
| House Arryn| true| 5|
|House Baratheon| false| 4|
|House Baratheon| true| 4|
| House Greyjoy| false| 14|
| House Greyjoy| true| 10|
|House Lannister| false| 12|
|House Lannister| true| 9|
| House Martell| false| 1|
| House Martell| true| 11|
| House Stark| false| 19|
| House Stark| true| 16|
|House Targaryen| false| 4|
|House Targaryen| true| 15|
+---------------+--------+-----+
only showing top 20 rows
## result 데이터프레임을 두개의 데이터 프레임으로 쪼갠다. is_alive가 true인지 false인지를 기준으로..
df1 = result_df.filter(col("is_alive") == True) \
.withColumnRenamed("count", "alive_count")
df1.show()
+---------------+--------+-----------+
| Allegiances|is_alive|alive_count|
+---------------+--------+-----------+
| Arryn| true| 20|
| Baratheon| true| 36|
| Greyjoy| true| 43|
| House Arryn| true| 5|
|House Baratheon| true| 4|
| House Greyjoy| true| 10|
|House Lannister| true| 9|
| House Martell| true| 11|
| House Stark| true| 16|
|House Targaryen| true| 15|
| House Tully| true| 3|
| House Tyrell| true| 10|
| Lannister| true| 63|
| Martell| true| 23|
| Night's Watch| true| 60|
| None| true| 177|
| Stark| true| 46|
| Targaryen| true| 12|
| Tully| true| 18|
| Tyrell| true| 14|
+---------------+--------+-----------+
only showing top 20 rows
df2 = result_df.filter(col("is_alive") == False) \
.withColumnRenamed("Allegiances", "Allegiances1") \
.withColumnRenamed("count", "death_count")
df2.show()
+---------------+--------+-----------+
| Allegiances1|is_alive|death_count|
+---------------+--------+-----------+
| Arryn| false| 3|
| Baratheon| false| 20|
| Greyjoy| false| 8|
| House Arryn| false| 2|
|House Baratheon| false| 4|
| House Greyjoy| false| 14|
|House Lannister| false| 12|
| House Martell| false| 1|
| House Stark| false| 19|
|House Targaryen| false| 4|
| House Tully| false| 5|
| House Tyrell| false| 1|
| Lannister| false| 18|
| Martell| false| 2|
| Night's Watch| false| 56|
| None| false| 76|
| Stark| false| 27|
| Targaryen| false| 5|
| Tully| false| 4|
| Tyrell| false| 1|
+---------------+--------+-----------+
only showing top 20 rows
df1.join(df2, df1["Allegiances"] == df2["Allegiances1"]).show()
+---------------+--------+-----------+---------------+--------+-----------+
| Allegiances|is_alive|alive_count| Allegiances1|is_alive|death_count|
+---------------+--------+-----------+---------------+--------+-----------+
| Arryn| true| 20| Arryn| false| 3|
| Baratheon| true| 36| Baratheon| false| 20|
| Greyjoy| true| 43| Greyjoy| false| 8|
| House Arryn| true| 5| House Arryn| false| 2|
|House Baratheon| true| 4|House Baratheon| false| 4|
| House Greyjoy| true| 10| House Greyjoy| false| 14|
|House Lannister| true| 9|House Lannister| false| 12|
| House Martell| true| 11| House Martell| false| 1|
| House Stark| true| 16| House Stark| false| 19|
|House Targaryen| true| 15|House Targaryen| false| 4|
| House Tully| true| 3| House Tully| false| 5|
| House Tyrell| true| 10| House Tyrell| false| 1|
| Lannister| true| 63| Lannister| false| 18|
| Martell| true| 23| Martell| false| 2|
| Night's Watch| true| 60| Night's Watch| false| 56|
| None| true| 177| None| false| 76|
| Stark| true| 46| Stark| false| 27|
| Targaryen| true| 12| Targaryen| false| 5|
| Tully| true| 18| Tully| false| 4|
| Tyrell| true| 14| Tyrell| false| 1|
+---------------+--------+-----------+---------------+--------+-----------+
only showing top 20 rows
** join 관련 참고자료
# join을 이용한 결과 데이터 프레임 산출
result = df1.join(df2, df1["Allegiances"] == df2["Allegiances1"]) \
.select("Allegiances", "alive_count", "death_count", \
(col("alive_count") / (col("alive_count") + col("death_count"))).alias("alive_rate"))
result.show()
+---------------+-----------+-----------+-------------------+
| Allegiances|alive_count|death_count| alive_rate|
+---------------+-----------+-----------+-------------------+
| Arryn| 20| 3| 0.8695652173913043|
| Baratheon| 36| 20| 0.6428571428571429|
| Greyjoy| 43| 8| 0.8431372549019608|
| House Arryn| 5| 2| 0.7142857142857143|
|House Baratheon| 4| 4| 0.5|
| House Greyjoy| 10| 14| 0.4166666666666667|
|House Lannister| 9| 12|0.42857142857142855|
| House Martell| 11| 1| 0.9166666666666666|
| House Stark| 16| 19|0.45714285714285713|
|House Targaryen| 15| 4| 0.7894736842105263|
| House Tully| 3| 5| 0.375|
| House Tyrell| 10| 1| 0.9090909090909091|
| Lannister| 63| 18| 0.7777777777777778|
| Martell| 23| 2| 0.92|
| Night's Watch| 60| 56| 0.5172413793103449|
| None| 177| 76| 0.6996047430830039|
| Stark| 46| 27| 0.6301369863013698|
| Targaryen| 12| 5| 0.7058823529411765|
| Tully| 18| 4| 0.8181818181818182|
| Tyrell| 14| 1| 0.9333333333333333|
+---------------+-----------+-----------+-------------------+
only showing top 20 rows
3) 스파크 데이터 프레임을 판다스 데이터 프레임으로 변환해보기
result = df1.join(df2, df1["Allegiances"] == df2["Allegiances1"]) \
.select("Allegiances", "alive_count", "death_count", \
(col("alive_count") / (col("alive_count") + col("death_count"))).alias("alive_rate"))
result.toPandas()
| Allegiances | alive_count | death_count | alive_rate | |
|---|---|---|---|---|
| 0 | Arryn | 20 | 3 | 0.869565 |
| 1 | Baratheon | 36 | 20 | 0.642857 |
| 2 | Greyjoy | 43 | 8 | 0.843137 |
| 3 | House Arryn | 5 | 2 | 0.714286 |
| 4 | House Baratheon | 4 | 4 | 0.500000 |
| 5 | House Greyjoy | 10 | 14 | 0.416667 |
| 6 | House Lannister | 9 | 12 | 0.428571 |
| 7 | House Martell | 11 | 1 | 0.916667 |
| 8 | House Stark | 16 | 19 | 0.457143 |
| 9 | House Targaryen | 15 | 4 | 0.789474 |
| 10 | House Tully | 3 | 5 | 0.375000 |
| 11 | House Tyrell | 10 | 1 | 0.909091 |
| 12 | Lannister | 63 | 18 | 0.777778 |
| 13 | Martell | 23 | 2 | 0.920000 |
| 14 | Night's Watch | 60 | 56 | 0.517241 |
| 15 | None | 177 | 76 | 0.699605 |
| 16 | Stark | 46 | 27 | 0.630137 |
| 17 | Targaryen | 12 | 5 | 0.705882 |
| 18 | Tully | 18 | 4 | 0.818182 |
| 19 | Tyrell | 14 | 1 | 0.933333 |
| 20 | Wildling | 17 | 23 | 0.425000 |
4) window function을 이용하여 매년 죽은인물들의 가문 Top5를 뽑아보기
-
window function이란 행과 행간의 관계를 쉽게 정의하기 위해 만든 함수다.
-
집계한다는 개념보다는 특정 컬럼에서 특정 데이터의 하나하나의 종합적인 정보를 알기위해 많이 쓴다.
from pyspark.sql.window import Window
from pyspark.sql.functions import *
result = char_death_df.groupBy('Death Year', 'Allegiances').count().orderBy(col('Death Year'))
result.show()
+----------+---------------+-----+
|Death Year| Allegiances|count|
+----------+---------------+-----+
| null| Wildling| 17|
| null| Night's Watch| 60|
| null| Tully| 18|
| null| Martell| 23|
| null| House Arryn| 5|
| null| House Stark| 16|
| null| Stark| 46|
| null| Baratheon| 36|
| null| Targaryen| 12|
| null| Tyrell| 14|
| null| Lannister| 63|
| null| House Martell| 11|
| null| House Tully| 3|
| null| None| 177|
| null|House Targaryen| 15|
| null|House Baratheon| 4|
| null| Greyjoy| 43|
| null| House Greyjoy| 10|
| null| Arryn| 20|
| null| House Tyrell| 10|
+----------+---------------+-----+
only showing top 20 rows
window = Window.partitionBy("Death Year").orderBy(col("count").desc())
# 윈도우네는 파티셔닝과 오더링 두개가 들어가는데 파티셔닝이 그룹바이랑 비슷하다고 생각하면 되는 것이고, 오더링은 로우 넘버를 정해서
# 무슨순서로 줄을 세운다음 로우넘버를 셀건지 라는 것이다.
result.select('*', row_number().over(window).alias('n')).filter(col('n') <= 5).show()
# row_number는 집계기준을 어떻게 잡을것인가 설정해주는 것이고, 그것을 .over(window) 처럼 윈도우를 기준으로 하겠다는 것이다.
+----------+-------------+-----+---+
|Death Year| Allegiances|count| n|
+----------+-------------+-----+---+
| null| None| 177| 1|
| null| Lannister| 63| 2|
| null|Night's Watch| 60| 3|
| null| Stark| 46| 4|
| null| Greyjoy| 43| 5|
| 297|Night's Watch| 3| 1|
| 298| Stark| 10| 1|
| 298| House Stark| 9| 2|
| 298| None| 7| 3|
| 298| Targaryen| 4| 4|
| 298| Wildling| 3| 5|
| 299| None| 35| 1|
| 299|Night's Watch| 33| 2|
| 299| Stark| 17| 3|
| 299| Baratheon| 16| 4|
| 299|House Greyjoy| 11| 5|
| 300| None| 34| 1|
| 300|Night's Watch| 18| 2|
| 300| Wildling| 14| 3|
| 300| Lannister| 7| 4|
+----------+-------------+-----+---+
only showing top 20 rows
5) UDF(User Defined Function) 생성 및 사용해보기
-
파이썬 함수를 만들어서 판다스 데이터프레임에 적용하는 것과 비슷한것으로 마찬가지로 파이썬 함수를 만들어서 스파크 데이터프레임에 적용이 가능하다.
-
간단한 함수는 람다를 써도 되고, 만드려는 함수가 복잡해지면 UDF가 유용한 기능이다.
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
def genderFunc(gender):
if gender == 1:
return "Female"
else:
return "Male"
genderStringUDF = udf(genderFunc, StringType())
char_death_df.withColumn("Gender2", genderStringUDF(col("Gender"))).show()
+--------------------+---------------+----------+-------------+-------------+------------------+------+--------+---+---+---+---+---+-------+
| Name| Allegiances|Death Year|Book of Death|Death Chapter|Book Intro Chapter|Gender|Nobility|GoT|CoK|SoS|FfC|DwD|Gender2|
+--------------------+---------------+----------+-------------+-------------+------------------+------+--------+---+---+---+---+---+-------+
| Addam Marbrand| Lannister| null| null| null| 56| 1| 1| 1| 1| 1| 1| 0| Male|
|Aegon Frey (Jingl...| None| 299| 3| 51| 49| 1| 1| 0| 0| 1| 0| 0| Male|
| Aegon Targaryen|House Targaryen| null| null| null| 5| 1| 1| 0| 0| 0| 0| 1| Male|
| Adrack Humble| House Greyjoy| 300| 5| 20| 20| 1| 1| 0| 0| 0| 0| 1| Male|
| Aemon Costayne| Lannister| null| null| null| null| 1| 1| 0| 0| 1| 0| 0| Male|
| Aemon Estermont| Baratheon| null| null| null| null| 1| 1| 0| 1| 1| 0| 0| Male|
|Aemon Targaryen (...| Night's Watch| 300| 4| 35| 21| 1| 1| 1| 0| 1| 1| 0| Male|
| Aenys Frey| None| 300| 5| null| 59| 0| 1| 1| 1| 1| 0| 1| Male|
| Aeron Greyjoy| House Greyjoy| null| null| null| 11| 1| 1| 0| 1| 0| 1| 0| Male|
| Aethan| Night's Watch| null| null| null| 0| 1| 0| 0| 0| 1| 0| 0| Male|
| Aggar| House Greyjoy| 299| 2| 56| 50| 1| 0| 0| 1| 0| 0| 0| Male|
| Aggo|House Targaryen| null| null| null| 54| 1| 0| 1| 1| 1| 0| 1| Male|
| Alan of Rosby| Night's Watch| 300| 5| 4| 18| 1| 1| 0| 1| 1| 0| 1| Male|
| Alayaya| None| null| null| null| 15| 0| 0| 0| 1| 0| 0| 0| Male|
| Albar Royce| Arryn| null| null| null| 38| 1| 1| 1| 0| 0| 1| 0| Male|
| Albett| Night's Watch| null| null| null| 26| 1| 0| 1| 0| 0| 0| 0| Male|
| Alebelly| House Stark| 299| 2| 46| 4| 1| 0| 0| 1| 0| 0| 0| Male|
| Alerie Hightower| House Tyrell| null| null| null| 6| 0| 1| 0| 0| 1| 1| 0| Male|
| Alesander Staedmon| Baratheon| null| null| null| 65| 1| 1| 0| 1| 0| 0| 0| Male|
| Alester Florent| Baratheon| 300| 4| null| 36| 1| 1| 0| 1| 1| 0| 0| Male|
+--------------------+---------------+----------+-------------+-------------+------------------+------+--------+---+---+---+---+---+-------+
only showing top 20 rows
6) 출력한 데이터 프레임을 파일로 저장하기
# 기본적으로 스파크는 하둡분산시스템이기 때문에 파일도 분산저장된다.
# 그래서 설정을 잘 걸어줘야 원하는 형태의 파일로 저장할 수 있다.
result.repartition(1).write.mode("overwrite").csv("result.csv", header=True, sep='\t')
# 또한 jdbc writer를 쓰면 데이터베이스에 데이터를 쉽게 저장할 수 있다.
# json이나 Parquet, 오알씨 같은 빅데이터 포멧으로도 저장할 수 있다. csv는 Parquet로 변환해서 저장하면 된다.
result.write.jdbc(url="jdbc://JDBC_ADDRESS/database_name", "table_name")
- Spark 클러스터
과거 하둡 클러스터는 구현하기 어려웠으나 스파크 클러스터는 클러스터를 이루고자 하는 여러대의 컴퓨터 각각에 스파크를 설치한다. 그리고 마스터를 지정하고 클러스터에서는 마스터 아이피주소를 넣고 실행하면 된다.
사실 AWS의 EMR은 스파크 같은것을 하라고 만들어진거다 보다는 하둡이 매우 set up하기 어렵기 때문에 나온것이다.
- spark 클러스터링 구조
-
데이터가 커지게 되면 성능에 대한 고민을 할텐데 RDD보다 데이터 프레임의 성능이 압도적으로 좋다. 왜냐하면 데이터 프레임은 최적화가 가능하기 때문이다.
-
wide dependency는 항상 조심해야한다. 그룹바이 같은것을 할때 생각없이 하기 쉬운데 데이터 프레임과 데이터 프레임 간에 조인을 할 경우에는 성능에 신경을 써야한다.
-
인스턴스 타입이나 설정에서 팁은 조그만 인스턴스를 여러개 쓰냐, 큰 인스턴스를 조금 쓸까 이런 고민을 할텐데 사실 별차이는 없다. 하지만 메모리 50기가 이상되는 것은 쓰지 않는 것이 좋다. 그 이하의 사양으로 여러개를 잘 조합해서 쓰는 것이 좋다.
-
또한 스토리지를 직접 구축할 것인가, 클라우드를 구축할 것인가 고민도 할 것인데 작은 회사들은 클라우드를 쓰는 것이 이득이고 엄청나게 큰 데이터를 다루는 대기업에서는 클라우드를 쓰는 것이 오히려 비싸기 때문에 직접구축을 할지 고민할 수도 있다.
-
파일 포맷도 성능에 많은 영향을 준다. csv나 json은 사람이 읽기는 좋지만 자동화 된 파이프라인에서 쓸때나, 데이터가 커지면 커질수록 Parquet나 빅데이터를 위한 파일포맷이 더 유리하다.
-
클러스터만 잘 구성되어있다면, 기본 설정으로도 대부분의 작업은 문제가 없다.
-
임시파일 및 디스크 용량 관련 작업 정도가 필수적이다.
-
Wide dependency (Group by, Join)를 활용해야 하는 경우 튜닝이 필요할 수 있다.
1) spark.memory.fraction: execution과 storage에 사용할 메모리 비율설정 가능
2) Wide dependency를 많이 활용한다면 storage에 비중을 많이 잡아줘야 한다.
- https://spark.apache.org/docs/latest/configuration.html 을 잘 참고하면 좋다.
굳이 설정을 하지 않더라도 어떤것들이 가능하고, 내부 구조에 대한 아이디어를 많이 얻을 수 있음