data pipeline что такое

What Is a Data Pipeline?

A data pipeline is a series of data processing steps. If the data is not currently loaded into the data platform, then it is ingested at the beginning of the pipeline. Then there are a series of steps in which each step delivers an output that is the input to the next step. This continues until the pipeline is complete. In some cases, independent steps may be run in parallel.

Data pipelines consist of three key elements: a source, a processing step or steps, and a destination. In some data pipelines, the destination may be called a sink. Data pipelines enable the flow of data from an application to a data warehouse, from a data lake to an analytics database, or into a payment processing system, for example. Data pipelines also may have the same source and sink, such that the pipeline is purely about modifying the data set. Any time data is processed between point A and point B (or points B, C, and D), there is a data pipeline between those points.

Build ultra-fast data pipelines with Hazelcast stream processing. Watch this

As organizations look to build applications with small code bases that serve a very specific purpose (these types of applications are called “microservices”), they are moving data between more and more applications, making the efficiency of data pipelines a critical consideration in their planning and development. Data generated in one source system or application may feed multiple data pipelines, and those pipelines may have multiple other pipelines or applications that are dependent on their outputs.

Consider a single comment on social media. This event could generate data to feed a real-time report counting social media mentions, a sentiment analysis application that outputs a positive, negative, or neutral result, or an application charting each mention on a world map. Though the data is from the same source in all cases, each of these applications are built on unique data pipelines that must smoothly complete before the end user sees the result.

Common steps in data pipelines include data transformation, augmentation, enrichment, filtering, grouping, aggregating, and the running of algorithms against that data.

What Is a Big Data Pipeline?

As the volume, variety, and velocity of data have dramatically grown in recent years, architects and developers have had to adapt to “big data.” The term “big data” implies that there is a huge volume to deal with. This volume of data can open opportunities for use cases such as predictive analytics, real-time reporting, and alerting, among many examples.

Like many components of data architecture, data pipelines have evolved to support big data. Big data pipelines are data pipelines built to accommodate one or more of the three traits of big data. The velocity of big data makes it appealing to build streaming data pipelines for big data. Then data can be captured and processed in real time so some action can then occur. The volume of big data requires that data pipelines must be scalable, as the volume can be variable over time. In practice, there are likely to be many big data events that occur simultaneously or very close together, so the big data pipeline must be able to scale to process significant volumes of data concurrently. The variety of big data requires that big data pipelines be able to recognize and process data in many different formats—structured, unstructured, and semi-structured.

Data Pipeline vs. ETL

ETL refers to a specific type of data pipeline. ETL stands for “extract, transform, load.” It is the process of moving data from a source, such as an application, to a destination, usually a data warehouse. “Extract” refers to pulling data out of a source; “transform” is about modifying the data so that it can be loaded into the destination, and “load” is about inserting the data into the destination.

ETL has historically been used for batch workloads, especially on a large scale. But a new breed of streaming ETL tools are emerging as part of the pipeline for real-time streaming event data.

Data Pipeline Considerations

Data pipeline architectures require many considerations. For example, does your pipeline need to handle streaming data? What rate of data do you expect? How much and what types of processing need to happen in the data pipeline? Is the data being generated in the cloud or on-premises, and where does it need to go? Do you plan to build the pipeline with microservices? Are there specific technologies in which your team is already well-versed in programming and maintaining?

Architecture Examples

Data pipelines may be architected in several different ways. One common example is a batch-based data pipeline. In that example, you may have an application such as a point-of-sale system that generates a large number of data points that you need to push to a data warehouse and an analytics database. Here is an example of what that would look like:

A basic example of a data pipeline.

Another example is a streaming data pipeline. In a streaming data pipeline, data from the point of sales system would be processed as it is generated. The stream processing engine could feed outputs from the pipeline to data stores, marketing applications, and CRMs, among other applications, as well as back to the point of sale system itself.

