ETL в анализе данных без перерывов на кофе и курилку / Хабр

Использование «статичных» скриптов из исходных систем

Очень часто бизнес-логика для сущностей DWH приходит от разработчиков или аналитиков исходных систем в виде SQL скриптов. И это большое подспорье для ETL-разработчика, но, как говорится, «бойтесь данайцев, дары приносящих»: как правило эти скрипты фиксируют некое условно «статичное» состояние исходное системы в некоторый момент времени, а ETL-разработчик обычно занимается отслеживанием динамики в данных и загрузкой только изменений («дельта»). Что же должно настораживать в этих «статичных» SQL скриптах? Вот некоторые из:

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

insert order_id into orders_from_callsselect order_id from orderswhere order_id IN (select order_id from calls where order_id <> -1)and changed_date > $last_loaded_date

Вроде бы все логично: грузить в нашу таблицу order_from_calls все заказы, на которые есть ссылка в таблице звонков, и для которых дата последнего изменения больше даты последней загрузки. А теперь представим, что обновление таблицы calls в DWH не произошло (например, она грузится из другой исходной системы и связь с ней по какой-то причине нарушена), и этот запрос не загрузил некоторые id заказов.

После этого таблица calls была дозагружена правильно, и там эти пропущенные id заказов появились, но мы их уже не загрузим в таблицу order_from_calls, т.к. в таблице orders ничего не поменялось и новые запуски этого запроса ничего не дадут.

«Механическая» реализация

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

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

Опишу на примере сущности «клиент» как одной из типичных для такого рода проблем: в источнике данных есть таблица «customer», имеющая уникальный натуральный ключ, ссылочная целостность в порядке. На основе этой таблицы в DWH была создана сущность «customer».

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

И это привело бы к неприятной коллизии для пользователей DWH, которые использовали эту сущность, например, для подсчета общего количества клиентов компании. В результате было принято решение разделить эту сущность на две: «customer_record» и «customer», связанные через FK отношением M:1.

А если бы ETL разработчик «механически» реализовал все по спецификации, то он бы конечно был бы не виноват, но у него была возможность заметить это, т.к. в любом случае он по сравнению с архитектором работает условно говоря «на земле», в отличие от «витающего в облаках» архитектора.

В целом можно упомянуть некоторые симптомы «механической» реализации:

Что же стоит делать, чтобы минимизировать риски «механической» реализации:

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

Etl listed mark

ETL was born into a culture of innovation. It was in Thomas Edison’s lighting laboratories where it all began, and to this day we still breathe the same air of innovation, safety and quality. We understand a manufacturer’s need to get new products to market quickly to achieve the greatest success, therefore we have built speed, responsiveness and urgency into our processes. Our commitment to helping customers gain the certifications they need quickly and efficiently has never been greater.

The ETL Mark is proof of product compliance to North American safety standards. Authorities Having Jurisdiction(AHJs) and code officials across the US and Canada accept the ETL Listed Mark as proof of product compliance to published industry standards. Retail buyers accept it on products they’re sourcing. And every day, more and more consumers recognize it on products they purchase as a symbol of safety.

Today, the ETL Mark is the fastest growing safety certification in North America and is featured on millions of products sold by major retailers and distributors every day. Browse through our most Frequently Asked Questions about the ETL Listed Mark.

Check out our white paper 2021 Guide To North American Product Testing & Certification for more information on the standards we test to for the North American market.

Global Network of Laboratories

Intertek is one of the world’s largest Testing, Inspection and Certification companies. We have a network of laboratories across North and South America, Europe and Asia to deliver safety testing and certification for your products. Our teams around the world provide Global Expertise with Local Service. No one partners with manufacturers better than Intertek to deliver the services you need, when you need them, and where you need them.

Про сертификаты:  Приказ Государственной корпорации по атомной энергии от 1 марта 2017 г. № 1/6-НПА "Об утверждении формы и порядка выдачи и учета бланков сертификатов соответствия продукции, для которой устанавливаются требования, связанные с обеспечением безопасности в области использования атомной энергии”

Here is a List of Standards we test to for the North American Market

• ASME
• ASTM
• ANSI
• CSA
• NFPA
• NOM
• NSF
• UL / ULC

Nationally Recognized Testing Laboratory:
As an OSHA Recognized NRTL in the U.S., Intertek is able to test and certify products for its customers across its global network of laboratories. Read more…

Intertek’s ETL Certification Now Accepted on Products Imported Into Mexico

Resolution from Mexican Government recognizes ETL as Equivalent to NOM Certification for three NOM Standards.

Our Accreditations

