Описание проекта

Анализ данных о заказах такси у терминала №5 Нью-Йоркского аэропорта с помощью библиотеки PySpark и данных, загруженных из файла CSV в базу данных PostgreSQL.

Содержание

1 Подготовка тетради
2 Инициализация БД и загрузка данных из файла CSV
3 Анализ данных

Подготовка тетради¶

In [1]:

# Загрузка библиотек
import numpy as np
import pandas as pd
from pyspark.sql import SparkSession

Инициализация БД и загрузка данных из файла CSV¶

In [2]:

# Инициализация объекта БД
APP_NAME = "DataFrames"
SPARK_URL = "local[*]"

spark = SparkSession.builder.appName(APP_NAME) \
        .config('spark.ui.showConsoleProgress', 'false') \
        .getOrCreate()

# Загрузка данных из файла в БД
taxi = spark.read.load('/datasets/pickups_terminal_5.csv', 
                       format='csv', header='true', inferSchema='true')

# Создание временной таблицы
taxi.registerTempTable("taxi")

Анализ данных¶

Первичный анализ¶

In [3]:

# Анализ данных датафрейма, 
# полученного из файла CSV

# Вывод количества значений в таблице
print("Количество объектов в датафрейме:", taxi.count()) 

# Вывод первых пяти значений на экран
taxi.show(n=5) 

Количество объектов в датафрейме: 128974
+----------+----+------+-------+
|      date|hour|minute|pickups|
+----------+----+------+-------+
|2009-01-01|   0|     0|   24.0|
|2009-01-01|   0|    30|   35.0|
|2009-01-01|   1|     0|   25.0|
|2009-01-01|   1|    30|   25.0|
|2009-01-01|   2|     0|   16.0|
+----------+----+------+-------+
only showing top 5 rows

In [4]:

# Вывод трех не целевых столбцов 
# первых пяти строк на экран
taxi['date', 'hour', 'minute'].show(5)

+----------+----+------+
|      date|hour|minute|
+----------+----+------+
|2009-01-01|   0|     0|
|2009-01-01|   0|    30|
|2009-01-01|   1|     0|
|2009-01-01|   1|    30|
|2009-01-01|   2|     0|
+----------+----+------+
only showing top 5 rows

In [5]:

# Анализ значений датафрейма 
# методом describe
taxi.describe().show()

+-------+----------+------------------+------------------+------------------+
|summary|      date|              hour|            minute|           pickups|
+-------+----------+------------------+------------------+------------------+
|  count|    128974|            128974|            128974|            128969|
|   mean|      null|11.566509529052366|15.004419495402175|29.009451883786028|
| stddev|      null| 6.908556452594711|15.000057500526209|  22.4493784836831|
|    min|2009-01-01|                 0|                 0|               1.0|
|    max|2016-06-30|                23|                30|             310.0|
+-------+----------+------------------+------------------+------------------+

In [6]:

# Анализ значений датафрейма 
# методом summary
print(taxi.summary().show())

+-------+----------+------------------+------------------+------------------+
|summary|      date|              hour|            minute|           pickups|
+-------+----------+------------------+------------------+------------------+
|  count|    128974|            128974|            128974|            128969|
|   mean|      null|11.566509529052366|15.004419495402175|29.009451883786028|
| stddev|      null| 6.908556452594711|15.000057500526209|  22.4493784836831|
|    min|2009-01-01|                 0|                 0|               1.0|
|    25%|      null|                 6|                 0|              11.0|
|    50%|      null|                12|                30|              27.0|
|    75%|      null|                18|                30|              40.0|
|    max|2016-06-30|                23|                30|             310.0|
+-------+----------+------------------+------------------+------------------+

None

Обработка пропусков¶

In [7]:

# Анализ значений датафрейма 
# с предварительно заполненными пропусками
taxi.fillna(0).describe().show()