This diagram models a streaming data pipeline. The data stream is is managed by the stream processing framework where it can be processed and delivered to apps and/or solutions.

A third example of a data pipeline is the Lambda Architecture, which combines batch and streaming pipelines into one architecture. The Lambda Architecture is popular in big data environments because it enables developers to account for both real-time streaming use cases and historical batch analysis. One key aspect of this architecture is that it encourages storing data in raw format so that you can continually run new data pipelines to correct any code errors in prior pipelines, or to create new data destinations that enable new types of queries.

The Lambda Architecture accounts for both a traditional batch data pipeline and a real-time data streaming pipeline. It also has a serving layer that responds to queries.

Источник

Что такое конвейеры Машинного обучения Azure?

В этой статье вы узнаете, как конвейер машинного обучения помогает создавать, оптимизировать и администрировать рабочий процесс машинного обучения.

Какую технологию конвейера Azure следует использовать?

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

Что могут выполнять конвейеры машинного обучения?

Конвейер Машинного обучения Azure содержит автоматизированный рабочий процесс, выполняющий полную задачу машинного обучения. Подзадачи инкапсулируются в последовательность шагов в конвейере. Конвейер Машинного обучения Azure может быть очень простым, например содержать один шаг для вызова скрипта Python, потому может делать примерно все. Конвейеры должны сосредоточиться на задачах машинного обучения, таких как:

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

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

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

Метрики для экспериментов конвейера можно отслеживать непосредственно в портале Azure или на целевой странице рабочей области (предварительная версия). После публикации конвейера можно настроить конечную точку RESTFUL, которая позволит перезапустить конвейер из любой платформы или стека.

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

Анализ зависимостей

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

Анализ зависимостей в конвейерах Машинное обучение Azure является более сложным, чем простые метки времени. Каждый шаг может выполняться в различных аппаратных и программных средах. Подготовка данных может быть трудоемким процессом, но не требовать оборудования с мощными графическими процессорами. Для некоторых шагов может потребоваться программное обеспечение, предназначенное для конкретной операционной системы. Возможно, вы предпочтете использовать распределенное обучение и т. п.

Машинное обучение Azure автоматически управляет всеми зависимостями между этапами конвейера. Это может привести к ускорению и понижению работы образов DOCKER, подключению и отсоединению вычислений и перемещению данных между действиями в единообразном и автоматическом режиме.

Координация необходимых действий

При создании и запуске объекта Pipeline выполняются следующие общие действия.

Создание конвейеров с помощью пакета SDK для Python

Конвейер Машинного обучения Azure связан с рабочей областью Машинного обучения Azure, а этап конвейера связан с целевым объектом вычислений, доступным в этой рабочей области. Дополнительные сведения см. в статьях создание рабочих областей машинное обучение Azure и управление ими в портал Azure или целевые объекты вычислений в машинное обучение Azure?.

Простой конвейер Python

В этом фрагменте кода показаны объекты и вызовы, необходимые для создания и выполнения Pipeline :

Затем код создает экземпляр Pipeline самого объекта, передавая в него рабочую область и массив шагов. Вызов начинается с experiment.submit(pipeline) запуска конвейера машинного обучения Azure. Вызов wait_for_completion() блокируется до завершения конвейера.

Создание конвейеров с помощью конструктора

Разработчики, предпочитающие визуальную область конструктора, могут использовать конструктор Машинное обучение Azure для создания конвейеров. Доступ к этому средству можно получить с помощью выбора конструктора на домашней странице рабочей области. Конструктор позволяет перетаскивать шаги в область конструктора.

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

Ключевые преимущества

Основные преимущества использования конвейеров для рабочих процессов машинного обучения:

Дальнейшие действия

См. справочную документацию по пакету SDK, ядру конвейера и этапам конвейера.

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

Источник

Использование Pipeline в работе с данными