Intertek is an OSHA (Occupational Safety & Health Administration) recognized NRTL (Nationally Recognized Testing Laboratory) and is accredited as a Testing Organization and Certification Body by the Standards Council of Canada.

Inspector Center

AHJs can learn more about our accreditations, certification marks, and public safety notices by visiting our dedicated Inspector Information Center. Or check out our “Partnership in Safety” website to learn more about how we work with inspectors and code officials to help keep people safe where they live, work and play. 

Get the ETL Listed Mark on Your Product Today!

Временная метка. немного трюков

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

Воспользуемся базовыми функциями R, на сгенерированном датасете получаем примерно 130 секунд. Большинство аналитиков (неважно, на каком языке пишут) скажут, что это вполне нормальная ситуация, нет смысла хотеть бОльшего.

tic("straightforward as.POSIXct")
dt[, timestamp := as.POSIXct(paste(date, time), 
                                  format = "%Y%m%d %H%M%S")]
toc()

Но не будем на этом останавливаться. Вдруг данных будет немного больше (это же IoT по постановке задачи). 50М или 500М? Поменяем библиотеку с пониманием того, как она устроена внутри и как устроен POSIXct.

Получаем 3.5 секунды на нашем датасете.

tic("straightforward lubridate")
dt[, timestamp := lubridate::ymd_hms(paste(date, time))]
toc()

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

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

> uniqueN(dt, by = c("date", "time")) # почти в 100 раз меньше!
[1] 53424
> uniqueN(dt, by = "date")
[1] 1113
> uniqueN(dt, by = "time")
[1] 48

Используем этот подход меняем библиотеку.Вариант через групповую обработку — получаем 2.9 сек.

tic("anytime")
# не забываем следить про таймзону
dt[, timestamp := anytime::anytime(stri_c(.BY[[1]], .BY[[2]], sep = " ")), 
   by = .(date, time)]
toc()

Для потоковых преобразований ограниченных подмножеств наилучшим подходом является предварительное создание словаря “входное значение — преобразованное значение”. Тогда вся трансформация будет сводится выборке из этого словаря. Вариант через слияние по ссылкам со словарем дает чуть лучший результат 1.3 сек:

tic("inverse dictionary")
time_dict <- unique(dt[, .(date, time)]) %>%
  .[, timestamp := anytime::anytime(stri_c(date, time, sep = " "))]

dt[time_dict, on = .(date, time), timestamp := i.timestamp]
toc()

Это все? Потенциально, если расщепить словари даты и времени, то можно сократить объем преобразований еще примерно в 10 раз (1113 48 против 53K). Таймзоны в исходных данных не наблюдаются, городов тоже, значит все условно можно считать в UTC.

В таком варианте получаем примерно 0.7 сек.

tic("split   anytime")
dt[, secs := {
  tm <- as.integer(.BY[[1]]);
  (tm %% 100L)   ((tm %/% 100L) %% 100L) * 60L   (tm %/% 10000L) * 3600L}, 
  by = time] %>%
  .[, timestamp := anytime::anytime(.BY[[1]])   secs, by = date]
toc()

Итого, путем краткого исследования исходной задачи и незначительных манипуляций получаем ускорение со 130 сек (которые многие аналитики сочли приемлемым) до 0.7 сек. (~ 180 раз) в базовом варианте, или с 3.5 сек до 0.7 сек (~ 5 раз) в варианте с lubridate. Можно, конечно, разбрасываться ресурсами, но если это делать на каждом шагу, то ресурсов и времени может легко не хватить.

Данные из xlsx-файла

Рассмотрим работу с petl на наборе результатов летних олимпиад по странам. Нам понадобится файл datasets/summer_olympics.xlsx, посмотрим на первые строки, пока не сохраняя таблицу в переменную.

etl.fromxlsx('datasets/summer_olympics.xlsx')

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

olympics = etl.fromxlsx('datasets/summer_olympics.xlsx').skip(1)

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

olympics2 = olympics.setheader(['country','games','gold','silver','bronze'])

Мы начали выстраивать цепочку преобразования таблицы, это удобно, так как можно просмотреть результат работы на каждом этапе. Иногда, наоборот, удобнее объединить цепочку сразу в одной команде. Например, совместим изменение заголовка с сортировкой по количеству золотых медалей.

olympics2 = olympics.setheader(['country','games','gold','silver','bronze']).sort('gold', reverse=True)

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

Мы также применим мощный инструмент Python – Анонимные функции. Анонимная функция (функция без имени) – это запись вида lambda x: <функция от x>. Читается как: “То, что было подано на вход этого выражения, будет положено в x, а результатом исполнения будет <функция от x>.

