Почему объединение по датам + фильтр по временным меткам быстрее, чем объединение по временному диапазону в Spark?

Question

У меня есть фрейм данных X, содержащий некоторые события (моменты времени, с временными метками) и другой фрейм данных Y, содержащий диапазоны времени (также заданные временными метками.)

Путем экспериментов и некоторого чтения Я обнаружил, что базовый подход к соединению по меткам времени напрямую:

return X.join(Y, (X.ts >= Y.start_ts) & (X.ts < Y.end_ts), "inner")

Оказывается намного медленнее, чем первое объединение по датам, а затем фильтрация по определенным временным меткам:

X = X.withColumn("event_date", ts.cast('date'))
Y = Y.withColumn("date", explode(array([start_ts.cast('date')), end_ts.cast('date'))])))
return X \
    .join(Y, (X.event_date == Y.date), "inner") \
    .filter((X.ts >= Y.start_ts) & (X.ts < Y.end_ts))

Насколько я понимаю, базовый подход к выполнению объединения в первом примере:

1. Порядок X на ts и Y на start_ts, а затем end_ts (возможно в nlogn в Spark?)
2. Соединение может быть выполнено линейно, с максимум 2 long сравнениями на кандидата

Во втором примере:

1. Линейно выполнить приведение на сегодняшний день, удваивая размер данных со взрывом
2. Упорядочить оба кадра данных по датам (nlogn, меньшая константа не более 2x)
3. Выполнить объединение линейно, самое большее 1 long сравнение на кандидата
4. Линейная фильтрация результатов с максимум 2 long сравнениями на строку

Достаточно ли меньшей константы из пункта 2 во втором примере, чтобы так сильно ее ускорить? Или есть какая-то оптимизация Spark, которая заставляет Perf вести себя таким образом?

Answer 1

Для начала эти два метода, как правило, не являются логически эквивалентными, поэтому сравнение их времени выполнения практически не имеет смысла.Представьте себе случай, когда Y.start_ts всегда является наименьшим представимым значением, а Y.end_ts является наибольшим представимым значением.Из контекста ясно, что вы не допускаете такой случай, но это знание, специфичное для предметной области, а не то, что может быть легко определено планировщиком.

На самом деле, разница проста, но принципиальна:

• В первом случае просто выполняется декартово произведение, а затем фильтруются данные, всегда проводимые N * M сравнения.
• Во втором случае данные перемешиваются и сравнивается среднее число записей за день в первом наборе * Среднее число записей за день в деньсравнения второго набора.

Подробнее см. Как улучшить скорость соединения с условием в Spark .