В своей работе Data Scientist используют различные модели для улучшения качества метрик. Чтобы применить модель, предварительно необходимо затратить существенные ресурсы на обработку всего массива необработанных данных. Мы расскажем об инструменте, которым пользуемся для оптимизации этого процесса. Инструмент Pipeline позволяет объединить несколько операций обработки данных в единую модель библиотеки Python «Scikit-learn».

Рассмотрим его применение более подробно.

Класс Pipeline предусматривает методы fit, predict и score, имеющие свойства, аналогичные свойствам модели в библиотеке «Sckit-learn». В Машинном обучении чаще всего класс Pipeline используется для объединения операций предварительной обработки (например, масштабирования данных или one-hot-encoding) с моделью машинного обучения типа классификаторов. Его использование позволяет избежать ошибок и сокращает временные издержки.

Вот как это выглядит схематически.

Другими словами, применяется определённая последовательность действий к необработанным данным (в первую очередь осуществляется трансформирование данных, а уже после: масштабирование числовых переменных, one-hot-encoding и так далее). В принципе в T1(), T2() …Tn() может быть любой код по предобработке данных.

Основное преимущество Pipeline:

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

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

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

Теперь рассмотрим инструмент Pipeline с дополнительным набором функций:

Таким образом код с использованием Pipeline будет выглядеть следующим образом:

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

Вот так, применение инструментов Pipeline позволяет оптимизировать работу Data Scientist, что способствует повышению качества и скорости работы в целом.

Источник

Строим Data Pipeline на Python и Luigi

Введение

В эпоху data-intensive приложений рядовым разработчикам всё чаще приходится сталкиваться с задачами по обработке и анализу данных. Ещё десять лет назад данные большинства проектов могли уместиться на жестком диске одного компьютера в какой-нибудь реляционной базе данных типа MySQL. А задачи по извлечению и обработке хранящихся данных решались за счёт непростых (или простых) SQL запросов. С тех пор мир информационных технологий значительно поменялся. С приходом Internet of Things, мобильных телефонов и дешевого мобильного интернета, объем генерируемых данных вырос в десятки тысяч раз. Ежедневно в мире генерируются эксабайты данных. Анализировать такой поток информации вручную, а тем более извлекать полезные для бизнеса или науки данные, практически невозможно. Но технологии как и время не стоят на месте, появляются новые инструменты, наука двигает прогресс. Если вы хоть чуточку следите за новостями из мира высоких технологий, то фразы «биг дата», «машинное обучение», «глубокое обучение» вас не испугают. С приходом больших данных появились новые профессии и специализации такие как Data Scientist/Analyst (по-русски аналитик данных), Data Engineer. Задачи этих ребят тесно связаны с обработкой, анализом и хранением «нефти 21 века», т.е. информации. Но насколько эффективно они выполняются?

Аббревиатура ETL в последнее время часто мелькает в материалах, посвященных data-driven приложениям. Но не пугайтесь, это всего лишь набор из 3-х простых слов: Extract, Transform, Load. Ничего не напоминает? Тот, кто сталкивался с задачами по обработке данных не раз замечал паттерн в своих действиях, а именно:

сначала данные выгружаются (Extract) из какого-нибудь источника типа базы данных, внешнего сервиса (Facebook Ads, Google Analytics, Yandex Metrics) или, на худой конец, это могут быть логи вашего приложения (например, веб-сервера).

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

и наконец загружаются (Load) для просмотра и дальнейшего анализа в базу данных или на какое-нибудь облако Amazon S3, не суть.

И как ни крути от этого не уйти. Чтобы данные проанализировать, их необходимо подготовить, иначе «мусор на входе — мусор на выходе». Процесс подготовки занимает львиную долю времени, отведенного на работу с данными. До 80% рабочего времени аналитик тратит на сбор и очистку. Поэтому от эффективности ETL-процесса зависит скорость и качество выполненной работы.

Перед тем как перейти к основной идеи этой статьи, я предлагаю кратко рассмотреть самый популярный на сегодняшний день метод построения ETL процесса в компании.

