101 практическое упражнение по PySpark

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, monotonically_increasing_id

# Определяем оконную спецификацию
w = Window.orderBy(monotonically_increasing_id())

# Добавляем индекс (нумерация с 0)
df = df.withColumn("index", row_number().over(w) - 1)

df.show()

+-------+-----+-----+
|   Name|Value|index|
+-------+-----+-----+
|  Alice|    1|    0|
|    Bob|    2|    1|
|Charlie|    3|    2|
+-------+-----+-----+

# Создаём RDD из списков и преобразуем в DataFrame
rdd = spark.sparkContext.parallelize(list(zip(list1, list2)))
df = rdd.toDF(["Column1", "Column2"])

df.show()

+-------+-------+
|Column1|Column2|
+-------+-------+
|      a|      1|
|      b|      2|
|      c|      3|
|      d|      4|
+-------+-------+

sc = spark.sparkContext

# Преобразуем списки в RDD
rdd_A = sc.parallelize(list_A)
rdd_B = sc.parallelize(list_B)

# Операция subtract
result_rdd = rdd_A.subtract(rdd_B)

result_list = result_rdd.collect()
print(result_list)

#> [1, 2, 3]

sc = spark.sparkContext

rdd_A = sc.parallelize(list_A)
rdd_B = sc.parallelize(list_B)

# Элементы A, которых нет в B, и элементы B, которых нет в A
result_rdd_A = rdd_A.subtract(rdd_B)
result_rdd_B = rdd_B.subtract(rdd_A)

# Объединяем два RDD
result_rdd = result_rdd_A.union(result_rdd_B)

result_list = result_rdd.collect()
print(result_list)

[1, 2, 3, 8, 6, 7]

# Вычисляем процентили
quantiles = df.approxQuantile("Age", [0.0, 0.25, 0.5, 0.75, 1.0], 0.01)

print("Min: ",              quantiles[0])
print("25-й процентиль: ", quantiles[1])
print("Медиана: ",         quantiles[2])
print("75-й процентиль: ", quantiles[3])
print("Max: ",             quantiles[4])

Min:  10.0
25-й процентиль:  20.0
Медиана:  30.0
75-й процентиль:  50.0
Max:  86.0

df.groupBy("job").count().show()

+---------+-----+
|      job|count|
+---------+-----+
| Engineer|    4|
|Scientist|    2|
|   Doctor|    1|
+---------+-----+

from pyspark.sql.functions import col, when

# Получаем 2 наиболее частых значения
top_2_jobs = (df.groupBy('job')
               .count()
               .orderBy('count', ascending=False)
               .limit(2)
               .select('job')
               .rdd.flatMap(lambda x: x)
               .collect())

# Заменяем все остальные значения на 'Other'
df = df.withColumn('job',
    when(col('job').isin(top_2_jobs), col('job')).otherwise('Other'))

df.show()

+----+---------+
|name|      job|
+----+---------+
|John| Engineer|
|John| Engineer|
|Mary|Scientist|
| Bob| Engineer|
| Bob| Engineer|
| Bob|Scientist|
| Sam|    Other|
+----+---------+

df_2 = df.dropna(subset=['Value'])
df_2.show()

+----+-----+----+
|Name|Value|  id|
+----+-----+----+
|   A|    1|null|
|   B|    3| 456|
+----+-----+----+

for old_name, new_name in zip(old_names, new_names):
    df = df.withColumnRenamed(old_name, new_name)

df.show()

+--------+--------+--------+
|new_col1|new_col2|new_col3|
+--------+--------+--------+
|       1|       2|       3|
|       4|       5|       6|
+--------+--------+--------+

# Определяем границы бакетов
num_buckets = 10
quantiles = df.stat.approxQuantile("values",
    [i/num_buckets for i in range(num_buckets + 1)], 0.01)

# Создаём Bucketizer
bucketizer = Bucketizer(splits=quantiles, inputCol="values", outputCol="buckets")

