Два explode в Spark размножили позиции: связываем массивы через arrays_zip

Заказ содержит два массива: артикулы [A, B] и количества [2, 3]. После последовательного explode каждого массива появляются четыре строки: A-2, A-3, B-2 и B-3. Каждая операция разворачивает массив для уже существующих строк, поэтому получается произведение элементов.

Если элементы связаны по позиции, сначала нужно объединить их в массив структур. В PySpark для этого подходит arrays_zip. Но до объединения проверьте длины: если массивы не согласованы, заполнение пропусками может скрыть ошибочную поставку.

Сохраним связь по индексу

Учебный заказ содержит два артикула и два количества. Второй заказ намеренно имеет разные длины. Он попадёт в invalid, а валидный заказ развернётся в две позиции. Предполагается, что обе колонки по контракту являются непустыми значениями-массивами, а длина может быть нулевой.

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.master("local[1]").appName("paired-arrays").getOrCreate()
try:
    orders = spark.createDataFrame([
        (1, ["A", "B"], [2, 3]),
        (2, ["C", "D"], [1]),
    ], "order_id long, skus array<string>, qtys array<long>")
    valid_shape = (F.col("skus").isNotNull() & F.col("qtys").isNotNull()
                   & (F.size("skus") == F.size("qtys")))
    invalid = orders.filter(~valid_shape)
    valid = orders.filter(valid_shape)
    positions = valid.select("order_id", F.explode(F.arrays_zip("skus", "qtys")).alias("p"))
    positions = positions.select("order_id", F.col("p.skus").alias("sku"),
                                 F.col("p.qtys").alias("qty"))
    actual = [(r.order_id, r.sku, r.qty) for r in positions.orderBy("sku").collect()]
    assert actual == [(1, "A", 2), (1, "B", 3)]
    assert invalid.count() == 1
    assert positions.agg(F.sum("qty")).first()[0] == 5
    positions.show()
finally:
    spark.stop()

Результат содержит A-2 и B-3, суммарное количество равно пяти. Второй заказ не превращается в частично заполненные позиции: его форма требует разбора. Карантин должен сохранить order_id, обе длины и причину, чтобы поставщик мог исправить соответствие.

arrays_zip при разной длине может добавить NULL для недостающего элемента. Это полезная техническая возможность, но для пары «товар–количество» отсутствие количества обычно не равно нулю. Проверка до разворачивания делает это правило видимым.

Не перенесите сумму заказа на каждую позицию

Если рядом хранится total_amount всего заказа, после explode она повторится для каждой позиции. Суммировать её на уровне развернутых строк нельзя. Для показателей заказов используйте исходную таблицу; для показателей позиций — таблицу positions с суммой отдельных позиций.

Пустые массивы требуют отдельного решения. Обычный explode не создаёт строку для пустого массива, поэтому заказ исчезнет из таблицы позиций. Это естественно для списка товаров. Если отчёт должен сохранить все заказы, считайте его от исходных заказов и присоединяйте агрегат позиций левым соединением.

Проверьте не только размеры

Равная длина не доказывает, что источник сохранил верный порядок. Надёжнее получать массив структур с sku и qty внутри одного элемента. Если формат источника изменить нельзя, зафиксируйте позиционную связь в контракте и проверьте её на известных заказах.

После обработки сравните число сформированных позиций с суммой длин валидных массивов. Затем проверьте допустимость артикулов и положительность количества. Так ошибка формы не смешивается с ошибкой значения, а каждая проверка объясняет конкретное решение загрузчика.

В курсе Data Engineering Fast-Track вы разберёте преобразования Spark, схемы и контракты, затем соберёте конвейер с Kafka и Airflow. Откройте программу и научитесь превращать вложенные события в согласованные таблицы.