Внимание! Я запустил полноценный курс по разработке дата-пайплайнов на Luigi. Luigi сильно недооценён, и к нему стоит присмотреться поближе. В курсе я рассказываю зачем нужны пайплайны, как их сделать надёжными и отказоустойчивыми. Всё это заправляется практическими примерами! Не забыл я и про тему деплоя, где затрагивается Docker контейнеры, а также облачный деплой дата-пайпланой в AWS с использованием таких технологий как AWS Fargate, Cloud Map, Elastic Container Service и другое.

Серые будни работы с данными

Сложно спорить с утверждением, что Python твердо занял позицию lingua franca в задачах по анализу данных. О его взрывной популярности свидетельствует и недавний пост от ребят из Stack Overflow. Но что же предшествует магическому процессу извлечения ценной информации (непосредственному анализу) из гигабайт структурированных и неструктурированных данных? Сбор. Задачи по анализу данных славятся своими жесткими сроками ведь на их основе часто принимают ключевые бизнес-решения. Это сильно отражается на том как мы, разработчики, подходим к процессу написания скриптов. Нам не стыдно создавать скрипты-однодневки с хрупким кодом и горой «разбитых окон». Хорошо ещё, если в этой массе кода есть хоть какая-нибудь структура или модульность для дальнейшего переиспользования в других скриптах, но обычно и этого нет. Краткосрочный выигрыш в скорости оборачивается головными болями в долгосрочной перспективе. Код обрастает «техническим долгом», и им становится сложно управлять. Ниже пример ETL-скрипта с соблюдением принципа Single Responsibility в каждой функции:

Всё бы хорошо, но у такого подхода есть ряд проблем:

Все они решаемы, но нужно ли изобретать ещё один, когда он давно изобретен и на нём успешно «катаются» специалисты в индустрии?!

Luigi

Luigi это один из немногих инструментов в экосистеме Python для построения т.н. pipeline’ов или, по-простому, выполнения пакетных задач (batch jobs). Разработан был инженерами из Spotify. Мне он понравился за свою простоту и широкий спектр возможностей, а именно:

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

failover recovery, т.е. если в одной из задач произошла ошибка, не нужно перезапускать цепочку снова

центральный планировщик задач с веб-интерфейсом, статусом выполнения задач и трекингом ошибок

“батарейки” для работы с HDFS, S3, MySQL, PostgreSQL, Redis, MongoDB, Redshift и т.д.

удобное построение CLI (Command Line Interface), в нём очень удобно построена передача параметров из командной строки

Основными строительными блоками Luigi являются 3 объекта: Task, Target и Parameter. Последний используется для взаимодействия с командной строкой и поэтому опционален. Чтобы установить Luigi достаточно выполнить:

Класс Task это основной блок, где происходит выполнение конкретного таска. Чтобы определить свою собственную задачу, необходимо создать класс, унаследованный от Task, и реализовать несколько методов. Зачастую переопределять нужно только 3 метода: run(), output(), requires().

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

Task.run

Здесь выполняется вся логика вашей будущей задачи, например, скачивание или парсинг данных с внешнего источника, запрос в базу данных для извлечения информации и т.д. Если задача объёмная, то лучше разбить её на функции и вызывать их внутри метода run(), это поможет избежать путанницы в будущем.

Task.requires

Помните я говорил об управлении зависимостями? В методе requires() необходимо их перечислить. Зависимостями выступают другие luigi.Task классы. Чуть позже я покажу реальный пример задачи с зависимостями.

Task.output

Этот метод должен возвращать 1 или более Target объектов. Target объектом может быть файл на диске, файл внутри HDFS, S3 или файл, лежащий на удалённом FTP сервере и т.д.. В Luigi уже встроено множество полезных Target классов, поэтому ситуация, когда вам понадобится создавать свой, маловероятна. Полный список доступных Target классов смотрите на сайте.

