[SparkSQL] 택시 데이터 다운/전처리/분석 feat. TLC

 

 

 

 

이전 포스팅에서 공부한 SparkSQL 지식을 바탕으로, 실제 Taxi 데이터를 전처리해보자.

* 전처리란? 이상치 제거, 그룹화 등 데이터 분석이 용이하도록 데이터를 변형하는 과정을 말한다.

 

 

TLC Trip Record Data에서 먼저 데이터를 받아오자. TLC는 미국의 택시 운전 데이터를 모아놓은 아주 유용한 사이트다.

[https://www1.nyc.gov/site/tlc/about/tlc-trip-record-data.page]

TLC Trip Record Data - TLC

 

 

 

"본 포스팅은 패스트캠퍼스의 강의를 듣고, 정리한 자료임을 밝힙니다."

 

 

 

 

 

Data Download

---

1) 기본 페이지

 

 

 

 

2) 2021년도 01월~07월 Yellow Taxi Trip Records(CSV) 클릭해서 다운로드하기

 

 

 

 

3) Taxi Zone Lookup Table (CSV) 다운로드

 

 

 

 

4) Spark 작업하는 경로에 data 폴더 생성하기 C:\자신이 작업하는 경로\data / data 파일 안에 trips 폴더 생성하기

 

 

 

5) data 폴더 안에 Lookup Table(CSV) 파일을 넣고, trips 폴더 안에 Yellow Taxi Trip Records(CSV) 넣기

 

C:\자신이 작업하는 경로\data

 

C:\자신이 작업하는 경로\data\trips

 

 

 

이제 데이터 전처리를 위한 파일 세팅이 끝났다. 본격적으로 Spark를 세팅하고 데이터 전처리를 시작해보자.

 

 

 

 

 

Basic Settings

---

# matplotlib 폰트 지정
from matplotlib import font_manager, rc
font_path = '자신이 쓰고 싶은 폰트 경로'
font = font_manager.FontProperties(fname=font_path).get_name()
rc('font', family=font)

 

import os
import findspark
findspark.init(os.environ.get("SPARK_HOME"))
import pyspark
from pyspark import SparkConf, SparkContext
import pandas as pd
import faulthandler
faulthandler.enable()
from pyspark.sql import SparkSession

spark = SparkSession.builder.master('local').appName("taxi-analysis").getOrCreate()


# 데이터가 있는 파일
# 'trips/*'을 통해 trips 폴더 안에 있는 모든 데이터를 불러옵니다.
zone_data = "C:/Spark 작업 경로/data/taxi_zone_lookup.csv"
trip_files = "C:/Spark 작업 경로/data/trips/*"


# 데이터 로드
trips_df = spark.read.csv(f"file:///{trip_files}", inferSchema = True, header = True)
zone_df = spark.read.csv(f"file:///{zone_data}", inferSchema = True, header = True)

 

데이터 스키마 살펴보고, createOrReplaceTempView() 적용하기

 

# 데이터 스키마
trips_df.printSchema()
zone_df.printSchema()


root
 |-- VendorID: integer (nullable = true)
 |-- tpep_pickup_datetime: string (nullable = true)
 |-- tpep_dropoff_datetime: string (nullable = true)
 |-- passenger_count: integer (nullable = true)
 |-- trip_distance: double (nullable = true)
 |-- RatecodeID: integer (nullable = true)
 |-- store_and_fwd_flag: string (nullable = true)
 |-- PULocationID: integer (nullable = true)
 |-- DOLocationID: integer (nullable = true)
 |-- payment_type: integer (nullable = true)
 |-- fare_amount: double (nullable = true)
 |-- extra: double (nullable = true)
 |-- mta_tax: double (nullable = true)
 |-- tip_amount: double (nullable = true)
 |-- tolls_amount: double (nullable = true)
 |-- improvement_surcharge: double (nullable = true)
 |-- total_amount: double (nullable = true)
 |-- congestion_surcharge: double (nullable = true)

root
 |-- LocationID: integer (nullable = true)
 |-- Borough: string (nullable = true)
 |-- Zone: string (nullable = true)
 |-- service_zone: string (nullable = true)

----

# 데이터 createOrReplaceTempView()
trips_df.createOrReplaceTempView("trips")
zone_df.createOrReplaceTempView("zone")

 

 

 

[trips_df의 칼럼 설명]

1. VendorID: 기록을 제공하는 the TPEP provider를 나타내는 코드

2. tpep_pickup_datetime: 승차 날짜 및 시간

3. tpep_dropoff_datetime: 하자 날짜 및 시간

4. passenger_count: 승객 수

5. trip_distance: 거리(mile)

6. PULocationID: 승차 장소(ID)

