Инициация заданий при получении новых файлов

Триггеры прибытия файла можно использовать для активации выполнения задания при поступлении новых файлов во внешнем хранилище, например Amazon S3, хранилище Azure или Google Cloud Storage. Эта функция полезна, если эффективность запланированного задания скомпрометирована нерегулярными новыми поступлениями данных.

Как работают триггеры прибытия файла

Триггеры прибытия файлов делают все возможное, чтобы проверить наличие новых файлов каждую минуту, хотя это может повлиять на производительность базового облачного хранилища. Триггеры прибытия файлов не влекут за собой дополнительных затрат, кроме затрат поставщика облачных услуг, связанных с перечислением файлов в расположении хранилища.

Триггер прибытия файла можно настроить для отслеживания корневого каталога внешнего местоположения или тома в Unity Catalog или подпути внешнего местоположения или тома. Например, для тома /Volumes/mycatalog/myschema/myvolume/каталога Unity ниже приведены допустимые пути для триггера прибытия файла:

/Volumes/mycatalog/myschema/myvolume/
/Volumes/mycatalog/myschema/myvolume/mydirectory/

Триггер прибытия файла рекурсивно проверяет наличие новых файлов во всех подкаталогах настроенного расположения. Например, вы создаёте триггер прибытия файла для расположения /Volumes/mycatalog/myschema/myvolume/mydirectory/, и там имеются следующие подкаталоги:

/Volumes/mycatalog/myschema/myvolume/mydirectory/subdirA
/Volumes/mycatalog/myschema/myvolume/mydirectory/subdirB
/Volumes/mycatalog/myschema/myvolume/mydirectory/subdirC/subdirD

Триггер проверяет наличие новых файлов в mydirectory, subdirA, subdirB, subdirCи subdirC/subdirD.

Триггеры прибытия файла с событиями файлов

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

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

Дополнительные сведения о преимуществах производительности и емкости событий файлов в внешних расположениях см. в разделе "Ограничения". Часто задаваемые вопросы о событиях файлов см. в разделе "Часто задаваемые вопросы о событиях файлов".

Перед тем как начать

Для использования триггеров прибытия файла необходимо следующее:

Добавление триггера прибытия файла

Чтобы добавить триггер прибытия файла в задание:

  1. На боковой панели рабочей области Azure Databricks щелкните Задания и Конвейеры.
  2. При необходимости выберите фильтры "Задания " и " Принадлежащие мне ".
  3. Щелкните по ссылке с названием вашей работы.
  4. В области сведений о задании справа нажмите кнопку "Добавить триггер".
  5. В Тип триггеравыберите прибытие файла.
  6. На размещении хранилища введите URL-адрес корня или подпути внешнего расположения каталога Unity или корня или подпути тома каталога Unity, чтобы выполнять мониторинг.
  7. (Необязательно) Настройте дополнительные параметры (минимальное время между триггерами в секундах и ожиданием после последнего изменения в секундах), чтобы контролировать частоту запуска. Примеры установки см. в разделе "Управление тем, как часто запускаются запуски".
  8. Чтобы проверить конфигурацию, нажмите кнопку "Проверить подключение".
  9. Нажмите кнопку Сохранить.

Чтобы изменить, приостановить или удалить этот триггер позже, используйте раздел "Расписания и триггеры " области сведений о задании . См. раздел "Управление существующим триггером".

Настройте частоту запусков

Два дополнительных параметра триггера прибытия файла определяют, как прибытие файлов преобразуется в выполнение задания. Эти параметры применяют два распространённых шаблона ограничения частоты: период ожидания и устранение дребезга.

  • Минимальное время между срабатываниями в секундах: ограничивает задание до не более одного запуска за указанный интервал (пауза между запусками). После завершения запуска файлы, поступающие в течение периода ожидания, не инициируют новый запуск, пока не истечёт этот интервал. Используйте этот параметр, чтобы ограничить частоту создания запусков, чтобы при частом поступлении не создавались запуски подряд.
  • Ожидание после последнего изменения, в секундах: система ожидает столько времени после поступления последнего файла перед запуском процесса, а каждое новое поступление сбрасывает таймер (устранение дребезга). Используйте этот параметр, когда файлы приходят в пакеты, и вы хотите обработать весь пакет в одном запуске после того, как все файлы приземлились.

