<jupyter_start><jupyter_text>1. Установка PySpark<jupyter_code>!pip install pyspark<jupyter_output>Requirement already satisfied: pyspark in /usr/local/lib/python3.7/dist-packages (3.2.0)
Requirement already satisfied: py4j==0.10.9.2 in /usr/local/lib/python3.7/dist-packages (from pyspark) (0.10.9.2)<jupyter_text>2. Импорт необходимых библиотек и создание проекта<jupyter_code>import pyspark
from pyspark import SparkContext, SparkConf
sc = SparkContext(conf=SparkConf().setAppName("L1").setMaster("local[*]"))
sc<jupyter_output><empty_output><jupyter_text>3. Считываем данные<jupyter_code>data_trip = sc.textFile("/content/data/trips.csv")
data_station = sc.textFile("/content/data/stations.csv")<jupyter_output><empty_output><jupyter_text>4. Предобработка данных<jupyter_code>trip_header=data_trip.first()
station_header=data_station.first()
trips=data_trip.filter(lambda x: x != trip_header).map(lambda x: x.split(","))
stations=data_station.filter(lambda x: x != station_header).map(lambda x: x.split(","))
trips.take(3)
stations.take(3)
# Индексирование по нулевому элементу
station_idx = stations.keyBy(lambda x: int(x[0]))
station_idx.take(3)
# Индексирование по начальной и конечной станции
trips_start = trips.keyBy(lambda x: x[3])
trips_end = trips.keyBy(lambda x: x[6])
trips_start.take(3), trips_end.take(3)
# Объединение по ключу
start_trips = station_idx.join(trips_start)
end_trips = station_idx.join(trips_end)<jupyter_output><empty_output><jupyter_text>5. Создание модели для Trip & Station<jupyter_code>from typing import NamedTuple
from datetime import datetime
from functools import reduce
def initStation(stations):
    class Station(NamedTuple):
        station_id: int
        name: str
        lat: float
        long: float
        dockcount: int
        landmark: str
        installation: str
    
    for station in stations:
        yield Station(
            station_id = int(station[0]),
            name = station[1],
            lat = float(station[2]),
            long = float(station[3]),
            dockcount = int(station[4]),
            landmark = station[5],
            installation = datetime.strptime(station[6], '%m/%d/%Y')
        )

def initTrip(trips):
    class Trip(NamedTuple):
        trip_id: int
        duration: int
        start_date: datetime
        start_station_name: str
        start_station_id: int
        end_date: datetime
        end_station_name: str
        end_station_id: int
        bike_id: int
        subscription_type: str
        zip_code: str
        
    for trip in trips:
        try:
            yield Trip(                             
             trip_id = int(trip[0]),
             duration = int(trip[1]),
             start_date = datetime.strptime(trip[2], '%m/%d/%Y %H:%M'),
             start_station_name = trip[3],
             start_station_id = int(trip[4]),
             end_date = datetime.strptime(trip[5], '%m/%d/%Y %H:%M'),
             end_station_name = trip[6],
             end_station_id = trip[7],
             bike_id = int(trip[8]),
             subscription_type = trip[9],
             zip_code = trip[10]
            ) 
        except:
            pass
# Применяем модель данных для Stations
stations_ = stations.mapPartitions(initStation)
stations_.first()
# Применяем модель данных для Trip
trips_= trips.mapPartitions(initTrip)
trips_.first()
trips_start = trips_.keyBy(lambda x: x.start_station_name)
# Расчет среденего времени для каждого парковочного места
avg = trips_start.mapValues(lambda x: x.duration).groupByKey()\
.mapValues(lambda y: reduce(lambda a, b: (a+b), y)/len(y))
avg.top(10, lambda x: x[1])
# Найдем первую поездку для каждой из стоянок
first_trip = trips_start.reduceByKey(lambda x, y: x if x.start_date < y.start_date else y)
first_trip.collect()<jupyter_output><empty_output><jupyter_text>6. Выполненная лабораторная работа  I. Велосипед с максимальным пробегом<jupyter_code>bike_max_mileage = trips_.keyBy(lambda x: x.bike_id)
bike_duration = bike_max_mileage.mapValues(lambda x: x.duration).reduceByKey(lambda x1, x2: x1 + x2)
bike_duration_top = bike_duration.top(1, key=lambda x: x[1])[0][0]
bike_duration_top<jupyter_output><empty_output><jupyter_text>II. Найти наибольшее расстояние между станциями<jupyter_code>from math import *


def distance_between_stations(latitude1, longitude1, latitude2, longitude2):
  earth_radius = 6371.0088
  latitude1, longitude1 = radians(latitude1), radians(longitude1)
  latitude2, longitude2 = radians(latitude2), radians(longitude2)

  difference_latitude = latitude2 - latitude1
  difference_longitude = longitude2 - longitude1
  
  haversinus_latitutde = (sin(difference_latitude / 2)) ** 2
  haversinus_longitude = (sin(difference_longitude / 2)) ** 2
  
  return 2 * earth_radius * sqrt(haversinus_latitutde + cos(latitude1)\
                                 * cos(latitude2) * haversinus_longitude)


distance = stations_.cartesian(stations_)\
                         .filter(lambda x: x[0].station_id != x[1].station_id)\
                         .map(lambda x: [x[0], x[1], distance_between_stations(x[0].lat, x[0].long, x[1].lat, x[1].long)])\
                         .keyBy(lambda x: (x[0].name, x[1].name))\
                         .reduce(lambda x1, x2: x1 if x1[1] > x2[1] else x2)
print(f"Максимальное расстояние между станциями: {distance[0]}, {distance[1][2]}.")<jupyter_output>Максимальное расстояние между станциями: ('Ryland Park', 'Mezes Park'), 34.317914350160784.<jupyter_text>III. Найти путь велосипеда с максимальным пробегом через станции<jupyter_code>bike_path = trips_.filter(lambda x: x.bike_id == bike_duration_top)\
.sortBy(lambda x: x.start_date).map(lambda x: (x.start_station_name, x.end_station_name))
bike_path.first()<jupyter_output><empty_output><jupyter_text>IV. Найти количество велосипедов в системе.<jupyter_code>count_bikes = trips_.map(lambda x: x.bike_id).distinct().count()
count_bikes<jupyter_output><empty_output><jupyter_text>V. Найти пользователей потративших на поездки более 3 часов.<jupyter_code>users = trips_.filter(lambda x: x.duration > (3 * 60 * 60))\
.map(lambda x: x.zip_code).filter(lambda x: x != "").distinct()
users.take(10)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11978075124939545, 'fraction_numerical': 0.02256972432693858, 'mean_word_length': 3.754022988505747, 'pattern_counts': {'":': 0, '<': 40, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '10', 'dir': 'jupyter-structured-clean-dedup', 'id': '153563', 'license': 'MIT', 'n_tokens_mistral': 2281, 'n_tokens_neox': 2307, 'n_words': 423, 'path': 'L1 - Introduction to Apache Spark/L1 Introduction to Apache Spark.ipynb', 'repo_name': 'nvnovitskiy/big-data', 'size': '6203'}