7. DOLocationID: 하차 장소(ID)

8. payment_type: 지불 방법

9. fare_amount: 요금
10. extra: 추가 요금
11. tip_amount: 팁
12. tolls_amount: 톨비
13. total_amount: 총 비용

 

 

[zone_df의 칼럼 설명]

1. LocationID: 지역 ID
2. Borough: 크게 본 지역구
3. Zone: 지역 동

 

 

 

 

 

 

데이터 전처리

---

1) 쓸만한 column만 모아놓기

승/하차 시간, 장소, 지불 방법, 승객 수, 전체 비용 등 분석에 필요한 칼럼만 남기고 나머진 지웠다.

 

# 쓸만한 column만 모아놓자.
query = '''
SELECT
    t.VendorID,
    TO_DATE(t.tpep_pickup_datetime) AS pickup_date,
    HOUR(t.tpep_pickup_datetime) AS pickup_time,
    TO_DATE(t.tpep_dropoff_datetime) AS dropoff_date,
    HOUR(t.tpep_dropoff_datetime) AS dropoff_time,
    t.passenger_count,
    t.trip_distance,
    t.payment_type,
    t.fare_amount,
    t.tip_amount,
    t.tolls_amount,
    t.total_amount,
    pz.Zone as pzone,
    dz.Zone as dzone
FROM
    trips t
    LEFT JOIN 
        zone pz
    ON
        t.PULocationID == pz.LocationID
    LEFT JOIN 
        zone dz
    ON
        t.DOLocationID == dz.LocationID
'''
taxi_df = spark.sql(query)
taxi_df.createOrReplaceTempView("taxi")

spark.sql('select * from taxi').show(5)


+--------+-----------+-----------+------------+------------+---------------+-------------+------------+-----------+----------+------------+------------+-----------------+--------------+
|VendorID|pickup_date|pickup_time|dropoff_date|dropoff_time|passenger_count|trip_distance|payment_type|fare_amount|tip_amount|tolls_amount|total_amount|            pzone|         dzone|
+--------+-----------+-----------+------------+------------+---------------+-------------+------------+-----------+----------+------------+------------+-----------------+--------------+
|       2| 2021-03-01|          0|  2021-03-01|           0|              1|          0.0|           2|        3.0|       0.0|         0.0|         4.3|               NV|            NV|
|       2| 2021-03-01|          0|  2021-03-01|           0|              1|          0.0|           2|        2.5|       0.0|         0.0|         3.8|   Manhattanville|Manhattanville|
|       2| 2021-03-01|          0|  2021-03-01|           0|              1|          0.0|           2|        3.5|       0.0|         0.0|         4.8|   Manhattanville|Manhattanville|
|       1| 2021-03-01|          0|  2021-03-01|           0|              0|         16.5|           1|       51.0|     11.65|        6.12|       70.07|LaGuardia Airport|            NA|
|       2| 2021-03-01|          0|  2021-03-01|           0|              1|         1.13|           1|        5.5|      1.86|         0.0|       11.16|     East Chelsea|            NV|
+--------+-----------+-----------+------------+------------+---------------+-------------+------------+-----------+----------+------------+------------+-----------------+--------------+
only showing top 5 rows

 

앞의 5개 데이터만 봐도 알 수 있듯이, trip_distance가 0이고 승차 장소가 NA로 나와있는 등 데이터 자체가 좀 더럽다. 분석에 바로 갖다 쓰면 십중팔구 이상한 결론에 도달할 것이다. 깨끗하게 정제해주자.

 

 

 

 

2) 데이터의 더러운 부분들을 찾아보자.

 

# step 1. 2021.01 자료부터 2021.06 자료까지만 다운했으므로, 그 외 데이터는 지운다.
# 2002년 데이터가 왜 여기에...?

spark.sql('select pickup_date from taxi order by pickup_date').show()


+-----------+
|pickup_date|
+-----------+
| 2002-12-31|
| 2003-01-05|
| 2004-04-04|
| 2008-12-31|
| 2008-12-31|
| 2008-12-31|
| 2008-12-31|
| 2008-12-31|
| 2008-12-31|
| 2008-12-31|
| 2008-12-31|
| 2008-12-31|
| 2008-12-31|
| 2008-12-31|
| 2008-12-31|
| 2008-12-31|
| 2008-12-31|
| 2008-12-31|
| 2008-12-31|
| 2008-12-31|
+-----------+

 

# step 2 - 다양한 값들에서 이상치 찾기
# 노란 택시에 승객이 9명 탈 수 있었나?
for i in ['passenger_count', 'trip_distance', 'fare_amount', 'tip_amount', 'tolls_amount', 'total_amount']:
    print(f'{i}의 데이터 요약본 -----------------------------------')
    taxi_df.select(i).describe().show()
    
    