# Применяем
df_buck = bucketizer.transform(df)

# Таблица частот
df_buck.groupBy("buckets").count().show()

df_buck.show(5)

+-------+-----+
|buckets|count|
+-------+-----+
|    8.0|   10|
|    0.0|    8|
|    7.0|   10|
|    1.0|   10|
|    4.0|   10|
|    3.0|   10|
|    2.0|   10|
|    6.0|   10|
|    5.0|   10|
|    9.0|   12|
+-------+-----+

# Частоты
df.cube("category1").count().show()

# Таблица сопряжённости
df.crosstab('category1', 'category2').show()

+---------+-----+
|category1|count|
+---------+-----+
|     null|    8|
|        A|    3|
|        B|    2|
|        C|    3|
+---------+-----+

+-------------------+---+---+
|category1_category2|  X|  Y|
+-------------------+---+---+
|                  A|  2|  1|
|                  B|  1|  1|
|                  C|  2|  1|
+-------------------+---+---+

from pyspark.sql.functions import col, when

df = df.withColumn("is_multiple_of_3",
    when(col("random") % 3 == 0, 1).otherwise(0))

df.show()

+---+------+----------------+
| id|random|is_multiple_of_3|
+---+------+----------------+
|  0|     7|               0|
|  1|     6|               1|
|  2|     9|               1|
|  3|     7|               0|
|  4|     3|               1|
|  5|     8|               0|
|  6|     9|               1|
|  7|     8|               0|
|  8|     3|               1|
|  9|     8|               0|
+---+------+----------------+

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, monotonically_increasing_id

pos = [0, 4, 8, 5]

w = Window.orderBy(monotonically_increasing_id())
df = df.withColumn("index", row_number().over(w) - 1)

# Фильтруем строки по указанным позициям
df_filtered = df.filter(df.index.isin(pos))
df_filtered.show()

+---+------+-----+
| id|random|index|
+---+------+-----+
|  0|     7|    0|
|  4|     3|    4|
|  5|     8|    5|
|  8|     3|    8|
+---+------+-----+

df_A.union(df_B).show()

+------+-----+-----+
|  Name|Col_1|Col_2|
+------+-----+-----+
| apple|    3|    5|
|banana|    1|   10|
|orange|    2|    8|
| apple|    3|    5|
|banana|    1|   15|
| grape|    4|    6|
+------+-----+-----+

from pyspark.sql.functions import col, pow

# Квадрат разности
df = df.withColumn("squared_error", pow((col("actual") - col("predicted")), 2))

# Среднее значение
mse = df.agg({"squared_error": "avg"}).collect()[0][0]

print(f"Mean Squared Error (MSE) = {mse}")

Mean Squared Error (MSE) = 116.8

from pyspark.sql.functions import initcap

df = df.withColumn("name", initcap(df["name"]))
df.show()

+-----+
| name|
+-----+
| John|
|Alice|
|  Bob|
+-----+

summary = df.summary()
summary.show()

+-------+-------+------------------+-----------------+
|summary|   name|               age|           salary|
+-------+-------+------------------+-----------------+
|  count|      5|                 5|                5|
|   mean|   null|              32.4|          66000.0|
| stddev|   null|3.2093613071762417|9617.692030835673|
|    min|  James|                29|            55000|
|    25%|   null|                30|            60000|
|    50%|   null|                32|            65000|
|    75%|   null|                34|            70000|
|    max| Robert|                37|            80000|
+-------+-------+------------------+-----------------+

from pyspark.sql import functions as F

df = df.withColumn('word_length', F.length(df.name))
df.show()

+-----+-----------+
| name|word_length|
+-----+-----------+
| john|          4|
|alice|          5|
|  bob|          3|
+-----+-----------+

from pyspark.sql import functions as F
from pyspark.sql.window import Window

df = df.withColumn("id", F.monotonically_increasing_id())
window = Window.orderBy("id")

# Предыдущее значение (лаг)
df = df.withColumn("prev_value", F.lag(df.salary).over(window))

