LATERAL VIEW и explode: unnesting массивов в строки

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, col

spark = SparkSession.builder.appName("explode-demo").getOrCreate()

data = [
    (1, ["python", "spark", "kafka"]),
    (2, ["scala", "java"]),
    (3, None),          # null массив - строка удалится
    (4, []),            # пустой массив - строка удалится
]

df = spark.createDataFrame(data, ["user_id", "skills"])

df.select("user_id", explode("skills").alias("skill")).show()

+-------+------+
|user_id| skill|
+-------+------+
|      1|python|
|      1| spark|
|      1| kafka|
|      2| scala|
|      2|  java|
+-------+------+

from pyspark.sql.functions import explode
from pyspark.sql.types import MapType, StringType, IntegerType

data = [(1, {"apples": 3, "bananas": 5}), (2, {"oranges": 2})]
df = spark.createDataFrame(data, ["id", "basket"])

df.select("id", explode("basket").alias("fruit", "count")).show()

+---+-------+-----+
| id|  fruit|count|
+---+-------+-----+
|  1| apples|    3|
|  1|bananas|    5|
|  2|oranges|    2|
+---+-------+-----+

from pyspark.sql.functions import explode_outer

df.select("user_id", explode_outer("skills").alias("skill")).show()

+-------+------+
|user_id| skill|
+-------+------+
|      1|python|
|      1| spark|
|      1| kafka|
|      2| scala|
|      2|  java|
|      3|  null|   ← null массив сохранён
|      4|  null|   ← пустой массив сохранён
+-------+------+

from pyspark.sql.functions import posexplode

data = [(1, ["impression", "click", "purchase"])]
df = spark.createDataFrame(data, ["session_id", "funnel_steps"])

df.select("session_id", posexplode("funnel_steps").alias("step_num", "step")).show()

+----------+--------+----------+
|session_id|step_num|      step|
+----------+--------+----------+
|         1|       0|impression|
|         1|       1|     click|
|         1|       2|  purchase|
+----------+--------+----------+

from pyspark.sql.functions import posexplode_outer

data = [(1, ["a", "b"]), (2, None)]
df = spark.createDataFrame(data, ["id", "vals"])

df.select("id", posexplode_outer("vals").alias("pos", "val")).show()

+---+----+----+
| id| pos| val|
+---+----+----+
|  1|   0|   a|
|  1|   1|   b|
|  2|null|null|
+---+----+----+

from pyspark.sql.functions import inline
from pyspark.sql.types import *

schema = StructType([
    StructField("order_id", IntegerType()),
    StructField("items", ArrayType(StructType([
        StructField("sku", StringType()),
        StructField("qty", IntegerType()),
        StructField("price", DoubleType()),
    ])))
])

data = [(1001, [("SKU-A", 2, 99.9), ("SKU-B", 1, 249.0)]),
        (1002, [("SKU-C", 3, 49.5)])]

df = spark.createDataFrame(data, schema)

df.select("order_id", inline("items")).show()

+--------+-----+---+-----+
|order_id|  sku|qty|price|
+--------+-----+---+-----+
|    1001|SKU-A|  2| 99.9|
|    1001|SKU-B|  1|249.0|
|    1002|SKU-C|  3| 49.5|
+--------+-----+---+-----+

from pyspark.sql.functions import explode, col

df.select(
    "order_id",
    explode("items").alias("item")
).select(
    "order_id",
    col("item.sku"),
    col("item.qty"),
    col("item.price")
).show()

SELECT
    o.order_id,
    item.sku,
    item.qty,
    item.price
FROM orders o
LATERAL VIEW explode(o.items) tmp AS item

SELECT
    session_id,
    step_pos,
    step_name
FROM sessions
LATERAL VIEW posexplode(funnel_steps) tmp AS step_pos, step_name
WHERE step_name = 'purchase'

SELECT
    user_id,
    skill,
    lang
FROM users
LATERAL VIEW explode(skills) t1 AS skill
LATERAL VIEW explode(languages) t2 AS lang

SELECT
    user_id,
    skill
