Привет! Меня зовут Артемий Козырь, и я Analytics Engineer в Wheely.

Мы могли бы долго и нудно обсуждать, кто такой Analytics (Data / Backend) Engineer, какими инструментами он должен владеть, какие buzzwords в тренде и ценятся в CV, однако, на мой взгляд, гораздо интереснее рассмотреть процесс и результаты его деятельности в рамках конкретной прикладной задачи.

В этой публикации:

• Что значит решение End-to-End и в чем его ценность?

• Организация Extract & Load данных из асинхронного API MaestroQA

• Моделирование витрин данных с помощью dbt

• Поставка ценности для пользователей с помощью Looker

Решение End-to-End - от идеи до создания ценности

В двух словах, End-to-End – это поставка полноценного функционального решения, включающая все детали пазла.

Предлагаю переходить к реальному сценарию - это работа с приложением MaestroQA, автоматизирующим мониторинг и оценку клиентского сервиса (Customer Support).

Одна из самых важных идей заключается в том, что заказчик, кем бы он ни был (Manager, Product Owner, CEO), почти никогда не ставит задачу в инженерных терминах:

• Налить 100500 гигабайт в Хранилище

• Добавить multithreading в код

• Написать супероптимальный запрос

• Создать 15 dbt-моделей

За любой инженерной задачей стоит решение конкретных бизнес-проблем. Для нас это:

• Прозрачность Customer Support (фиксируем все оценки, инциденты)

• Результативность на ладони (отслеживаем динамику показателей во времени)

• Отчитываемся о KPI команд поддержки (агрегирующие показатели по командам, городам, странам и т.д.)

• Получаем обратную связь и исправляем ошибки (идентификация слабых/проблемных мест и быстрый feedback)

• Постоянно учимся и разбираем кейсы (категоризация тем, организация тренингов и разборов)

И это ключевой фокус, который отличает Analytics Engineer от, например, классических Data Engineer, Backend Engineer. Обладая всем спектром инженерных навыков и практик, Analytics Engineer создает ценность для бизнеса и решает прикладные задачи. Говорит на одном языке с заказчиком решений и мыслит в терминах бизнес-показателей.

Получим исходные данные – Extract & Load Data

Окей, теперь ближе к делу. Я выбрал MaestroQA намеренно - это источник, для которого нет готовых коннекторов в SaaS-решениях.

Поэтому нам придется реализовать эту интеграцию и установить ее на расписание самостоятельно.

Выбор инструментов для интеграции данных - во многом вопрос вкуса, но я предпочитаю использовать простые shell-скрипты и оркестрировать их с помощью Airflow.

1. Начнем с изучения документации к API сервиса:

• Нам доступен ряд методов: request-raw-export, request-groups-export, request-audit-log-export, get-export-data

• Методы принимают набор параметров: apiToken, startDate, endDate, exportId

• Результирующие отчеты формируются асинхронно

Асинхронный API означает то, что в ответ на запрос той или иной выгрузки вы получите не саму выгрузку а номер в очереди. Предъявив этот номер в другое окно (метод get-export-data), вы получите выгрузку, как только она будет готова.

Использование Async API несколько усложняет задачу, а именно использованием нескольких методов и сохранением exportId, но так задача становится даже интереснее.

2. Получим API Token.

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

3. Готовим скрипты для выгрузки.

Шаг 1. Запросить выгрузку сырых данных (total_scores.sh):

• Сформировать запрос к API (JSON_DATA)

• Получить exportId (талон в очереди)

• Проверить значение exportId и если ОК, то перейти к выгрузке результата (retrieve.sh)

# 1. Prepare request
JSON_DATA=$(jq -n \
              --arg maestroqa_token "$MAESTROQA_TOKEN" \
              --arg start_date "$START_DATE" \
              --arg end_date "$END_DATE" \
              --arg single_file_export "$SINGLE_FILE_EXPORT" \
              --arg name "$FILE_NAME" \
              '{apiToken: $maestroqa_token, startDate: $start_date, endDate: $end_date, singleFileExport: $single_file_export, name: $name }' )

# 2. Get exportId
EXPORT_ID=$(curl -s -X POST $ENDPOINT \
    -H 'Content-Type: application/json' \
    -d "${JSON_DATA}" \
    | jq -r '.exportId')

# 3. Retrieve data by exportId
if [ -z "$EXPORT_ID" ]
then
      echo "EXPORT_ID is empty"
      exit 1
