In [1]:
import os
from pyspark.sql import SparkSession
# Python Version Mismatch Error 일 경우, os.environ 으로 직접 설정 후 실행
# Exception: Python in worker has different version 2.7 than that in driver 3.5,
# PySpark cannot run with different minor versions.Please check environment variables
# PYSPARK_PYTHON and PYSPARK_DRIVER_PYTHON are correctly set.
# os.environ["PYSPARK_PYTHON"] = "/usr/bin/python3"
# os.environ["PYSPARK_DRIVER_PYTHON"] = "/usr/bin/python3"
spark_home = os.environ.get('SPARK_HOME', None)
print(spark_home)
spark = SparkSession.builder.master("local[*]").appName("spark")
spark = spark.config("spark.driver.memory", "8g")
spark = spark.config("spark.executor.memory", "8g")
spark = spark.config("spark.python.worker.memory", "8g")
spark = spark.getOrCreate()
sc = spark.sparkContext
In [2]:
# import pyspark class
from pyspark.sql import *
from pyspark.sql import functions as f
from pyspark.sql import types as t
# SparkContext 를 이용해 생성
s = sc.parallelize([
(1, 'MacBook Pro', 2015, '15"', '16GB', '512GB SSD', 13.75, 9.48, 0.61, 4.02)
, (2, 'MacBook', 2016, '12"', '8GB', '256GB SSD', 11.04, 7.74, 0.52, 2.03)
, (3, 'MacBook Air', 2016, '13.3"', '8GB', '128GB SSD', 12.8, 8.94, 0.68, 2.96)
, (4, 'iMac', 2017, '27"', '64GB', '1TB SSD', 25.6, 8.0, 20.3, 20.8)
])
columns = ['Id', 'Model', 'Year', 'ScreenSize', 'RAM', 'HDD', 'W', 'D', 'H', 'Weight']
df = spark.createDataFrame(s, columns)
df.show()
In [3]:
# 직접 Row 클래스를 사용해서 행 생성
department1 = Row(id='123456', name='Computer Science')
department2 = Row(id='789012', name='Mechanical Engineering')
department3 = Row(id='345678', name='Theater and Drama')
department4 = Row(id='901234', name='Indoor Recreation')
# Class 를 사용해서 행 생성
Employee = Row("firstName", "lastName", "email", "salary")
employee1 = Employee('michael', 'armbrust', 'no-reply@berkeley.edu', 100000)
employee2 = Employee('xiangrui', 'meng', 'no-reply@stanford.edu', 120000)
employee3 = Employee('matei', None, 'no-reply@waterloo.edu', 140000)
employee4 = Employee(None, 'wendell', 'no-reply@berkeley.edu', 160000)
# Create the DepartmentWithEmployees instances from Departments and Employees
Group = Row("department", "employees")
group1 = Group(department1, [employee1, employee2])
group2 = Group(department2, [employee3, employee4])
group3 = Row(department=department3, employees=[employee1, employee4])
group4 = Row(department=department4, employees=[employee2, employee3])
print(department1)
print(employee1)
print(group1)
print(group3)
In [4]:
s1 = [group1, group2]
df1 = spark.createDataFrame(s1)
df1.show()
s2 = [group3, group4]
df2 = spark.createDataFrame(s2)
df2.show(df2.count(), False)
In [5]:
df_union = df1.unionAll(df2)
print(df_union.show(df_union.count(), False))
In [6]:
df_explode = df_union.select(f.col("department").alias("d"), f.explode("employees").alias("e"))
df_explode = df_explode.selectExpr("d.id", "d.name as departmentName", "e.firstName", "e.lastName", "e.email", "e.salary")
#df_explode = df_explode.select(f.col("d.id"), f.col("d.name").alias("departmentName"), f.col("e.firstName"), f.col("e.lastName"), f.col("e.email"), f.col("e.salary"))
df_explode.show(3)
In [7]:
# https://docs.databricks.com/spark/latest/data-sources/read-parquet.html
path_explode = "/tmp/df_explode.parquet"
df = df_explode
df = df.repartition(1)
df.write.mode('overwrite').parquet(path_explode)
df = spark.read.parquet(path_explode)
df.show(3)
# explode를 사용할 때, explode 대상이 되는 값이 빈 리스트([])라면, 해당 Row 는 제거됩니다.
# 만약 Row 는 유지하고 값만 null 로 처리하려면 Spark 2.2+ 부터 지원하는 explode_outer 함수를 사용하면 됩니다.
In [8]:
# https://docs.databricks.com/spark/latest/data-sources/read-csv.html
path_explode = "/tmp/df_explode.csv"
df = df_explode
df = df.repartition(1)
df.write.format("csv").mode('overwrite').option("header", "true").save(path_explode)
df = spark.read.format("csv").option("header", "true").option("inferSchema", "true").load(path_explode)
df.show(3)
In [9]:
#https://docs.databricks.com/spark/latest/data-sources/read-json.html
path_explode = "/tmp/df_explode.json"
df = df_explode
df = df.repartition(1)
df.write.format("json").mode('overwrite').save(path_explode)
df = spark.read.format("json").load(path_explode)
df.show(3)
In [10]:
import pandas as pd
df_pandas = df_explode.toPandas()
print(df_pandas.head())
df_spark = spark.createDataFrame(df_pandas)
print(df_spark.show())
In [11]:
df = df_explode
# DataFrame 컬럼 정보 살펴보기
print(df.printSchema())
print(df.schema)
print(df.columns)
print(df.dtypes)
In [12]:
df.show(3)
df.show(3, False)
print(df.first())
print(df.head(2))
print(df.take(2))
print(df.count())
print(df.select("id").distinct().show())
print(df.select("id").distinct().count())
In [13]:
# 열 선택
df.select("id", "departmentName", "firstName", "salary").show()
# 중복 제거
df.drop_duplicates(subset = ['firstName']).show()
In [14]:
# 조건을 통한 선택
# Where, Filter
df.select("id", "salary").filter(df["salary"] > 140000).show()
df.select("id", "salary").where(f.col("salary") > 140000).show()
# Between
df.select("id", "salary").where(df["salary"].between(10000, 140000)).show()
In [15]:
# Like
df.select("id", "departmentName").where(df['departmentName'].like("%Com%")).show()
# Startswith, endswith
df.select("id", "departmentName").where(df['departmentName'].startswith("Indoor")).show()
df.select("id", "departmentName").where(df['departmentName'].endswith("Drama")).show()
# isin
df.select("id", "departmentName").where(df["departmentName"].isin("Computer Science", "Indoor Recreation")).show()
In [16]:
# 연산을 통한 컬럼 생성
df.select("id", "salary", (df["salary"] * 0.5).alias("bonus")).show(2)
df.select("id", "salary").withColumn("bonus", df["salary"] * 0.5).show(2)
df.select("id", "departmentName", (df["departmentName"].substr(1, 3)).alias("substr")).show(2)
# 조건을 통한 컬럼 생성
df.select("id", "salary", f.when(df["salary"] > 120000, "High").otherwise("Low").alias("cost")).show(3)
In [17]:
# 사용자 함수를 통한 컬럼 생성
# Lambda 함수 방식
bonus = f.udf(lambda x, y: x * y, t.FloatType())
df.withColumn('bonus', bonus(df['salary'], f.lit(0.5))).show()
# Annotation 방식
@f.udf('float')
def bonus(x, y):
return x * y
df.withColumn('bonus', bonus(df['salary'], f.lit(0.5))).show()
In [18]:
df1 = sc.parallelize([
[1, 2, 3, 4, 5],
[1, 2, 3, 4, 5],
[1, 2, 3, 4, 5],
]).toDF(['c1', 'c2', 'c3', 'c4', 'c5'])
df1.show()
# Array 방식을 통한 여러 컬럼 연산
def sum_two(*args):
return args[0] + args[1]
udf_sum_two = f.udf(sum_two, t.IntegerType())
def sum_all(*args):
return sum(args)
udf_sum_all = f.udf(sum_all, t.IntegerType())
# Sum Two Columns c1 + c2
df1.withColumn("sum_two", udf_sum_two(f.col("c1"), f.col("c2"))).show()
# Sum All Columns
df1.withColumn("sum_all", udf_sum_all(*[f.col(i) for i in df1.columns])).show()
# Sum Some Columns
df1.withColumn("sum_some", udf_sum_all(f.col("c1"), f.col("c2"), f.col("c3"))).show()
In [19]:
df.select("id", "firstName", "lastName").show()
# 결측치 체크
df.select("id", "firstName").where(df["firstName"].isNotNull()).show()
df.select("id", "firstName").where(df["firstName"].isNull()).show()
In [20]:
# 상수로 체우기
df.select("id", "firstName", "lastName").fillna({ 'firstName': 'Unknown', 'lastName': 'Unknown' }).show()
In [21]:
# 컬럼 이름 변경
df.withColumnRenamed('id', 'ID').show()
# 컬럼 삭제
df.drop("email").show()
# 통계 확인
print("n rows: {}".format(df.count()))
df.describe().show()
In [22]:
# GroupBy
df.groupBy("departmentName").agg(f.sum('salary').alias('tot_salary'), f.mean('salary').alias('avg_salary')).show()
# Pivot
df.groupBy("departmentName").pivot('firstName').agg(f.mean('salary').alias('avg_salary')).show()
In [23]:
# Sort
df.sort(df['salary'].desc(), df['id'].asc()).show()
df.sort(["salary", "id"], ascending=[False, True]).show()
df.orderBy(["salary", "id"],ascending=[0, 1]).show()
In [24]:
sa = [(1, 'Pirate'),(2, 'Monkey'),(3, 'Ninja'),(4, 'Spaghetti')]
df_a = spark.createDataFrame(sa, ['a_id','a_name'])
sb = [(1, 'Rutabaga'),(2, 'Pirate'),(3, 'Ninja'),(4, 'Darth Vader')]
df_b = spark.createDataFrame(sb, ['b_id','b_name'])
df_a.show()
df_b.show()
# Join; Inner
df_join = df_a.alias('a').join(df_b.alias('b'), f.col("a.a_name") == f.col("b.b_name"), 'inner')
df_join = df_join.select(f.col("a.*"), f.col("b.*"))
df_join.show()
In [25]:
# Join: Left
df_join = df_a.alias('a').join(df_b.alias('b'), f.col("a.a_name") == f.col("b.b_name"), 'left')
df_join = df_join.select(f.col("a.*"), f.col("b.*"))
df_join.show()
# Join: Left
df_join = df_a.alias('a').join(df_b.alias('b'), f.col("a.a_name") == f.col("b.b_name"), 'right')
df_join = df_join.select(f.col("a.*"), f.col("b.*"))
df_join.show()
# Join: Full
df_join = df_a.alias('a').join(df_b.alias('b'), f.col("a.a_name") == f.col("b.b_name"), 'full')
df_join = df_join.select(f.col("a.*"), f.col("b.*"))
df_join.show()
In [26]:
# register the DataFrame as a temp table so that we can query it using SQL
df.registerTempTable("df_example")
# Perform the same query as the DataFrame above and return ``explain``
df_table = spark.sql("SELECT departmentName, SUM(salary) AS tot_salary, AVG(salary) AS avg_salary FROM df_example GROUP BY departmentName")
df_table.show()
In [27]:
spark.stop()