FROM users
LATERAL VIEW OUTER explode(skills) tmp AS skill

from pyspark.sql.functions import explode

df_exploded = df.select("order_id", explode("items").alias("item"))
df_exploded.explain(mode="formatted")

== Physical Plan ==
Generate explode(items), [order_id], false, [item]
+- Scan parquet orders

# Посчитаем распределение размеров массивов до explode
from pyspark.sql.functions import size, percentile_approx, max as spark_max, avg

df.select(
    avg(size("items")).alias("avg_items"),
    spark_max(size("items")).alias("max_items"),
    percentile_approx(size("items"), 0.95).alias("p95_items"),
    percentile_approx(size("items"), 0.99).alias("p99_items"),
).show()

# Перед repartition - оцениваем skew
from pyspark.sql.functions import explode, spark_partition_id, count

df_exploded = df.select("user_id", explode("tags").alias("tag"))

# Сколько строк в каждой партиции после explode?
df_exploded.groupBy(spark_partition_id()).count().orderBy("count", ascending=False).show(20)

# Перераспределить после explode
df_normalized = df.select("user_id", explode("tags").alias("tag"))
df_normalized = df_normalized.repartition(200)  # или по ключу

from pyspark.sql.functions import explode

data = [(1, ["python", "spark"], ["backend", "data"])]
df = spark.createDataFrame(data, ["user_id", "skills", "roles"])

# ОПАСНО: двойной explode
df.select(
    "user_id",
    explode("skills").alias("skill"),
    explode("roles").alias("role")
).show()

+-------+------+-------+
|user_id| skill|   role|
+-------+------+-------+
|      1|python|backend|
|      1|python|   data|
|      1| spark|backend|
|      1| spark|   data|
+-------+------+-------+

skills_df = df.select("user_id", explode("skills").alias("skill"))
roles_df  = df.select("user_id", explode("roles").alias("role"))

# Если нужна нормализованная таблица фактов по навыкам и роли
# JOIN только если есть смысловая связь между позициями
result = skills_df.join(roles_df, "user_id")  # только если это уместно

from pyspark.sql.functions import posexplode

# Предположим: timestamps[i] соответствует events[i]
ts_df    = df.select("session_id", posexplode("timestamps").alias("pos", "ts"))
event_df = df.select("session_id", posexplode("events").alias("pos", "event"))

result = ts_df.join(event_df, ["session_id", "pos"])

from pyspark.sql.types import *

order_schema = StructType([
    StructField("order_id", LongType()),
    StructField("customer_id", LongType()),
    StructField("order_date", StringType()),
    StructField("status", StringType()),
    StructField("items", ArrayType(StructType([
        StructField("sku", StringType()),
        StructField("product_name", StringType()),
        StructField("category", StringType()),
        StructField("qty", IntegerType()),
        StructField("unit_price", DoubleType()),
        StructField("discount_pct", DoubleType()),
    ]))),
])

data = [
    (1001, 42, "2024-01-15", "completed", [
        ("SKU-001", "Laptop Pro 15", "electronics", 1, 89999.0, 0.05),
        ("SKU-002", "Mouse Wireless", "electronics", 2, 1299.0, 0.0),
        ("SKU-003", "Laptop Bag", "accessories", 1, 3499.0, 0.1),
    ]),
    (1002, 17, "2024-01-15", "completed", [
        ("SKU-004", "Python Book", "books", 3, 599.0, 0.15),
    ]),
    (1003, 99, "2024-01-15", "pending", None),  # заказ без позиций
]

orders_df = spark.createDataFrame(data, order_schema)

from pyspark.sql.functions import inline_outer, col, round as spark_round, to_date

# inline_outer сохраняет заказ 1003 (без позиций)
silver_items = orders_df.select(
    col("order_id"),
    col("customer_id"),
    to_date("order_date").alias("order_date"),
    col("status"),
    inline_outer("items"),  # разворачивает и сразу выводит поля структуры
)

# Добавим вычисляемые поля
silver_items = silver_items.withColumn(
    "line_total",
    spark_round(col("qty") * col("unit_price") * (1 - col("discount_pct")), 2)
)

