setting
Temp View 등록
%pyspark
df.createOrReplaceTempView("df")df = spark.read.csv("/skybluelee/dat/airline_on_time")
sql에서 사용하기 위해 tempview로 등록
Catalog 확인
%pyspark
print(spark.catalog.listDatabases())
print(spark.catalog.listTables())
[Database(name='default', description='default database', locationUri='file:/home/spark/spark-warehouse')]
[Table(name='df', database=None, description=None, tableType='TEMPORARY', isTemporary=True)]Cache 처리(Persist 처리)
df이 자주 사용될 예정이므로 캐시처리함
%pyspark
import pyspark
df.persist(pyspark.StorageLevel.MEMORY_AND_DISK)
df.count()action을 위한 count
Cache 해제(Unpersist)
%pyspark
df.unpersist()데이터 탐색....
- [Q-01] 항공사 목록?
%pyspark
from pyspark.sql.functions import asc
df.select("UniqueCarrier").distinct()\
.orderBy(asc("UniqueCarrier"))\
.show(50)
%sql
SELECT DISTINCT UniqueCarrier
FROM df
ORDER BY 1 asc- [Q-02] 항공사 개수?
%pyspark
from pyspark.sql.functions import count
df.select("UniqueCarrier").distinct().select(count("UniqueCarrier").alias("carrier count"))
.show(50)
%sql
SELECT COUNT(DISTINCT UniqueCarrier) AS `carrier count` -- 공백이 존재한다면 ``로 감싸준다
FROM df- [Q-03] 항공사별 비행 횟수?
%pyspark
from pyspark.sql.functions import count
df.select("UniqueCarrier").agg(count("*").alias("flight_count"))
.show(50)
-- 집계함수는 agg로 감싸주어야 함
%sql
SELECT UniqueCarrier, COUNT(*) AS flight_count
FROM df
GROUP BY UniqueCarrier- [Q-04] 항공사별 계획된 비행 횟수 vs. 실제 비행 횟수 vs. 취소된 비행 횟수? + 취소율(%)?
비행 취소 attribute 확인
%pyspark
df.printSchema()
root
...
|-- Cancelled: integer (nullable = true)
...Cancelled의 type이 int임 -> NA값은 없을 것으로 추정
취소된 값을 파악하기 위해 Cancelled와 연관있는 attribute을 선택해 확인
%sql
SELECT Cancelled, CancellationCode, DepDelay, ArrDelay
FROM df
WHERE TRUE -- and Cancelled = 1 // true로 먼저 설정하고 뒤의 값을 주석처리하여 where문을 더 빠르게 사용
ORDER BY 1 desc
%sql
SELECT Cancelled, CancellationCode, DepDelay, ArrDelay
FROM df
WHERE TRUE -- and Cancelled = 1 // true로 먼저 설정하고 뒤의 값을 주석처리하여 where문을 더 빠르게 사용
ORDER BY 1 asc
CancellationCode의 경우 값이 제대로 나오지 않는 경우가 많을 것으로 추정
출발 지연과 도착 지연의 값을 토대로 취소된 경우는 Cancelled의 값이 1인 경우로 추정할 수 있음
%sql
SELECT DISTINCT Cancelled
FROM df에서 Cancelled의 값이 1과 0으로만 이루어져 있는 것을 확인
실제 값 확인
%pyspark
from pyspark.sql.functions import count, sum, when, col, round
df.groupBy("UniqueCarrier")\
.agg(count(*).alias("flight_count"),
sum("Cancelled").alias("flight_cancel_count"),
sum(when(col("Cancelled") == 0, 1).otherwise(0)).alias("flight_real_count"),
(sum(when(df.Cancelled == 1, 1).otherwise(0)) + sum(when(df.Cancelled == 0, 1).otherwise(0))).alias("flight_total_count"))\
.withColumn("flight_gap", col("flight_count") - col("flight_total_count"))\
.withColumn("cancel_rate(%)", round(col("flight_cancel_count") / col("flight_count") * 100, 1)) \
.orderBy(col("cancel_rate(%)").desc(), col("flight_cancel_count").desc(), col("flight_count").desc()) \
.show(100)
%sql
WITH temp AS -- 각 항공사별 전체 비행 횟수, 실제 비행 횟수, 취소된 비행 횟수
(
SELECT UniqueCarrier, COUNT(*) AS flight_count, SUM(Cancelled) AS flight_cancel_count,
SUM(case when Cancelled = 0 then 1 else 0 end) AS flight_real_count,
SUM(case when Cancelled = 1 then 1 else 0 end) +
SUM(case when Cancelled = 0 then 1 else 0 end) as flight_total_count
-- (실제 비행 + 비행 취소 = 전체 비행) 확인 용도
FROM df
GROUP BY UniqueCarrier
)
SELECT *, (flight_count - flight_total_count) AS flight_gap,
ROUND(flight_cancel_count / flight_count * 100, 1) as `cancel_rate(%)` -- 기호가 있다면 `사용
FROM temp
ORDER BY `cancel_rate(%)` desc, flight_cancel_count desc, flight_count desc
SELECT DISTINCT(flight_count - flight_total_count) AS flight_gap
FROM temp에서 (실제 비행 + 비행 취소 = 전체 비행) 임을 확인할 수 있다.
- [Q-05] 붐비는 공항?
- [Q-06] 실제 비행시간 vs. 예상 비행시간.... 차이가 큰 노선은?
- [Q-07] 각 공항별 노선별 운항 비중(%)은?
- [Q-08] 각 공항별 출발만 하는 항공사 vs. 도착만 하는 항공사 vs. 출발/도착 모두 하는 항공사?
- [Q-09] 각 노선별 항공사 운항 횟수?
