Поделиться через

Работа с соединениями в Azure Databricks

  • Статья

Databricks поддерживает стандартный синтаксис соединения ANSI. В этой статье описываются различия между соединениями с пакетной и потоковой обработкой и приведены некоторые рекомендации по оптимизации производительности соединения.

Примечание.

Databricks также поддерживает стандартный синтаксис для операторов UNIONнабора , INTERSECTа EXCEPTтакже . См. раздел "Задать операторы".

Различия между потоковой передачей и пакетными соединениями

Соединения в Azure Databricks являются отслеживанием состояния или без отслеживания состояния.

Все пакетные соединения являются соединениями без отслеживания состояния. Результаты обрабатываются немедленно и отражают данные во время выполнения запроса. Каждый раз при выполнении запроса новые результаты вычисляются на основе указанных исходных данных. См . статью "Пакетные соединения".

Соединения между двумя источниками данных потоковой передачи являются отслеживанием состояния. В соединениях с отслеживанием состояния Azure Databricks отслеживает сведения о источниках данных и результатах и итеративно обновляет результаты. Соединения с отслеживанием состояния могут предоставлять мощные решения для обработки данных в сети, но могут оказаться трудными для эффективной реализации. Они имеют сложную операционную семантику в зависимости от выходного режима, интервала триггера и водяного знака. См . статью "Соединения потокового потока".

Статические соединения потока являются бессерверными, но предоставляют хороший вариант для присоединения добавочного источника данных (например, таблицы фактов) со статическим источником данных (например, медленно изменяющейся размерной таблицей). Вместо объединения всех записей с обеих сторон каждый раз при выполнении запроса только недавно полученные записи из источника потоковой передачи присоединяются к текущей версии статической таблицы. См. статью "Поток-статические соединения".

Пакетные соединения

Azure Databricks поддерживает стандартный синтаксис соединения SQL, включая внутренние, внешние, полу, анти и кросс-соединения. См. раздел JOIN.

Примечание.

Databricks рекомендует использовать материализованное представление для оптимизации добавочного вычисления результатов внутреннего соединения. См. статью "Использование материализованных представлений" в Databricks SQL.

Соединения потокового потока

Присоединение двух источников потоковых данных может представлять значительные проблемы в управлении сведениями о состоянии и обоснованием вычислений результатов и выходных данных. Перед реализацией соединения потоковой передачи Databricks рекомендует разработать четкое представление о операционной семантике потоковой передачи с отслеживанием состояния, включая влияние управления состоянием подложки. См. следующие статьи:

Databricks рекомендует указывать подложки для обеих сторон всех соединений потокового пара. Поддерживаются следующие типы соединения:

  • Внутренние соединения
  • Левые внешние соединения
  • Правые внешние соединения
  • Полные внешние соединения
  • Левое полусоединяние

См. документацию по структурированной потоковой передаче Apache Spark в соединениях потоковой передачи.

Статические соединения потоковой передачи

Примечание.

Описанное поведение для потоковых статических соединений предполагает, что статические данные хранятся с помощью Delta Lake.

Потоковое статическое объединение объединяет последнюю допустимую версию дельта-таблицы (статические данные) с потоком данных с помощью объединения без сохранения состояния.

Когда Azure Databricks обрабатывает микропакет данных в потоковом статическом объединении, последняя действительная версия данных из статической разностной таблицы объединяется с записями, представленными в текущем микропакете. Так как объединение не имеет состояния, вам не нужно настраивать водяные знаки, и вы можете обрабатывать результаты с малой задержкой. Данные в статической дельта-таблице, используемой в объединении, должны изменяться медленно.

В следующем примере показан этот шаблон:

streamingDF = spark.readStream.table("orders")
staticDF = spark.read.table("customers")

query = (streamingDF
  .join(staticDF, streamingDF.customer_id==staticDF.id, "inner")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .table("orders_with_customer_info")
)

Оптимизация производительности при объединении

Вычисление с включенной функцией Photon всегда выбирает лучший тип соединения. См. статью " Что такое Фотон?".

Использование последней версии среды выполнения Databricks с поддержкой Photon обычно обеспечивает хорошую производительность соединения, но следует также рассмотреть следующие рекомендации:

  • Перекрестные соединения очень дороги. Удалите перекрестные соединения из рабочих нагрузок и запросов, требующих низкой задержки или частой повторной компиляции.
  • Вопросы заказа соединения. При выполнении нескольких соединений всегда присоединяйте наименьшие таблицы, а затем присоединяйте результат к большим таблицам.
  • Оптимизатор может бороться с запросами с множеством соединений и агрегатов. Сохранение промежуточных результатов может ускорить планирование запросов и вычисление результатов.
  • Сохраняйте свежие статистические данные, чтобы повысить производительность. Запустите запрос ANALYZE TABLE table_name COMPUTE STATISTICS , чтобы обновить статистику в планировщике запросов.

Примечание.

В Databricks Runtime 14.3 LTS и более поздних версиях можно изменить столбцы, которые Delta Lake собирает статистику для пропуска данных, а затем повторно компьютировать существующую статистику в журнале Delta. См. раздел "Указание столбцов статистики delta".

Указания на присоединение к Azure Databricks

Apache Spark поддерживает указание подсказок соединения для соединений диапазона и перекоса соединений. Подсказки для скошенных соединений не необходимы, так как Azure Databricks автоматически оптимизирует эти соединения. См . подсказки

Указания для соединений диапазона могут оказаться полезными, если производительность соединения плоха, и вы выполняете соединения с неравенством. Примеры включают присоединение к диапазонам меток времени или диапазону идентификаторов кластеризации. См . статью "Оптимизация соединения диапазона".