silver_items.show(truncate=False)

+--------+-----------+----------+---------+-------+--------------+-----------+---+----------+------------+----------+
|order_id|customer_id|order_date|   status|    sku|  product_name|   category|qty|unit_price|discount_pct|line_total|
+--------+-----------+----------+---------+-------+--------------+-----------+---+----------+------------+----------+
|    1001|         42|2024-01-15|completed|SKU-001| Laptop Pro 15|electronics|  1|  89999.0|        0.05|  85499.05|
|    1001|         42|2024-01-15|completed|SKU-002|Mouse Wireless|electronics|  2|   1299.0|         0.0|   2598.00|
|    1001|         42|2024-01-15|completed|SKU-003|    Laptop Bag|accessories|  1|   3499.0|         0.1|   3149.10|
|    1002|         17|2024-01-15|completed|SKU-004|   Python Book|      books|  3|    599.0|        0.15|   1527.45|
|    1003|         99|2024-01-15|  pending|   null|          null|       null|  0|      null|        null|      null|
+--------+-----------+----------+---------+-------+--------------+-----------+---+----------+------------+----------+

from pyspark.sql.functions import sum as spark_sum, countDistinct

# Выручка по категориям
silver_items.filter(col("status") == "completed") \
    .groupBy("category") \
    .agg(
        spark_sum("line_total").alias("revenue"),
        countDistinct("order_id").alias("orders"),
        spark_sum("qty").alias("units_sold")
    ) \
    .orderBy("revenue", ascending=False) \
    .show()

+-----------+---------+------+----------+
|   category|  revenue|orders|units_sold|
+-----------+---------+------+----------+
|electronics|91596.05|     1|         3|
|accessories| 3149.10|     1|         1|
|      books| 1527.45|     1|         3|
+-----------+---------+------+----------+

from pyspark.sql.functions import transform, col

data = [(1, [100.0, 200.0, 350.0])]
df = spark.createDataFrame(data, ["order_id", "prices"])

# Применить скидку 10% ко всем ценам
df.select(
    "order_id",
    transform("prices", lambda x: x * 0.9).alias("discounted_prices")
).show()

+--------+-----------------------+
|order_id|      discounted_prices|
+--------+-----------------------+
|       1|[90.0, 180.0, 315.0]  |
+--------+-----------------------+

SELECT
    order_id,
    transform(prices, p -> p * 0.9) AS discounted_prices
FROM orders

from pyspark.sql.functions import filter

# Оставить только позиции дороже 1000 рублей
data = [(1, [("SKU-A", 500.0), ("SKU-B", 1500.0), ("SKU-C", 200.0)])]
# Для простоты - массив чисел
data = [(1, [500.0, 1500.0, 200.0, 2000.0])]
df = spark.createDataFrame(data, ["order_id", "prices"])

df.select(
    "order_id",
    filter("prices", lambda x: x > 1000).alias("premium_prices")
).show()

+--------+--------------+
|order_id|premium_prices|
+--------+--------------+
|       1| [1500.0, 2000.0]|
+--------+--------------+

from pyspark.sql.functions import aggregate

# Сумма всех цен в заказе без explode
df.select(
    "order_id",
    aggregate(
        "prices",
        lit(0.0),       # начальное значение аккумулятора
        lambda acc, x: acc + x  # лямбда: аккумулятор + текущий элемент
    ).alias("total_price")
).show()

+--------+-----------+
|order_id|total_price|
+--------+-----------+
|       1|     4200.0|
+--------+-----------+

SELECT
    order_id,
    aggregate(prices, CAST(0.0 AS DOUBLE), (acc, x) -> acc + x) AS total_price
FROM orders

from pyspark.sql.functions import exists, forall

# Есть ли хоть одна цена выше 1000?
df.select("order_id", exists("prices", lambda x: x > 1000).alias("has_premium")).show()

# Все ли цены выше 100?
df.select("order_id", forall("prices", lambda x: x > 100).alias("all_premium")).show()