# Разность с лагом
df = df.withColumn("diff",
    F.when(F.isnull(df.salary - df.prev_value), 0)
     .otherwise(df.salary - df.prev_value)).drop("id")

df.show()

+-------+---+------+----------+------+
|   name|age|salary|prev_value|  diff|
+-------+---+------+----------+------+
|  James| 34| 55000|      null|     0|
|Michael| 30| 70000|     55000| 15000|
| Robert| 37| 60000|     70000|-10000|
|  Maria| 29| 80000|     60000| 20000|
|    Jen| 32| 65000|     80000|-15000|
+-------+---+------+----------+------+

from pyspark.sql.functions import to_date, dayofmonth, weekofyear, dayofyear, dayofweek

df = df.withColumn("date_1", to_date(df.date_str_1, 'yyyy-MM-dd'))
df = df.withColumn("date_2", to_date(df.date_str_2, 'dd MMM yyyy'))

df = (df
    .withColumn("day_of_month", dayofmonth(df.date_1))
    .withColumn("week_number",  weekofyear(df.date_1))
    .withColumn("day_of_year",  dayofyear(df.date_1))
    .withColumn("day_of_week",  dayofweek(df.date_1)))

df.show()

+----------+-----------+----------+----------+------------+-----------+-----------+-----------+
|date_str_1| date_str_2|    date_1|    date_2|day_of_month|week_number|day_of_year|day_of_week|
+----------+-----------+----------+----------+------------+-----------+-----------+-----------+
|2023-05-18|01 Jan 2010|2023-05-18|2010-01-01|          18|         20|        138|          5|
|2023-12-31|01 Jan 2010|2023-12-31|2010-01-01|          31|         52|        365|          1|
+----------+-----------+----------+----------+------------+-----------+-----------+-----------+

from pyspark.sql.functions import expr

# Преобразуем строку в дату (по умолчанию 1-й день месяца)
df = df.withColumn('Date', expr("to_date(MonthYear, 'MMM yyyy')"))
df.show()

# Заменяем день на 4-й
df = df.withColumn('Date', expr("date_add(date_sub(Date, day(Date) - 1), 3)"))
df.show()

+---------+----------+
|MonthYear|      Date|
+---------+----------+
| Jan 2010|2010-01-01|
| Feb 2011|2011-02-01|
| Mar 2012|2012-03-01|
+---------+----------+

+---------+----------+
|MonthYear|      Date|
+---------+----------+
| Jan 2010|2010-01-04|
| Feb 2011|2011-02-04|
| Mar 2012|2012-03-04|
+---------+----------+

from pyspark.sql.functions import col, length, translate

# Фильтруем слова с не менее 2 гласными
df_filtered = df.where(
    (length(col('Word')) - length(translate(col('Word'), 'AEIOUaeiou', ''))) >= 2
)
df_filtered.show()

+------+
|  Word|
+------+
| Apple|
|Orange|
| Money|
+------+

# Регулярное выражение для email
pattern = "^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\\.[a-zA-Z0-9-.]+$"

df_filtered = df.filter(F.col("value").rlike(pattern))
df_filtered.show()

+-----------------+
|            value|
+-----------------+
|rameses@egypt.com|
|        matt@t.co|
|narendra@modi.com|
+-----------------+

pivot_df = df.groupBy("year", "quarter").pivot("region").sum("revenue")
pivot_df.show()

+----+-------+----+----+
|year|quarter|  EU|  US|
+----+-------+----+----+
|2021|      2|4500|5500|
|2021|      1|4000|5000|
|2021|      3|5000|6000|
|2021|      4|6000|7000|
+----+-------+----+----+

from pyspark.sql.functions import mean

result = df.groupBy("Product").agg(mean("Price").alias("Avg_Price"))
result.show()

+----------+---------+
|   Product|Avg_Price|
+----------+---------+
|    Laptop|   1100.0|
|     Mouse|     40.0|
|Smartphone|    700.0|
+----------+---------+