passenger_count의 데이터 요약본 -----------------------------------
+-------+------------------+
|summary|   passenger_count|
+-------+------------------+
|  count|          14166672|
|   mean|1.4253783104458126|
| stddev|  1.04432704905968|
|    min|                 0|
|    max|                 9|
+-------+------------------+

trip_distance의 데이터 요약본 -----------------------------------
+-------+-----------------+
|summary|    trip_distance|
+-------+-----------------+
|  count|         15000700|
|   mean|6.628629402627818|
| stddev|671.7293482115828|
|    min|              0.0|
|    max|        332541.19|
+-------+-----------------+

fare_amount의 데이터 요약본 -----------------------------------
+-------+------------------+
|summary|       fare_amount|
+-------+------------------+
|  count|          15000700|
|   mean| 12.89269334830367|
| stddev|145.54843567115813|
|    min|            -643.5|
|    max|         398466.38|
+-------+------------------+

tip_amount의 데이터 요약본 -----------------------------------
+-------+-----------------+
|summary|       tip_amount|
+-------+-----------------+
|  count|         15000700|
|   mean|2.146797558780939|
| stddev|2.610914434555077|
|    min|          -333.32|
|    max|          1140.44|
+-------+-----------------+

tolls_amount의 데이터 요약본 -----------------------------------
+-------+-------------------+
|summary|       tolls_amount|
+-------+-------------------+
|  count|           15000700|
|   mean|0.31795104561765897|
| stddev| 1.6542914124457562|
|    min|             -38.02|
|    max|             956.55|
+-------+-------------------+

total_amount의 데이터 요약본 -----------------------------------
+-------+-----------------+
|summary|     total_amount|
+-------+-----------------+
|  count|         15000700|
|   mean|18.75545205708744|
| stddev|145.7442452805979|
|    min|           -647.8|
|    max|         398469.2|
+-------+-----------------+

 

자세히 보면 재밌는 데이터들을 발견할 수 있다. trip_distance의 최고 값이 332541 mile (535172.863km)이다. 지구 한 바퀴에 42000km인 것을 생각하면, 택시는 도대체 지구를 몇 바퀴 돈 것일까? 말도 안 되므로 삭제해줘야 하는 데이터다. 그 외에도 삭제해야 할 데이터들이 많다. 슬쩍 보면서 어떤 데이터를 전처리해줘야 할지 스스로 생각하길 바란다.

 

 

passenger_count: 9명은 너무 많다. 5명까지 제한.
trip_distance: 0보단 커야 하고, max는 100마일로 제한.
fare_amount: 0보다 커야하고, max는 100으로 제한
tip_amount: 0보다 커야하고, max는 50으로 제한
tolls_amount: 0보다 커야하고, max는 10으로 제한
total_amount: 0보다 커야하고, max는 1000으로 제한

 

 

 

3) Data Cleaning

 

# data clearning
# fare, tip, tolls 에 대한 것은 나중에 df 만들때 전처리해주자.
query = '''
SELECT
    *
FROM
    taxi t
WHERE
    t.total_amount < 2000
    AND t.total_amount > 0
    AND t.trip_distance < 100
    AND t.passenger_count < 6
    AND t.pickup_date >= '2021-01-01'
    AND t.pickup_date < '2021-08-01'
    AND t.dropoff_date >= '2021-01-01'
    AND t.dropoff_date < '2021-08-03'
'''

df_c = spark.sql(query)
df_c.createOrReplaceTempView('cleaned')

 

SQL문을 이용하여 데이터 전처리를 해주었다.

전처리가 알맞게 되었는지 확인해보자.

 

# 전처리 확인
spark.sql('SELECT pickup_date FROM cleaned ORDER BY pickup_date').show()


+-----------+
|pickup_date|
+-----------+
| 2021-01-01|
| 2021-01-01|
| 2021-01-01|
| 2021-01-01|
| 2021-01-01|
| 2021-01-01|
| 2021-01-01|
| 2021-01-01|
| 2021-01-01|
| 2021-01-01|
| 2021-01-01|
| 2021-01-01|
| 2021-01-01|
| 2021-01-01|
| 2021-01-01|
| 2021-01-01|
| 2021-01-01|
| 2021-01-01|
| 2021-01-01|
| 2021-01-01|
+-----------+

 

2021-01-01 데이터부터 존재하는 것을 확인할 수 있다.

다른 칼럼들의 요약 데이터들을 확인해보자.

 

# step 2 - 다양한 값들에서 이상치 찾아서 없애기
for i in ['passenger_count', 'trip_distance', 'fare_amount', 'tip_amount', 'tolls_amount', 'total_amount']:
    print(f'{i}의 데이터 요약본 -----------------------------------')
    df_c.select(i).describe().show()