# Антипаттерн: explode → groupBy → shuffle с огромным датасетом
df.select("user_id", explode("purchases").alias("purchase")) \
    .groupBy("user_id") \
    .agg(spark_sum("purchase.amount").alias("total_spent"))

# Оптимально: aggregate без explode
from pyspark.sql.functions import aggregate, col

df.select(
    "user_id",
    aggregate(
        "purchases",
        lit(0.0),
        lambda acc, x: acc + x.getField("amount")
    ).alias("total_spent")
)

df_exploded = df.select("user_id", explode("tags").alias("tag"))

# Рекластеризуем по ключу агрегации
df_exploded = df_exploded.repartition(200, "tag")

# Теперь groupBy("tag") не будет дополнительно шафлить
df_exploded.groupBy("tag").count().show()

from pyspark.sql.functions import col, inline_outer, to_date, current_timestamp

# Bronze: читаем из Delta/Parquet как есть
bronze_orders = spark.read.format("delta").load("s3a://datalake/bronze/orders/")

# Silver: нормализованные позиции заказов
silver_order_items = (
    bronze_orders
    .filter(col("_corrupt_record").isNull())  # убираем битые записи
    .select(
        col("order_id"),
        col("customer_id"),
        to_date("created_at").alias("order_date"),
        col("currency"),
        col("status"),
        inline_outer("line_items"),  # разворачиваем позиции
    )
    .withColumn("extracted_at", current_timestamp())
)

# Пишем в Silver с партиционированием по дате
silver_order_items.write \
    .format("delta") \
    .mode("overwrite") \
    .partitionBy("order_date") \
    .option("overwriteSchema", "true") \
    .save("s3a://datalake/silver/order_items/")

from delta.tables import DeltaTable

silver_table = DeltaTable.forPath(spark, "s3a://datalake/silver/order_items/")

# Новые/изменённые заказы из Bronze
new_batch = (
    bronze_orders
    .filter(col("updated_at") > last_processed_ts)
    .select(col("order_id"), col("customer_id"), inline_outer("line_items"))
)

# MERGE: обновляем существующие, вставляем новые
silver_table.alias("silver").merge(
    new_batch.alias("new"),
    "silver.order_id = new.order_id AND silver.sku = new.sku"
).whenMatchedUpdateAll() \
 .whenNotMatchedInsertAll() \
 .execute()

# Заказы содержат позиции (items) и историю статусов (status_history)
# Создаём две Silver-таблицы

# Silver: позиции
silver_items = bronze_orders.select(
    "order_id", "customer_id",
    inline_outer("items")
).write.format("delta").save("s3a://datalake/silver/order_items/")

# Silver: история статусов (отдельная таблица!)
silver_statuses = bronze_orders.select(
    "order_id",
    posexplode_outer("status_history").alias("step_num", "status_record"),
    col("status_record.status"),
    col("status_record.changed_at"),
    col("status_record.changed_by"),
).write.format("delta").save("s3a://datalake/silver/order_statuses/")

# Регистрируем Bronze-таблицу как temporary view
bronze_orders.createOrReplaceTempView("bronze_orders")

# Аналитика: топ-10 SKU по выручке за последние 30 дней
result = spark.sql("""
    WITH order_items AS (
        SELECT
            o.order_id,
            o.customer_id,
            o.order_date,
            item.sku,
            item.product_name,
            item.category,
            item.qty,
            item.unit_price,
            item.discount_pct,
            ROUND(item.qty * item.unit_price * (1 - item.discount_pct), 2) AS line_total
        FROM bronze_orders o
        LATERAL VIEW OUTER explode(o.items) tmp AS item
        WHERE o.order_date >= DATE_SUB(CURRENT_DATE(), 30)
          AND o.status = 'completed'
    )
    SELECT
        sku,
        product_name,
        category,
        SUM(qty)         AS units_sold,
        SUM(line_total)  AS revenue,
        COUNT(DISTINCT order_id) AS orders_count,
        ROUND(AVG(unit_price), 2) AS avg_unit_price
    FROM order_items
    WHERE sku IS NOT NULL  -- фильтруем null от outer
    GROUP BY sku, product_name, category
    ORDER BY revenue DESC
    LIMIT 10
""")