Можно задать любой из вариантов самостоятельно или задать оба варианта вместе. См. следующие примеры.

Запускать не чаще чем раз в 15 минут

Чтобы создавать запуски по мере поступления файлов, но не чаще чем раз в 15 минут, задайте следующий дополнительный параметр:

  • Минимальное время между триггерами в секундах: 900

После завершения каждого запуска триггер ожидает 900 секунд (15 минут) до запуска другого запуска, даже если файлы продолжают поступать. Это ограничивает создание запусков: не чаще одного раза в 15 минут.

Дождитесь поступления всей партии

Когда файлы поступают пакетами и вы хотите обрабатывать каждый пакет за один запуск, установите Ожидание после последнего изменения, с на значение меньше, чем интервал между пакетами, но больше, чем интервал между файлами внутри пакета. Например, если новый пакет начинается примерно каждые 5 минут, задайте следующий расширенный параметр:

  • Задержка после последнего изменения (в секундах): 60

Каждый новый файл сбрасывает таймер, поэтому триггер запускает выполнение только если в течение 60 секунд не поступает ни одного нового файла. В этой установке предполагается, что файлы в пакете прибывают в течение 60 секунд друг от друга, поэтому таймер не истекает в середине пакета и что пакеты находятся в более чем 60 секундах друг от друга, поэтому последовательные пакеты не объединяются в один запуск.

Ограничить частоту и дождаться завершения всех пакетов

Вы можете объединить оба варианта, если требуется ограничение частоты создания запусков, а также избегайте запуска в середине пакета. Рассмотрим пример.

  • Минимальное время между триггерами в секундах: 900
  • Ожидание после последнего изменения в секундах: 60

В этой конфигурации триггер ожидает завершения посадки пакета (60 секунд без новых файлов) перед запуском и не запускает более одного запуска в течение 15 минут.

Обнаружение и обработка файлов при поступлении

Для обработки файлов, которые активировали триггеры прибытия файла, можно использовать автозагрузчик. Автозагрузчик постепенно и эффективно обрабатывает новые файлы с гарантиями единственного выполнения. Например, используйте приведенный ниже фрагмент кода для загрузки файлов в таблицу Delta.

Чтобы использовать это решение, создайте задание с триггером прибытия файла и добавьте записную книжку, содержащую приведенный ниже код. Замените каждый [REPLACE] заполнитель соответствующим значением.

# Configuration
file_location = "[REPLACE]" # The same URL configured for the file arrival trigger.
checkpoint_location = "[REPLACE]" # a separate URL (outside `file_location`) used to store the Auto Loader checkpoint, which enables exactly-once processing.
sink_table = "[REPLACE]" # Delta table to write to

# Use Auto Loader to discover new files.
# Do not modify code below this line
streamingQuery = spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "json") \
  .option("cloudFiles.schemaLocation", checkpoint_location) \
  .option("cloudFiles.useManagedFileEvents","true") \
  .load(file_location) \
  .writeStream \
  .option("checkpointLocation", checkpoint_location) \
  .trigger(availableNow = True) \
  .toTable(sink_table)

Если вам нужно обработать новые файлы с помощью пользовательской логики и найти URL-адрес для новых файлов, можно использовать foreachBatch вместо этого, как показано в фрагменте кода ниже. Обратите внимание, что foreachBatch предоставляет гарантии обработки только по крайней мере один раз. Дополнительные сведения об использовании foreachBatchсм. в разделе "Использование foreachBatch для записи в произвольные приемники данных"