Task.input

Этот метод не нужно переопределять. Он выступает «оберткой» над Task.requires и возвращает Target объекты, полученные от выполнения задач, определенных в Task.requires. Таким образом строится граф зависимостей, когда одна задача зависит от результата выполнения другой. Продемонстрирую на примере кода:

Здесь таск B зависит от выполнения таска A, поэтому перед началом выполнения B выполнится A, результат которого вернётся при вызове метода B.input (объекта файла result.txt).

Target

Ранее я вкратце описал что из себя представляет объект Target и зачем он нужен. Здесь отмечу, что благодаря этому классу Luigi реализует механизм fault tolerance и свойство идемпотентности. Проще говоря, если ваш pipeline аварийно завершается где-то в середине выполнения задач, повторный запуск не приведёт к повторному запуску успешно завершившихся задач, выполнение начнется в месте аварийной остановки скрипта. Это достигается за счёт вызова метода exists() у Target класса.

Parameter

При создании ETL скриптов часто приходится писать код для работы с командной строкой, а именно уметь принимать и обрабатывать аргументы. Даже наличие в стандартной библиотеке Python модулей для работы с консолью не уменьшает количество boilerplate кода. Luigi решил эту проблему по-своему.

Чтобы принимать аргументы из командной строки достаточно присвоить переменной объект класса Parameter или его наследников на уровне класса.

Пример запуска такого скрипта:

Luigid

Задача демона Luigi заключается в следующем:

Следить за выполнением задач, чтобы исключить ситуацию одновременного исполнения одного и того же таска

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

Ниже скриншот графа зависимостей внутри демона Luigi на примере простого скрипта о котором расскажу чуть ниже.

По умолчанию демон слушает 8082 порт и запускается командой:

Пример пайплайна

Человек лучше всего запоминает информацию на практических примерах, поэтому я придумал скрипт в задачу которого входит:

Вот как выглядит решение этой задачи в Luigi:

Скачать скрипт можно по ссылке. Для корректной работы необходимо помимо luigi также установить requests, pandas и beautifulsoup4:

Запускайте в терминале демон luigid, а сам скрипт вот таким образом:

Отправной точкой будет класс AggregateMovieRatingTask которому передается список интересующих нас лет. В методе requires() определяется зависимость от GetMovieMetaDataTask, поэтому до тех пор пока не будет получен результат от GetMovieMetaDataTask, код в методе run() у класса AggregateMovieRatingTask не будет исполнен.

При удачном раскладе AggregateMovieRatingTask.input вернёт список, содержащий объекты LocalTarget, полученные от выполнения GetMovieMetaDataTaskпо каждому году. Дальше необходимо пробежаться по списку, сформировать DataFrame и отсортировать его по убыванию.

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

Ограничения Luigi

Как и у любого другого инструмента, у Luigi есть свои ограничения с которыми приходится мириться в зависимости от ситуаций.

Отсутствие механизма запуска задач по расписанию. Если такая потребность имеется, то можно использовать crontab.

Luigi не предназначен для real-time обработки, его стихия это batch processing.

Сложность масштабирования. Luigi не умеет распределять задачи между воркерами на разных узлах/нодах как это умеет делать Celery, используя единый брокер сообщений (например, Redis или RabbitMQ). Без серьёзного ручного вмешательства тут не обойтись.

Заключение

Моей главной задачей в статье было рассказать про основные возможности Luigi. Вероятно те из вас, кто до сих пор мучается с boilerplate кодом при написании ETL скриптов взглянут на свою работу иначе, и полученная информация сделает вашу работу эффективной и приятной. За более подробным описанием стоит сходить на сайт с документацией. Отмечу, что Luigi не единственный инструмент в своём роде, обратите внимание на продукт под названием Airflow, разработанный в стенах Airbnb и с недавних пор перешедший в «руки» Apache Foundation (на момент написания статьи проект находится в статусе «incubating»).

Источник