Transcription
У моей команды была большая проблема. Ну, на самом деле, пять проблем. Представьте себе. Каждое утро понедельника одна из очень талантливых инженеров моей команды приходила в офис, садилась за свой стол и запускала скрипт на Python. Каждую неделю. И хотя это уже звучит как утомительная трата ее времени, становится хуже. Скрипт едва работал. Конечно, он загружал последние данные и запускал все модели нашей команды. Проблема в том, что скрипт не был распараллелен, поэтому он был очень медленным. И если бы даже в одной из наших моделей была ошибка, это могло бы привести к полному сбою всего скрипта. И когда это случалось, у нас оставался только неполный индикатор выполнения и трассировка стека, чтобы попытаться собрать осколки и понять, что пошло не так, начиная все сначала. В этом видео я расскажу вам историю очень важного скрипта, который задержался дольше положенного. Вы услышите о пяти проблемах, которые чуть не довели одну из очень талантливых инженеров моей команды до грани безумия. И я покажу вам, как я использовал всего одну библиотеку, Prefect, чтобы исправить их все. К концу этого видео вы узнаете, как ускорить свои устаревшие скрипты, модернизировав их в приложения для рабочих процессов производства с помощью Prefect. Я использую Prefect с 2019 года, и это просто последний раз, когда он спас меня. Вот почему я был так взволнован, когда Prefect связался со мной, попросив спонсировать видео. Это история, которую я уже планировал рассказать, потому что она оказала такое большое влияние на то, как мы с моей командой работаем. Так что спасибо Prefect за то, что помогли мне представить ее вам всем немного раньше. Давайте приступим к видео. Но сначала позвольте мне задать вам вопрос. Если вы уже проделали всю тяжелую работу по написанию всего рабочего процесса с нуля, например, загрузку последних данных, бэктестинг множества моделей, вычисление метрик, чтобы вы могли выбрать лучшую модель, а затем использование этой модели для получения действенного прогноза, на который полагается вся ваша команда, почему бы вам не остановиться на шаг раньше, чтобы, если вас не будет за столом в понедельник утром для запуска вашего скрипта, весь мир остановился? Я предполагаю, что, когда я излагаю это так, вы предпочтете довести дело до конца и сделать так, чтобы все работало автоматически. В области оркестрации рабочих процессов это называется планированием. Итак, давайте запланируем это. Для этого мы начнем с преобразования нашей основной функции в поток Prefect. Это очень просто. Просто установите Prefect через pip, импортируйте декоратор потока и примените его к нашей функции. Это превращает нашу базовую функцию Python в поток Prefect и открывает множество мощных функций. Поскольку я намерен взять следующий понедельник выходным, давайте запланируем наш поток. Для этого мы будем использовать метод serve нашего потока. Он поддерживает множество различных способов указания, когда должен выполняться наш поток, например, предоставление строки cron, указание интервала времени между запусками или определение строки правила повторения. Если мы запустим наш flow.serve, дадим развернутому потоку имя и предоставим эту строку cron, то наш поток будет запланирован на выполнение каждое утро понедельника в полночь по UTC. И теперь, когда наш поток запланирован, я имею в виду, я думаю, я мог бы просто откинуться назад, закинуть ноги на стол и ждать до утра понедельника, когда... Черт. (Вздох...) Так что здорово, что мы запланировали наш код, но это была только наша первая проблема. Я как-то забыл, что наш скрипт изначально был практически сломан. Так что нам нужно будет сделать наш поток более устойчивым к сбоям, прежде чем мы сможем ожидать его успеха. Посмотрите на эту часть кода здесь. Иногда, когда мы пытаемся получить данные, они просто не удаются, и нам придется продолжать попытки, пока они не увенчаются успехом. Это вне контроля нашей команды, так что это лучшее, что мы можем сделать. О, и я чуть не забыл. Одна из наших моделей все еще немного экспериментальна. Команда, работающая над ней, делает все возможное, но иногда они вводят ошибку. И когда это происходит, весь конвейер рушится. Поэтому обычно мы комментируем эту модель, перезапускаем конвейер, а затем раскомментируем ее, чтобы добавить обратно в конвейер, когда они исправят ошибку. Да. Хотя мы не можем исправить основные сбои, мы можем сделать наш поток более устойчивым к ним. Для этого мы преобразуем наши функции в задачи, используя декоратор task. Круто, но что это делает? Как и потоки, задачи предоставляют нам множество мощных функций, и многие из них специально разработаны для улучшения масштабирования и обработки ошибок. Например, вот так... Для нашей задачи get data мы можем добавить retries равное 42. Теперь Prefect будет продолжать повторять нашу задачу до 42 раз, пока она не увенчается успехом. Она никогда не терпит неудачу так много раз, так что это решает нашу первую проблему. Вторая проблема немного сложнее, так что оставайтесь со мной. Наша цель — отбросить любую модель, которая терпит неудачу. Так что, если бы у нас было что-то, что отдаленно выглядело бы так... Ну, тогда это сработало бы, но откуда бы мы взяли результат? Ну, используя оператор моржового уса, мы можем оценить наше выражение бэктестинга и немедленно присвоить его переменной result. Это позволяет нам сделать что-то вроде этого... но это все еще не решает нашу проблему, потому что наша функция бэктестинга не возвращает ошибку при сбое, она ее генерирует. Так что это все равно вызовет сбой. К счастью, это не проблема. Вот как мы можем это исправить. Вместо того, чтобы вызывать нашу задачу напрямую, мы можем использовать ее метод submit. Здесь submit немедленно возвращает будущий объект Prefect, который по сути является долговой распиской. Нам придется вернуться позже, чтобы получить готовый результат. Но тем временем мы можем выполнять другую работу в нашей основной функции. Чтобы получить результат, мы можем вызвать метод result на нашем будущем объекте. Это дождется завершения нашей задачи и затем вернет ее результат. Теперь вы, вероятно, думаете, что это, по сути, то же самое, с чего мы начали, за исключением дополнительных шагов. И да, я понимаю. Но метод result на самом деле имеет очень полезный аргумент. Если мы установим raise_on_failure в false, то Prefect вернет ошибку как значение, а не сгенерирует ее как исключение. Со всеми этими изменениями наш поток теперь всегда будет успешным, даже если произойдут некоторые сбои, которые нам пришлось обрабатывать. Так что, если вы дадите мне одну минуту, я быстро что-нибудь напечатаю. Дорогой босс, я буду вне офиса сегодня, но вы можете найти результаты конвейера за эту неделю на нашем сервере Prefect здесь. Спасибо, Даг. Одной из моих любимых функций Prefect является его веб-интерфейс. И хорошая новость в том, что мне больше не нужно писать код, чтобы им пользоваться. Мы можем запустить локальный сервер Prefect, выполнив команду prefect server start, или мы можем использовать бесплатный тариф Prefect Cloud для еще более классных функций. Все, что нам нужно сделать, это создать учетную запись на веб-сайте Prefect, создать рабочее пространство, а затем выполнить команду prefect cloud login. Здесь я использую опцию login with browser, потому что это просто очень легко. Давайте начнем с вкладки flows. Нажав на наш поток, мы видим предстоящие запланированные потоки, а также все наши прошлые запуски потоков из нашего тестирования. Давайте нажмем на последний запуск. Мы можем фактически видеть представление всех наших задач в реальном времени, включая когда они начинаются, когда они заканчиваются и какой у них статус. И все это в графическом представлении, показывающем, какие задачи зависят друг от друга. Это первое место, куда я обращаюсь, когда пытаюсь отладить ошибку. Это просто дает вам так много информации с первого взгляда. Далее, если мы посмотрим на вкладку logs, мы фактически увидим, что Prefect автоматически добавил логирование в наше приложение. Мы получили это бесплатно, потому что мы сделали наши задачи, ну, задачами Prefect. Это уже очень полезно, но Prefect Cloud идет еще дальше. Когда что-то идет не так, вам не обязательно копаться в сообщениях журнала, чтобы понять, что пошло не так. Именно поэтому Prefect использует AI-суммирование для автоматического извлечения сбоя и размещения его на вашей панели. Таким образом, вы можете быстрее обнаруживать и устранять свои сбои. Если вы управляете несколькими потоками, это может даже помочь вам расставить приоритеты, какой сбой следует устранить первым. Но если вы хотите углубиться в детали, мы можем перейти на вкладку task run. Там мы можем получить информацию о статусе, времени выполнения и журналах для каждого из наших запусков задач. Но я только что заметил кое-что. Для задач, которые мы запускаем несколько раз, например, когда мы выполняем бэктестинг на нескольких моделях, может быть немного запутанно, какой бэктест соответствует какой модели. Вам, вероятно, потребуется немного больше информации, чем просто имя задачи 0, 1 и 2. Для этого существует аргумент task run name. Это позволяет нам дать нашему запуску задачи пользовательское имя на основе его аргументов. Если мы используем имя backtest, а затем фигурные скобки, model name, своего рода как f-строка, но без F, то в следующий раз, когда мы запустим наш поток, наши модели будут backtest_slow model, backtest_buggy model, backtest и так далее. Теперь совершенно очевидно, что наша медленная модель медленная. Ах, отлично. Я только что подписался на еще одну еженедельную задачу. К счастью, Prefect Cloud имеет очень мощную систему автоматизации, управляемую событиями. Мы можем использовать ее для автоматизации этого электронного письма. Перейдите на вкладку automations и нажмите add a new automation. Мы хотим, чтобы наша автоматизация срабатывала, когда наш поток переходит в состояние completed. Теперь мы можем настроить действие, которое происходит при срабатывании нашего события. Я хочу отправить уведомление. После нажатия кнопки add мы можем выбрать блок email. Давайте назовем этот блок notify-stakeholders и укажем адрес электронной почты моего босса. Мы можем настроить тему и содержимое электронного письма. Мы можем использовать шаблонизацию Jinja для включения информации, специфичной для Prefect, но я просто оставлю сообщение по умолчанию на данный момент. Затем мы просто даем нашей автоматизации имя, и все готово. Теперь, когда наш поток будет успешным, Prefect автоматически отправит это электронное письмо. Если я хочу пойти еще дальше, я могу даже включить такую информацию, как прогноз на эту неделю, чтобы наши заинтересованные стороны имели еще больше информации в этом письме. Все это потрясающе. И в пользовательском интерфейсе Prefect есть еще больше возможностей. Но да, подождите минуту. Посмотрите на это. Если мы вернемся к нашему графическому представлению, мы тратим так много времени на обработку наших бэктестов и метрик по одному. Нам не нужно этого делать. Каждый раз, когда мы отправляем задачу, мы немедленно ждем ее результата. Но нам не нужно этого делать. Если между задачами нет зависимостей, в принципе, они все могут выполняться одновременно. Вот так, если мы отправим задачу, а затем еще одну задачу, а затем еще одну задачу, ну, в принципе, они все могут обрабатываться одновременно. Мы фактически используем одну и ту же задачу и просто передаем в нее множество различных входных данных из итерируемого контейнера. Это настолько распространенная проблема, что Prefect фактически имеет для нее специальный метод, task.map. Так давайте используем его. Для бэктестов мы можем отобразить ключи models.keys, чтобы динамически создать запуск задачи для каждой модели. Но нам также нужно передать весь наш набор данных, и мы не хотим случайно его итерировать. Мы можем использовать аннотацию unmapped, чтобы просто передать весь набор данных. Таким образом, мы не итерируем по нему. Task.map возвращает список будущих объектов Prefect, и они находятся в том же порядке, что и входные данные, которые мы передали. Нам все еще нужно в конечном итоге получить наши результаты и отфильтровать любые сбои. Мы можем сделать это с помощью словарного включения, похожего на то, как мы делали ранее. Мы можем сделать то же самое для наших метрик. Запустить metrics.map, передать весь наш набор данных unmapped, а затем отобразить наши прогнозы. После этого мы можем создать наш словарь метрик, как мы делали раньше. И теперь, если мы снова запустим наш поток, мы увидим, что наши бэктесты и метрики выполнялись параллельно, что потрясающе. По умолчанию Prefect использует параллельный исполнитель задач. Он хорошо работает для нашего игрушечного примера и очень хорош, если у вас много операций ввода-вывода или сетевых вызовов. Но если вы хотите выйти на новый уровень и достичь истинного параллелизма на уровне ЦП для задач, ограниченных ЦП, тогда я очень рекомендую библиотеки PrefectDask или PrefectRay. Они интегрируются непосредственно с Prefect, и вы можете использовать исполнители задач PrefectDask или PrefectRay. Эти исполнители могут автоматически развернуть локальный кластер, если вам нужен базовый параллелизм на уровне ЦП, или вы можете подключить их к существующим выделенным кластерам, если вам нужна большая мощность, чем может предоставить ваше локальное устройство. Но пока я просто буду придерживаться параллельного исполнителя задач, потому что я не думаю, что кластер может сделать это быстрее. Наш код действительно формируется, но все еще есть много дел. Что это вообще такое? О, я имею в виду, хорошо. Это просто жалуется, что каждый раз, когда мы запускаем наш поток, мы фактически перезаписываем результаты прошлой недели. Это нехорошо, но есть простое решение. Мы можем использовать систему артефактов Prefect для создания табличных или markdown артефактов. Артефакты в Prefect — это просто хранимые фрагменты данных, которые генерируются в запусках потоков или задач. И самое главное, они не перезаписываются между запусками. Здесь мы можем использовать функцию create_markdown_artifact, чтобы заменить эту строку, где мы записываем наш отчет в файл. Или здесь мы можем использовать функцию create_table_artifact, чтобы сохранить наш словарь метрик или наш окончательный прогноз. Мы можем найти наши артефакты на страницах запуска потока или задачи в пользовательском интерфейсе Prefect. Или, что еще лучше, поскольку мы назвали наши артефакты, мы можем найти их на вкладке Artifacts. Давайте нажмем на артефакт next week prediction. Мы можем легко просмотреть прогнозы предыдущих недель, чтобы выяснить, какие модели мы использовали и что они предсказали. Давайте подведем итоги пяти проблем, которые Prefect помог нам решить. Номер один, мы спасли чье-то утро понедельника, запланировав наш код на Python с помощью Prefect. Номер два, мы добавили повторные попытки и обработку ошибок для плавного разрешения и восстановления от непредвиденных сбоев. Номер три, мы получили очень полезный веб-интерфейс, который предоставляет нам всю информацию, которая нам может понадобиться для отладки и мониторинга наших конвейеров данных и реагирования на неожиданные ситуации. Номер четыре, мы значительно ускорили наш конвейер, запуская множество наших задач параллельно. И если нам нужно масштабировать наши вычислительные возможности за пределы того, что может предоставить наше локальное устройство, Prefect позволяет очень легко запускать наши задачи на внешнем оборудовании или на локальных вычислительных кластерах. И номер пять, мы получили простой способ поддерживать историю всех данных и отчетов, которые генерирует наш рабочий процесс, через систему артефактов Prefect. Prefect позволяет вам создавать, наблюдать и реагировать на всевозможные конвейеры данных. И я только царапаю поверхность всех потрясающих вещей, которые может делать Prefect. Облачный продукт Prefect выводит вашу оркестрацию рабочих процессов на новый уровень благодаря истинной сквозной наблюдаемости. Так что вам никогда не придется гадать, что происходит с вашими рабочими процессами. Чтобы получить доступ к мощной системе автоматизации Prefect, AI-суммированию журналов или контролю доступа на основе ролей, вам следует зарегистрироваться на совершенно бесплатном тарифе Prefect Cloud, отсканировав QR-код на экране или нажав на ссылку вверху описания видео сегодня. Использование этой ссылки действительно помогает каналу. Если вы дошли до конца видео, пожалуйста, поставьте лайк видео. Это помогает видео достичь большего количества замечательных людей, таких как вы. И если вам действительно понравилось видео, то вам следует посмотреть следующее видео, чтобы узнать, почему скомпилированный Python невероятно быстр и на удивление прост в использовании. Спасибо за просмотр.