<jupyter_start><jupyter_text>Практика PySpark и SparkSQL (Dataframe API)<jupyter_code># cоздаём SparkSession

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("Practice 2").getOrCreate()<jupyter_output><empty_output><jupyter_text>Чтение текстовых данных Данные [отсюда](https://www.kaggle.com/lepchenkov/usedcarscatalog) (Kaggle)<jupyter_code>!head -n5 data/cars.csv
# читаем данные в DataFrame c автоматическим определение схемы:

df = spark.read.format("csv").option("header", "true").load("data/cars.csv")<jupyter_output><empty_output><jupyter_text>Смотрим на данные<jupyter_code># вывести в консоль 5 строк из датафрейма:

df.show(5)<jupyter_output>+-----------------+----------+------------+------+--------------+-------------+-----------+--------------+-----------+---------------+---------+------------+-----+----------+---------+---------------+---------------+----------------+----------+---------+---------+---------+---------+---------+---------+---------+---------+---------+---------+---------------+
|manufacturer_name|model_name|transmission| color|odometer_value|year_produced|engine_fuel|engine_has_gas|engine_type|engine_capacity|body_type|has_warranty|state|drivetrain|price_usd|is_exchangeable|location_region|number_of_photos|up_counter|feature_0|feature_1|feature_2|feature_3|feature_4|feature_5|feature_6|feature_7|feature_8|feature_9|duration_listed|
+-----------------+----------+------------+------+--------------+-------------+-----------+--------------+-----------+---------------+---------+------------+-----+----------+---------+---------------+---------------+----------------+----------+---------+---------+---------+----[...]<jupyter_text>Дополнительный параметр `vertical=True` выведет каждую строку данных построчно в виде `колонка | значение`<jupyter_code>df.show(1, vertical=True)<jupyter_output>-RECORD 0-------------------------
 manufacturer_name | Subaru       
 model_name        | Outback      
 transmission      | automatic    
 color             | silver       
 odometer_value    | 190000       
 year_produced     | 2010         
 engine_fuel       | gasoline     
 engine_has_gas    | False        
 engine_type       | gasoline     
 engine_capacity   | 2.5          
 body_type         | universal    
 has_warranty      | False        
 state             | owned        
 drivetrain        | all          
 price_usd         | 10900.0      
 is_exchangeable   | False        
 location_region   | Минская обл. 
 number_of_photos  | 9            
 up_counter        | 13           
 feature_0         | False        
 feature_1         | True         
 feature_2         | True         
 feature_3         | True         
 feature_4         | False        
 feature_5         | True         
 feature_6         | False        
 feature_7         | True         
 feature_8         |[...]<jupyter_text>Смотрим на DataFrame API  .select() .select() выберет только нужные колонки (по аналогии с выражением `SELECT` в SQL)<jupyter_code># выбираем несколько колонок для отображения

df.select("manufacturer_name", "model_name", "transmission").show(5)<jupyter_output>+-----------------+----------+------------+
|manufacturer_name|model_name|transmission|
+-----------------+----------+------------+
|           Subaru|   Outback|   automatic|
|           Subaru|   Outback|   automatic|
|           Subaru|  Forester|   automatic|
|           Subaru|   Impreza|  mechanical|
|           Subaru|    Legacy|   automatic|
+-----------------+----------+------------+
only showing top 5 rows<jupyter_text>Можно обратиться к колонке через указание датафрейма:<jupyter_code>df.select(df["manufacturer_name"], df["model_name"], df["transmission"]).show(5)<jupyter_output>+-----------------+----------+------------+
|manufacturer_name|model_name|transmission|
+-----------------+----------+------------+
|           Subaru|   Outback|   automatic|
|           Subaru|   Outback|   automatic|
|           Subaru|  Forester|   automatic|
|           Subaru|   Impreza|  mechanical|
|           Subaru|    Legacy|   automatic|
+-----------------+----------+------------+
only showing top 5 rows<jupyter_text>Или используя функцию колонки `.col()` из пакета `functions`<jupyter_code>import pyspark.sql.functions as F

df.select(F.col("manufacturer_name"), F.col("model_name"), F.col("transmission")).show(5)<jupyter_output>+-----------------+----------+------------+
|manufacturer_name|model_name|transmission|
+-----------------+----------+------------+
|           Subaru|   Outback|   automatic|
|           Subaru|   Outback|   automatic|
|           Subaru|  Forester|   automatic|
|           Subaru|   Impreza|  mechanical|
|           Subaru|    Legacy|   automatic|
+-----------------+----------+------------+
only showing top 5 rows<jupyter_text>.filter() Метод .filter() принимает условия для фильтрации:<jupyter_code># выберем только марки Audi

df.select("manufacturer_name", "model_name", "transmission").filter("manufacturer_name = 'Audi'").show(5)<jupyter_output>+-----------------+----------+------------+
|manufacturer_name|model_name|transmission|
+-----------------+----------+------------+
|             Audi|        Q7|   automatic|
|             Audi|        TT|   automatic|
|             Audi|       100|  mechanical|
|             Audi|        A6|   automatic|
|             Audi|        Q3|   automatic|
+-----------------+----------+------------+
only showing top 5 rows<jupyter_text>Цепочка условий:<jupyter_code># выберем только марки Audi и ручной коробкой

(
    df
    .select("manufacturer_name", "model_name", "transmission")
    .filter("manufacturer_name = 'Audi'")
    .filter("transmission = 'mechanical'")
    .show(5)
)<jupyter_output>+-----------------+----------+------------+
|manufacturer_name|model_name|transmission|
+-----------------+----------+------------+
|             Audi|       100|  mechanical|
|             Audi|A6 Allroad|  mechanical|
|             Audi|       100|  mechanical|
|             Audi|        A4|  mechanical|
|             Audi|        80|  mechanical|
+-----------------+----------+------------+
only showing top 5 rows<jupyter_text>Цепочка условий в виде одного SQL выражения:<jupyter_code>(
    df
    .select("manufacturer_name", "model_name", "transmission")
    .filter("manufacturer_name = 'Audi' and transmission = 'mechanical'")
    .show(5)
)<jupyter_output>+-----------------+----------+------------+
|manufacturer_name|model_name|transmission|
+-----------------+----------+------------+
|             Audi|       100|  mechanical|
|             Audi|A6 Allroad|  mechanical|
|             Audi|       100|  mechanical|
|             Audi|        A4|  mechanical|
|             Audi|        80|  mechanical|
+-----------------+----------+------------+
only showing top 5 rows<jupyter_text>Удобнее и логичнее использовать `col()` для составления условий фильтрации:<jupyter_code>(
    df
    .select("manufacturer_name", "model_name", "transmission")
    .filter(F.col("manufacturer_name") == "Audi")
    .filter(F.col("transmission") != "mechanical")
    .show(5)
)<jupyter_output>+-----------------+----------+------------+
|manufacturer_name|model_name|transmission|
+-----------------+----------+------------+
|             Audi|        Q7|   automatic|
|             Audi|        TT|   automatic|
|             Audi|        A6|   automatic|
|             Audi|        Q3|   automatic|
|             Audi|        Q5|   automatic|
+-----------------+----------+------------+
only showing top 5 rows<jupyter_text>.count() Подсчет кол-ва строк:<jupyter_code>df.count()<jupyter_output><empty_output><jupyter_text>Либо уникальных строк:<jupyter_code>df.select("manufacturer_name").distinct().count()<jupyter_output><empty_output><jupyter_text>.groupBy() and .orderBy() Группировка и агрегация (аналог `GROUP BY` в SQL):<jupyter_code># сгрупировать по manufacturer_name и посчитать кол-во каждого

df.groupBy("manufacturer_name").count().show(5)<jupyter_output>+-----------------+-----+
|manufacturer_name|count|
+-----------------+-----+
|       Volkswagen| 4243|
|         Infiniti|  162|
|          Peugeot| 1909|
|            Lexus|  213|
|           Jaguar|   53|
+-----------------+-----+
only showing top 5 rows<jupyter_text>Сортировка по колонке по возрастанию:<jupyter_code>df.groupBy("manufacturer_name").count().orderBy("count").show(5)<jupyter_output>+-----------------+-----+
|manufacturer_name|count|
+-----------------+-----+
|          Lincoln|   36|
|       Great Wall|   36|
|              ЗАЗ|   42|
|          Pontiac|   42|
|         Cadillac|   43|
+-----------------+-----+
only showing top 5 rows<jupyter_text>Сортировка по колонке по убыванию. Тут надо явно задать колонку через `col()`:<jupyter_code>df.groupBy("manufacturer_name").count().orderBy(F.col("count").desc()).show(5)<jupyter_output>+-----------------+-----+
|manufacturer_name|count|
+-----------------+-----+
|       Volkswagen| 4243|
|             Opel| 2759|
|              BMW| 2610|
|             Ford| 2566|
|          Renault| 2493|
+-----------------+-----+
only showing top 5 rows<jupyter_text>.withColumnRenamed() and .withColumn() Переименовать существующую колонку:<jupyter_code>df.withColumnRenamed("manufacturer_name", "manufacturer").select("manufacturer").show(5)<jupyter_output>+------------+
|manufacturer|
+------------+
|      Subaru|
|      Subaru|
|      Subaru|
|      Subaru|
|      Subaru|
+------------+
only showing top 5 rows<jupyter_text>Создать новую колонку. Первый аргумент это название новой колонки, второй агрумент это выражение (обязательно использовать `col()` если ссылаемся на другую колонку):<jupyter_code>df.withColumn("next_year", F.col("year_produced") + 1).select("year_produced", "next_year").show(5)<jupyter_output>+-------------+---------+
|year_produced|next_year|
+-------------+---------+
|         2010|   2011.0|
|         2002|   2003.0|
|         2001|   2002.0|
|         1999|   2000.0|
|         2001|   2002.0|
+-------------+---------+
only showing top 5 rows<jupyter_text>.printSchema() and .describe() Вывести схему датафрейма (типы колонок):<jupyter_code>df.printSchema()
df.schema<jupyter_output><empty_output><jupyter_text>Вывести сводную статистику по датафрейму:<jupyter_code>df.select("manufacturer_name", "model_name", "year_produced", "price_usd").describe().show()
spark.stop()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.2727709454265949, 'fraction_numerical': 0.018447348193697154, 'mean_word_length': 1.7442657527023464, 'pattern_counts': {'":': 0, '<': 67, '<?xml version=': 0, '>': 67, 'https://': 1, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '20', 'dir': 'jupyter-structured-clean-dedup', 'id': '89709', 'license': 'MIT', 'n_tokens_mistral': 3360, 'n_tokens_neox': 3294, 'n_words': 534, 'path': 'practice-2/1-dataset-exploration.ipynb', 'repo_name': 'oleg-agapov/getting-started-with-pyspark-ru', 'size': '10408'}