from pyspark.sql.functions import expr

# Квадрат разности
df = df.withColumn("squared_diff", expr("POW(series1 - series2, 2)"))

# Корень из суммы квадратов
df.agg(expr("SQRT(SUM(squared_diff))").alias("euclidean_distance")).show()

+------------------+
|euclidean_distance|
+------------------+
|18.16590212458495|
+------------------+

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
from collections import Counter

def least_freq_char_replace_spaces(s):
    counter = Counter(s.replace(" ", ""))
    least_freq_char = min(counter, key=counter.get)
    return s.replace(' ', least_freq_char)

udf_replace = udf(least_freq_char_replace_spaces, StringType())

df.withColumn('modified_string', udf_replace(df['string'])).show()

+-----------------+-----------------+
|           string|  modified_string|
+-----------------+-----------------+
|dbc deb abed gade|dbccdebcabedcgade|
+-----------------+-----------------+

from pyspark.sql.functions import expr, explode, sequence, rand

start_date = '2000-01-01'
end_date   = '2000-03-04'

# Генерируем последовательность дат
df = spark.range(1).select(
    explode(
        sequence(
            expr(f"date '{start_date}'"),
            expr(f"date '{end_date}'"),
            expr("interval 1 day")
        )
    ).alias("date")
)

# Оставляем только субботы (7 в Spark)
df = df.filter(expr("dayofweek(date) = 7"))

# Добавляем случайные числа
df = df.withColumn("random_numbers", ((rand(seed=42) * 10) + 1).cast("int"))

df.show()

+----------+--------------+
|      date|random_numbers|
+----------+--------------+
|2000-01-01|             8|
|2000-01-08|             3|
|2000-01-15|             8|
|2000-01-22|             5|
|2000-01-29|             4|
|2000-02-05|             6|
|2000-02-12|             8|
|2000-02-19|             1|
|2000-02-26|             9|
|2000-03-04|             3|
+----------+--------------+

# Количество строк
nrows = df.count()
print("Количество строк:", nrows)

# Количество столбцов
ncols = len(df.columns)
print("Количество столбцов:", ncols)

# Типы данных каждого столбца
datatypes = df.dtypes
print("Типы данных:", datatypes)

Количество строк:  10000
Количество столбцов:  14
Типы данных:  [('RowNumber', 'int'), ('CustomerId', 'int'), ('Surname', 'string'), ('CreditScore', 'int'), ('Geography', 'string'), ('Gender', 'string'), ('Age', 'int'), ('Tenure', 'int'), ('Balance', 'double'), ('NumOfProducts', 'int'), ('HasCrCard', 'int'), ('IsActiveMember', 'int'), ('EstimatedSalary', 'double'), ('Exited', 'int')]

for old_name, new_name in zip(old_names, new_names):
    df = df.withColumnRenamed(old_name, new_name)

df.show()

+-----+--------+--------+
| name|user_age|user_qty|
+-----+--------+--------+
|Alice|       1|      30|
|  Bob|       2|      35|
+-----+--------+--------+

from pyspark.sql.functions import col, sum

missing = df.select(*(sum(col(c).isNull().cast("int")).alias(c) for c in df.columns))
has_missing = any(missing.collect()[0].asDict().values())
print("Есть пропуски:", has_missing)

missing_count = missing.collect()[0].asDict()
print("Количество пропусков:", missing_count)

Есть пропуски: True
Количество пропусков: {'Name': 0, 'Value': 2, 'id': 2}

from pyspark.ml.feature import Imputer

column_names = ["var1", "var2"]

imputer = Imputer(inputCols=column_names, outputCols=column_names, strategy="mean")
model = imputer.fit(df)
imputed_df = model.transform(df)

imputed_df.show()

+----+----+----+
|Name|var1|var2|
+----+----+----+
|   A|   1| 289|
|   B|   3| 123|
|   B|   3| 456|
|   D|   6| 289|
+----+----+----+