else
      echo "EXPORT_ID=$EXPORT_ID"
      EXPORT_ID=$EXPORT_ID bash retrieve.sh
fi

Шаг 2. Получить готовую выгрузку (retrieve.sh):

• Определить функцию для запроса статуса готовности выгрузки (get_status())

• Сформировать запрос к API (JSON_DATA)

• Опрашивать API о статусе готовности каждые 10 секунд

• По готовности (complete) получить файл с результатами и сохранить в S3

# 1. function used to poll to get current status
get_status() {
    curl -s -X GET $RETRIEVE_ENDPOINT \
    -H 'Content-Type: application/json' \
    -d "${JSON_DATA}" \
    | jq -r '.status' \
    | cat
}

# 2. prepare request
JSON_DATA=$(jq -n \
              --arg maestroqa_token "$MAESTROQA_TOKEN" \
              --arg export_id "$EXPORT_ID" \
              '{apiToken: $maestroqa_token, exportId: $export_id }' )

# 3. get current status ("in progress" / "complete")
STATUS="$(get_status)"
printf "STATUS=$STATUS\n"

# 4. poll every 10 seconds
while [ "$STATUS" != "complete" ]; do
  printf "STATUS=$STATUS\n"
  sleep 10
  STATUS="$(get_status)"
done

# 5. Store resulting file to S3
curl -s -X GET $RETRIEVE_ENDPOINT \
  -H 'Content-Type: application/json' \
  -d "${JSON_DATA}" \
  | jq -r '.dataUrl' \
  | xargs curl -s \
  | aws s3 cp - s3://$BUCKET/$BUCKET_PATH/$FILE_NAME/$FILE_NAME-$START_DATE-to-$END_DATE.csv

echo "UPLOADED TO s3://$BUCKET/$BUCKET_PATH/$FILE_NAME/$FILE_NAME-$START_DATE-to-$END_DATE.csv"

4. Автоматизация на Airflow.

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

Пример DAG:

from airflow import DAG
from airflow.operators.bash_operator import BashOperator

import os
import yaml
from datetime import datetime, timedelta
from slack.notifications import failed_task_slack_notification

### INIT DAG
DAG_NAME = "maestroqa_api"
SCHEDULE_INTERVAL = '0 0 * * *'

DAG_PATH = os.path.dirname(__file__)
CONFIG_FILE_NAME = "config.yml"
CONFIG_PATH = os.path.join(DAG_PATH, CONFIG_FILE_NAME)
CONFIG = yaml.safe_load(open(CONFIG_PATH))["endpoints"]

default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "start_date": datetime(2022, 2, 1),
    "retries": 1,
    "retry_delay": timedelta(minutes=3),
    "catchup": False,
    "dagrun_timeout": timedelta(minutes=5),
    'on_failure_callback': failed_task_slack_notification
}

dag = DAG(
    DAG_NAME,
    default_args=default_args, 
    schedule_interval=SCHEDULE_INTERVAL
)

os.environ["START_DATE"] = "{{ execution_date.isoformat() }}"
os.environ["END_DATE"] = "{{ next_execution_date.isoformat() }}"

groups = BashOperator(
            task_id=CONFIG['groups']['FILE_NAME'],
            bash_command=f"cd {DAG_PATH}/ && bash {CONFIG['groups']['FILE_NAME']}.sh ",
            env={ **os.environ.copy(), **CONFIG['groups'] },
            trigger_rule="all_done",
            dag=dag
    )

total_scores = BashOperator(
            task_id=CONFIG['total_scores']['FILE_NAME'],
            bash_command=f"cd {DAG_PATH}/ && bash {CONFIG['total_scores']['FILE_NAME']}.sh ",
            env={ **os.environ.copy(), **CONFIG['total_scores'] },
            trigger_rule="all_done",
            dag=dag
    )

groups >> total_scores

Смоделируем витрины данных – Transform Data

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

Зарегистрируем файлы в S3 в качестве EXTERNAL TABLE

Чтобы иметь возможность обращаться к данным с помощью SELECT-запросов. В этом нам поможет package dbt-labs/dbt_external_tables:

version: 2