В PETL это часто применяется, чтобы выполнить быстрое преобразования значения какого-либо из полей. Например, если нужно все значения таблицы table в поле field умножить на два, это можно написать как table.convert(‘field’, lambda x : x * 2). В примере ниже функция применяется не к отдельным значениям, а к строке целиком.

olympics2.addfield('total', lambda row : row['gold']   row['silver']   row['bronze'])

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

olympics3 = olympics2
    .addfield('total', lambda x: int(x['gold'])   int(x['silver'])   int(x['bronze']))
    .sort('total', reverse=True)

Видим, что в таблице есть сумма по всем странам, что нас не интересует в данной задаче. Можем выбрать из таблицы все строки, кроме строки со значением country == Totals. Воспользуемся функцией select.

Кроме того, дополнительно рассчитаем новый показатель – результативность страны, определив её как среднее число медалей за игру.

olympics4 = olympics3
    .select(lambda x: x.country != 'Totals')
    .addfield('effectiveness', lambda x: round(x['total'] / float(x['games']), 2))

Сохраним полученные результаты в новый xlsx-файл.

olympics4.toxlsx('olympics.xlsx')

Данные из бд (postgresql)

В состав ViXtract входит предустановленная СУБД PostgreSQL, её удобно использовать как промежуточное хранилище данных, из которого их уже забирает BI-система. Похожие подходы могут быть использованы и с любой другой СУБД.

Рассмотрим следующий пример.Доступны данные о состояниях различных типов транспортных средств. В базе есть 2 таблицы:

Необходимо подготовить таблицу, содержащую валидные данные по бульдозерам:

statuses = etl.fromdb(connection, 'SELECT * FROM status_ts')
ts_types = etl.fromdb(connection, 'SELECT * FROM ts_types')

# Вспомогательные функции
# Определяем фильтр для исключения строк с пустыми значениями
row_without_nones = lambda x: all(x[field] != '' for field in statuses.fieldnames())
# Перевод отметки времени в формат datetime
to_datetime = lambda x: dt.fromtimestamp(int(x))

Чтобы исключить строки с пропусками, используем функцию select и определенный выше фильтр row_without_nones

statuses.select(row_without_nones)

Переведём столбцы со временем в требуемый формат. Для этого необходимо воспользоваться функцией convert. Сразу можем добавить расчёт продолжительности функцией addfield.

statuses.
    convert('Начало', to_datetime).
    convert('Окончание', to_datetime).
    addfield('Продолжительность', lambda x: x['Окончание'] - x['Начало'])

Объединим обе таблицы и выберем данные только по бульдозерам, сразу уберём строки с состоянием “Отсутствие данных”.

statuses.
    join(ts_types, lkey='id ТС', rkey='id').
    select(lambda x: 'Бульдозер' in x['Тип ТС'] and x['Состояние'] != 'Отсутствие данных')

Все перечисленные операции можно произвести за раз, сформируем цепочку функций. Заметим, что столбец id ТС уже не требуется, его можно убрать функцией cutout.

В дополнение ко всему отсортируем таблицу по времени начала состояний, применив sort.

result = statuses.
    join(ts_types, lkey='id ТС', rkey='id').
    select(lambda x: 'Бульдозер' in x['Тип ТС'] and x['Состояние'] != 'Отсутствие данных').
    select(row_without_nones).
    convert('Начало', to_datetime).
    convert('Окончание', to_datetime).
    addfield('Продолжительность', lambda x: x['Окончание'] - x['Начало']).
    convert('Начало', str).convert('Окончание', str).convert('Продолжительность', str).
    cutout('id ТС').
    sort('Начало')

Решение = jupyterhub petl cronicle

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

1. JupyterHub —  интерактивная среда выполнения Python-кода. По сути, это среда разработки, которая позволяет работать с кодом в интерактивном режиме. Она очень удобна для тех, кто не является профессиональным разработчиком, не накопил готовых навыков программирования на уровне спинномозговых рефлексов. JupyterHub помогает, когда ты разбираешься с кодом, пробуешь что-то новое, экспериментируешь. 

2. Библиотека PETL была разработана на Python специально для обработки данных. Она берёт на себя огромное количество рутинных задач, например, разбор CSV файлов различных форматов или создание схемы в БД при выгрузке данных.

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

Чтобы всем этим было проще пользоваться, мы объединили три инструмента в ViXtract. Речь идет о сборке набора open-source технологий, которая позволяет легко установить решение одной командой и использовать ETL, не заморачиваясь по поводу Linux, по поводу прав, нюансов интеграций и других тонкостей.

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

Сравнение etl и elt