new_order = ["Age", "First_Name", "Last_Name"]
df = df.select(*new_order)
df.show()

+---+----------+---------+
|Age|First_Name|Last_Name|
+---+----------+---------+
| 30|      John|      Doe|
| 25|      Jane|      Doe|
| 22|     Alice|    Smith|
+---+----------+---------+

from pyspark.sql.functions import format_number

decimal_places = 10
df = df.withColumn("your_column", format_number("your_column", decimal_places))
df.show()

+---+------------+
| id| your_column|
+---+------------+
|  1|0.0000001230|
|  2|0.0000234560|
|  3|0.0003456780|
+---+------------+

from pyspark.sql.functions import concat, col, lit

columns = ["numbers_1", "numbers_2"]

for col_name in columns:
    df = df.withColumn(col_name,
        concat((col(col_name) * 100).cast('decimal(10, 2)'), lit("%")))

df.show()

+---------+---------+
|numbers_1|numbers_2|
+---------+---------+
|  10.00% |   8.00%|
|  20.00% |   6.00%|
|  33.00% |   2.00%|
+---------+---------+

window = Window.orderBy(monotonically_increasing_id())
df = df.withColumn("rn", row_number().over(window))

n = 5  # каждая 5-я строка
df = df.filter((df.rn % n) == 0)
df.show()

+----+------+---+
|Name|Number| rn|
+----+------+---+
| Eve|     5|  5|
|Jack|    10| 10|
+----+------+---+

from pyspark.sql.window import Window
from pyspark.sql.functions import desc, row_number

window = Window.orderBy(desc("column1"))
df = df.withColumn("row_number", row_number().over(window))

n = 3  # 3-е наибольшее значение
row = df.filter(df.row_number == n).first()

if row:
    print("Номер строки:", row.row_number)
    print("Значение в столбце:", row.column1)

Номер строки: 3
Значение в столбце: 8

from pyspark.sql import functions as F
from functools import reduce

# Добавляем столбец суммы строки
df = df.withColumn('row_sum', reduce(lambda a, b: a + b, [F.col(x) for x in df.columns]))

# Фильтруем строки, где сумма > 100
df = df.filter(F.col('row_sum') > 100)

# Берём последние 2 строки
df = df.withColumn('id', F.monotonically_increasing_id())
df_last_2 = df.sort(F.desc('id')).limit(2)

df_last_2.show()

+----+----+----+-------+-----------+
|col1|col2|col3|row_sum|         id|
+----+----+----+-------+-----------+
|  40|  50|  20|    110|25769803776|
|  70|  80| 100|    250|17179869184|
+----+----+----+-------+-----------+

from pyspark.sql.functions import udf, array
from pyspark.sql.types import FloatType

def min_max_ratio(row):
    return float(min(row)) / max(row)

min_max_ratio_udf = udf(min_max_ratio, FloatType())

df = df.withColumn('min_by_max', min_max_ratio_udf(array(df.columns)))
df.show()

+----+----+----+----------+
|col1|col2|col3|min_by_max|
+----+----+----+----------+
|   1|   2|   3|0.33333334|
|   4|   5|   6| 0.6666667|
|   7|   8|   9| 0.7777778|
|  10|  11|  12| 0.8333333|
+----+----+----+----------+

from pyspark.sql import functions as F
from pyspark.sql.types import ArrayType, IntegerType

sort_array_asc = F.udf(lambda arr: sorted(arr), ArrayType(IntegerType()))

df = df.withColumn("row_as_array", sort_array_asc(F.array(df.columns)))
df = df.withColumn("Penultimate", df['row_as_array'].getItem(1))
df = df.drop('row_as_array')

df.show()

+-------+-------+-------+-----------+
|Column1|Column2|Column3|Penultimate|
+-------+-------+-------+-----------+
|     10|     20|     30|         20|
|     40|     60|     50|         50|
|     80|     70|     90|         80|
+-------+-------+-------+-----------+