# Configuration
file_location = "[REPLACE]" # The same URL configured for the file arrival trigger.
checkpoint_location = "[REPLACE]" # a separate URL (outside `file_location`) used to store the Auto Loader checkpoint, which enables exactly-once processing.

def process_batch(batch_df, batch_id):
  file_url = batch_df.select("path").collect()[0].path
  # [REPLACE] Your custom function for processing newly arrived files


# Use Auto Loader to discover new files.
# Do not modify code below this line
streamingQuery = spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "binaryFile") \
  .option("cloudFiles.useManagedFileEvents","true") \
  .load(file_location) \
  .drop("content") \
  .writeStream \
  .foreachBatch(process_batch) \
  .option("checkpointLocation", checkpoint_location) \
  .trigger(availableNow = True) \
  .start()

Оповещения о сбоях триггеров на прибытие файлов

Чтобы получать оповещения, если триггеру "При получении файла" не удается выполнить оценку, настройте уведомления по электронной почте или системные уведомления о сбое задания. См. , добавьте уведомления для задания.

Ограничения

  • Запуски инициируются только новыми файлами. Перезапись существующего файла с тем же именем не запускает выполнение.
  • Если расположение хранилища включено для событий файлов, добавление содержимого в существующий пустой файл активирует выполнение, так как это рассматривается как новое прибытие файла.
  • События файлов прослушивают FlushWithClose событие для обработки файла. Некоторые операции Azure API могут не генерировать это событие, что может отложить обнаружение файлов. Сведения об этом сценарии см. в разделе "Классические события уведомлений о файлах".

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

  • Путь, используемый для триггера прибытия файла, не может содержать подстановочные знаки, например * или ?.

  • Если расположение хранилища настроено как внешнее расположение в каталоге Unity, а внешнее расположение включено для событий файлов:

    • Нет ограничений на количество файлов в расположении хранилища.

    • Триггеры могут приводить к ошибке тайм-аута, если есть слишком много лишних обновлений файлов.

      Если триггер прибытия файла установлен в подпате внешнего расположения или тома каталога Unity, изменения за пределами этого подпата, например в корневом каталоге внешнего расположения, могут увеличить метаданные, необходимые для обработки триггера. В средах с высоким уровнем изменений триггер может превысить ограничение времени обработки и ввести состояние ошибки.

      Чтобы предотвратить это, создайте том каталога Unity, который сопоставляется специально с подкаталогом, который необходимо отслеживать и задать триггер прибытия файла в корневом каталоге этого тома. Этот подход изолирует целевой путь, делая его основным корневым каталогом триггера, что сокращает несвязанные изменения на уровне корня и предотвращает переход триггера в состояние ошибки.

    • Если существующий файл изменяется, а его метаданные выходят за пределы скользящего периода хранения, это изменение рассматривается как новое прибытие файла, активируя выполнение задания. Это можно предотвратить путем приема только неизменяемых файлов или использовать триггеры прибытия файлов с автозагрузчиком для отслеживания хода приема.

  • Если место хранения не включено для обработки файловых событий:

    • Можно настроить не более 50 заданий с триггером прибытия файла в таких расположениях в рабочей области Azure Databricks.
    • Расположение хранилища может содержать до 10 000 файлов. Если настроенное расположение хранилища является подпутем к внешнему расположению или томом в каталоге Unity, ограничение в 10 000 файлов применяется к данному подпутю, а не к корневому каталогу хранилища. Например, корневой каталог расположения хранилища может содержать более 10 000 файлов в его подкаталогах, но настроенный подкаталог не должен превышать 10 000 файлов.

См. также ограничения событий файлов.

Триггеры прибытия файлов в несуществующие пути в внешних хранилищах S3 и GCS

Если настроенный каталог не существует или удаляется из Amazon S3 или Google Cloud Storage, триггеры прибытия файлов продолжают оцениваться без ошибок. Это происходит, поскольку как S3, так и GCS не различаются между несуществующими, удаленными и пустыми каталогами.

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

  • Last updated on