sources:

  - name: maestroqa
    database: wheely
    schema: spectrum
    tags: ["sources", "maestroqa"]
    loader: Airflow (S3 via External Tables)
    description: "MaestroQA – customer service quality assurance software"

    tables:

        - name: groups
          identifier: maestroqa_groups
          description: "Agent Groups"
          external:
            location: "s3://{{ var('s3_bucket_name') }}/maestroqa/GROUPS/"
            row_format: serde 'org.apache.hadoop.hive.serde2.OpenCSVSerde'
            table_properties: "('skip.header.line.count'='1')"
          columns:
            - name: group_name
              data_type: varchar
              description: "Group / Team name: All Agents, UK Team, Ru Team, FR Team, UK Team"
            - name: group_id
              data_type: varchar
            - name: agent_name
              data_type: varchar
            - name: agent_email
              data_type: varchar
            - name: agent_ids
              data_type: varchar
              description: "List of semicolon separated Agent IDs. Used to link with MaestroQA Total Scores table"
            - name: available
              data_type: bool
              description: "Flag indicating if agent is available at the moment"

Результирующие EXTERNAL TABLES будут использоваться в dbt в качестве sources (источников данных)

Моделирование витрины данных

Предлагаю взглянуть на то, как выглядят исходные данные, полученные из ответов API. Выгружаются 2 типа файлов:

• Groups - справочник агентов, команд

• Scores - факты оценок и скорингов коммуникаций

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

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

{{
    config (
      materialized='table',
      dist='auto',
      sort=['graded_dt', 'country'],
      tags=['maestroqa']
    )
}}

SELECT

    -- IDs
	  scores.gradable_id
	, scores.agent_id 

    -- dimensions
    , scores.grader
	, scores.agent_name
	, scores.agent_email
	, groups.group_name
	, CASE groups.group_name
      WHEN 'Ru Team' THEN 'RU'
      WHEN 'FR Team' THEN 'FR'
      WHEN 'UK Team' THEN 'GB'
      WHEN 'All Agents' THEN 'All'
    END AS country
        
    -- dates
    , scores.date_graded::DATE AS graded_dt

    -- measures
    , scores.rubric_score
	, scores.max_rubric_score

FROM {{ ref('stg_maestroqa_total_scores') }} AS scores
    LEFT JOIN {{ ref('flt_maestroqa_groups') }} AS groups ON groups.agent_id = scores.agent_id 
        AND groups.group_name IN ('All Agents', 'Ru Team', 'FR Team', 'UK Team')

Итоговый граф зависимостей моделей dbt (DAG) выглядит следующим образом:

Обеспечим доступ к данным через BI – Deliver value

Отлично, к этому этапу помимо набора файлов в S3 у нас есть постоянно обновляющаяся широкая таблица в СУБД (витрина), обращаясь к которой мы можем получить ответы на любые вопросы.

Однако не все пользователи одинаково способны формулировать свои вопросы на чистом SQL. Этот барьер призваны устранить BI-инструменты, основные задачи которых сводятся к:

• Предоставлению визуального конструктора запросов к данным

• Формированию набора измерений, метрик и фильтров для последующего использования бизнес-пользователями

• Группировке ряда визуализаций и ответов в пользовательские дашборды

• Настройке рассылок данных и уведомлений по определенным правилам

В Looker каждой колонке исходной таблицы можно присвоить статус измерения или метрики, задать правила агрегации, добавить комментарий. Это делается на языке LookML, напоминающем Javascript:

Далее любой пользователь может воспользоваться конструктором для визуального формирования ответов на свои вопросы – в Looker это называется Explore:

Готовые плиточки (tiles) можно группировать в дашборды, рассылать всем заинтересованным пользователям, устанавливать уведомления при достижении пороговых значений метрик:

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

Умение строить комплексные решения, отвечающие на запросы бизнеса

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

Именно эти аспекты я держал в уме, когда работал над программами курсов Analytics Engineer и Data Engineer в OTUS.

Это не просто набор занятий по темам, а единая, связная история, в которой акцент делается на понимание потребностей заказчиков. На live-сессиях я и мои коллеги делимся своим опытом и реальными кейсами:

• Продвинутое моделирование в dbt

• Развертывание и особенности работы c BI-инструментами

• Аналитические паттерны и SQL

• Кейсы: Сквозная аналитика, Company’s KPI, Timeseries analysis

Также своими наблюдениями, опытом и практиками я делюсь в ТГ-канале Technology Enthusiast.

Напишите комментарий, если сталкивались с потребностью строить подобные решения, и какой подход применяли?

Спасибо!