result.show(truncate=False)

# Пример: пользователи → заказы → позиции (три уровня)
deep_data = [(
    1,  # user_id
    [   # orders[]
        {
            "order_id": 101,
            "items": [
                {"sku": "A", "qty": 2},
                {"sku": "B", "qty": 1},
            ]
        },
        {
            "order_id": 102,
            "items": [
                {"sku": "C", "qty": 3},
            ]
        }
    ]
)]

from pyspark.sql.functions import explode, inline

# Шаг 1: разворачиваем заказы
df_orders = df.select(
    col("user_id"),
    explode("orders").alias("order")
).select(
    col("user_id"),
    col("order.order_id"),
    col("order.items"),
)

# Шаг 2: разворачиваем позиции
df_items = df_orders.select(
    col("user_id"),
    col("order_id"),
    inline("items"),  # разворачивает struct-поля
)

df_items.show()

+-------+--------+---+---+
|user_id|order_id|sku|qty|
+-------+--------+---+---+
|      1|     101|  A|  2|
|      1|     101|  B|  1|
|      1|     102|  C|  3|
+-------+--------+---+---+

data = [
    (1, "Intro to Spark",    ["spark", "python", "data-engineering"]),
    (2, "Kafka Streams",     ["kafka", "streaming", "java"]),
    (3, "Delta Lake Guide",  ["delta", "spark", "lakehouse"]),
    (4, "Python Basics",     ["python", "beginner"]),
    (5, "Empty Article",     []),
]

articles = spark.createDataFrame(data, ["article_id", "title", "tags"])

# 1. Топ-5 тегов по частоте использования
from pyspark.sql.functions import explode_outer, count, desc

articles \
    .select("article_id", explode_outer("tags").alias("tag")) \
    .filter(col("tag").isNotNull()) \
    .groupBy("tag") \
    .agg(count("article_id").alias("article_count")) \
    .orderBy(desc("article_count")) \
    .limit(5) \
    .show()

# 2. Статьи с тегом "spark"
articles \
    .select("article_id", "title", explode("tags").alias("tag")) \
    .filter(col("tag") == "spark") \
    .select("article_id", "title") \
    .show()

# 3. Статьи без тегов (используем explode_outer)
articles \
    .select("article_id", "title", explode_outer("tags").alias("tag")) \
    .filter(col("tag").isNull()) \
    .select("article_id", "title") \
    .show()

data = [
    (1, ["landing", "product", "cart", "checkout", "purchase"]),
    (2, ["landing", "product", "cart"]),
    (3, ["landing", "product"]),
    (4, ["landing", "product", "cart", "checkout"]),
]

sessions = spark.createDataFrame(data, ["session_id", "funnel"])

from pyspark.sql.functions import posexplode, count, countDistinct

# Сколько сессий доходит до каждого шага?
sessions \
    .select("session_id", posexplode("funnel").alias("step_num", "step")) \
    .groupBy("step_num", "step") \
    .agg(countDistinct("session_id").alias("sessions_reached")) \
    .orderBy("step_num") \
    .show()

+--------+----------+----------------+
|step_num|      step|sessions_reached|
+--------+----------+----------------+
|       0|   landing|               4|
|       1|   product|               4|
|       2|      cart|               3|
|       3|  checkout|               2|
|       4|  purchase|               1|
+--------+----------+----------------+

# Self-join после explode для нахождения пар
tags_df = articles.select("article_id", explode("tags").alias("tag"))

# Cross-join с самим собой, исключаем пары tag=tag и дубликаты
tag_pairs = tags_df.alias("t1").join(
    tags_df.alias("t2"),
    (col("t1.article_id") == col("t2.article_id")) &
    (col("t1.tag") < col("t2.tag"))  # < чтобы избежать (A,B) и (B,A)
).select(
    col("t1.tag").alias("tag1"),
    col("t2.tag").alias("tag2")
)

tag_pairs.groupBy("tag1", "tag2") \
    .agg(count("*").alias("co_occurrences")) \
    .orderBy(desc("co_occurrences")) \
    .show()