+-------+----------+------------------+------------------+------------------+
|summary|      date|              hour|            minute|           pickups|
+-------+----------+------------------+------------------+------------------+
|  count|    128974|            128974|            128974|            128974|
|   mean|      null|11.566509529052366|15.004419495402175| 29.00832725975778|
| stddev|      null| 6.908556452594711|15.000057500526209|22.449669931429067|
|    min|2009-01-01|                 0|                 0|               0.0|
|    max|2016-06-30|                23|                30|             310.0|
+-------+----------+------------------+------------------+------------------+

In [8]:

# Расчет количества объектов без пропусков 

print('Количество объектов без пропусков до удаления пропусков')
print('\nДо удаления пропусков   :', taxi.dropna(how='any', subset='pickups').count()) 

# Заполнение пропусков нулями
taxi = taxi.fillna(0)

print('После удаления пропусков:', taxi.dropna(how='any', subset='pickups').count()) 

Количество объектов без пропусков до удаления пропусков

До удаления пропусков   : 128969
После удаления пропусков: 128974

Данные в различных разрезах¶

In [9]:

# Количество заказов такси по дням 
# от большего к меньшему
result = spark.sql("""
SELECT   *
FROM     taxi
ORDER BY pickups DESC
--LIMIT    5
;
""")

print(result.show(5))

+----------+----+------+-------+
|      date|hour|minute|pickups|
+----------+----+------+-------+
|2015-11-01|   1|    30|  310.0|
|2010-09-23|  22|    30|  288.0|
|2012-03-07|  21|     0|  268.0|
|2011-03-02|  20|    30|  264.0|
|2011-03-02|  18|    30|  263.0|
+----------+----+------+-------+
only showing top 5 rows

None

In [10]:

# Количество дат в которые было 
# более 200 заказов за 30 минут
result = spark.sql("""
SELECT   COUNT(DISTINCT date)
FROM     taxi
WHERE    pickups > 200
;
""")

print(result.show())

+--------------------+
|count(DISTINCT date)|
+--------------------+
|                  21|
+--------------------+

None

In [11]:

# Группировка записей по месяцам. Рассчет среднего количества заказов по каждому месяцу. 
# Печать на экране таблицы с месяцами и средним количеством заказов по убыванию.
print(spark.sql("""
SELECT EXTRACT(MONTH FROM date), 
       AVG(pickups)
FROM   taxi
GROUP BY EXTRACT(MONTH FROM date)
ORDER BY AVG(pickups) DESC;
""").show())

+---------------------------------+------------------+
|extract('MONTH' FROM taxi.`date`)|      avg(pickups)|
+---------------------------------+------------------+
|                                3| 34.61413319776309|
|                               10|31.492839171666343|
|                                2|29.856671982987773|
|                                5| 29.81593638978176|
|                                4|29.313725490196077|
|                                9|29.158446485623003|
|                               11|28.860367558929283|
|                                1|28.559511612021858|
|                                6| 27.03835736129314|
|                                7| 26.45983005021244|
|                               12| 26.45916884626562|
|                                8| 25.88592750533049|
+---------------------------------+------------------+

None

In [12]:

# Вычисление среднего количества заказов за каждый час. Сортировка данных по убыванию. 
# Вывод самых загруженных 10 часов и среднего количества заказов такси в эти часы.
print(spark.sql("""
SELECT   hour,
         AVG(pickups)
FROM     taxi
GROUP BY hour
ORDER BY AVG(pickups) DESC
""").show(10))

+----+------------------+
|hour|      avg(pickups)|
+----+------------------+
|   8| 48.98208348725527|
|   9| 45.74220335855324|
|  18|45.131967515688444|
|  19| 40.18456995201181|
|  17| 37.68493909191584|
|  12| 36.91678966789668|
|  10|36.391031555637575|
|  14|35.965867158671585|
|   7| 35.94376618571957|
|  13| 35.34939091915836|
+----+------------------+
only showing top 10 rows

None