passenger_count의 데이터 요약본 -----------------------------------
+-------+------------------+
|summary|   passenger_count|
+-------+------------------+
|  count|          13855011|
|   mean|1.3471861552473687|
| stddev|0.8633321134239649|
|    min|                 0|
|    max|                 5|
+-------+------------------+

trip_distance의 데이터 요약본 -----------------------------------
+-------+------------------+
|summary|     trip_distance|
+-------+------------------+
|  count|          13855011|
|   mean|2.8443242809406217|
| stddev|3.6296305328849514|
|    min|               0.0|
|    max|             99.96|
+-------+------------------+

fare_amount의 데이터 요약본 -----------------------------------
+-------+------------------+
|summary|       fare_amount|
+-------+------------------+
|  count|          13855011|
|   mean|12.173092585057173|
| stddev|10.921311212928543|
|    min|              -0.8|
|    max|            1320.0|
+-------+------------------+

tip_amount의 데이터 요약본 -----------------------------------
+-------+------------------+
|summary|        tip_amount|
+-------+------------------+
|  count|          13855011|
|   mean| 2.189352297880318|
| stddev|2.5818332665863704|
|    min|               0.0|
|    max|             700.0|
+-------+------------------+

tolls_amount의 데이터 요약본 -----------------------------------
+-------+-------------------+
|summary|       tolls_amount|
+-------+-------------------+
|  count|           13855011|
|   mean|0.27130274598821397|
| stddev|  1.540763209706977|
|    min|                0.0|
|    max|             956.55|
+-------+-------------------+

total_amount의 데이터 요약본 -----------------------------------
+-------+------------------+
|summary|      total_amount|
+-------+------------------+
|  count|          13855011|
|   mean| 18.08681097854774|
| stddev|13.216750985022102|
|    min|              0.01|
|    max|            1320.8|
+-------+------------------+

 

fare, tolls 등 전처리를 하지 않은 칼럼은 그대로이지만, passenger와 trip_distance 등의 데이터는 확실히 변한 것을 확인할 수 있다. 그렇다면 정제된 데이터로 간단한 분석을 진행해보자.

 

 

 

 

4) 간단한 분석 1. 요일 별 trips 수 세기

SQL 문의 DATE_FORMAT()를 사용하여 date를 요일 데이터로 변환했다. 요일 별로 그룹 하여 trips 수를 셌다.

 

import matplotlib.pyplot as plt
import seaborn as sns

# 1. 요일 별 trips 수 세기

query = '''
SELECT
    DATE_FORMAT(c.pickup_date, 'EEEE') AS day_of_week,
    COUNT(*) AS trips
FROM
    cleaned c
GROUP BY
    day_of_week
'''
weekday_df = spark.sql(query).toPandas()


# 그래프 그리기
fig, ax = plt.subplots(figsize=(16,6))
sns.barplot(
    x = 'day_of_week',
    y = 'trips',
    data = weekday_df
)

 

 

 

 

 

4) 간단한 분석 2. 지불 방법 통계 내기

지불 코드를 지불 방법 String으로 변환하고, 지불 방법을 Grouping 하여 trips 수를 세도록 했다.

 

# 6. 지불 방법 단순 통계
payment_type_to_string = {
    1: "Credit Card",
    2: "Cash",
    3: "No Charge",
    4: "Dispute",
    5: "Unknown",
    6: "Voided Trip",
}


# UDF 지정하기
def parse_payment_type(payment_type):
    return payment_type_to_string[payment_type]

spark.udf.register('parse_payment_type', parse_payment_type)


# 쿼리 짜기
query = '''
SELECT
    parse_payment_type(payment_type) as payment,
    COUNT(*) AS trips
FROM
    cleaned c
GROUP BY
    payment_type
'''

df_payment = spark.sql(query).toPandas()

----

display(df_payment)


	 payment	     trips
0	 Credit Card	 10534838
1	 No Charge	     59181
2	 Dispute	     23856
3	 Cash	         3237135
4	 Unknown	     1

----

# 지불 방법으로 그래프 그리기
fig, ax = plt.subplots(figsize=(16,6))
plt.title('payment', fontsize = 30)
sns.barplot(
    x = 'payment',
    y = 'trips',
    data = df_payment
)

 

 

 

 

지금까지 TLC Trip Record Data의 데이터를 전처리하고 간단한 분석을 해보았다.

스스로 데이터를 이리저리 만져가며 SparkSQL에 익숙해지길 바란다.

Analytics 섹션에서 TLC 데이터 분석을 더 상세하게 해서 포스팅하도록 하겠다.

수고하셨습니다.