deploy ml_course :rocket:

Содержание :pagewithcurl:

• О курсе
• Jupyter в ProductCode
• Версионирование процесса
• API для модели
• СI/CD
• Feature Store
• Docker
• AirFlow
• Мониторинг

---

О курсе :key:

>

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

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

В ходе прохождения курса вы научитесь выстраивать процесс разработки модели для продуктивного решения бизнес-задач, пройдете путь трансформации от jupyter notebook кода до контейнеризации модели и автоматизации необходимых для её работы процессов. В курсе также будет уделено внимание метрикам и способам мониторинга моделей.

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

Курс разработан при поддержке НИУ ВШЭ

Ссылка на курс на сайте НИУ ВШЭ

Знания для прохождения курса

• Язык программирования Python (базовые библиотеки для Python, которые используются для работы с данными: numpy, pandas, sklearn)

• Знание принципов построения моделей машинного обучения

• Командная строка Linux

Видео - плайлист курса на YouTube

YouTube

---

Часть 1 (Jupyter в ProductCode) :beginner:

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

Отвечая на вопрос, какие умения разработчика важны для DS, можно вспомнить статью software dev skills for ds от Trey Causey и сказать:

• уметь тестировать
• уметь версионировать
• уметь логгировать -> отличный пример
• уметь писать модули

Архитектуры ML приложений

Курс базируется на архитекутуре ML - train by batch, predict on fly

Виды ML pipeline (или модели вне jupyter notenook)

Будут рассмотрены основные виды ML Pipeline:

• Процедурные (на основе набора функций) с запуском через main.py

• Scikit-learn pipeline, построение и использование функций и классов в библиотеке Scikit-learn (sklearn.pipeline.Pipeline)

• Custom (или объектно-ориентированный pipeline)

Тестирование

Разбор основных видов тесто для ML проекта. В этой части рассматриваем 6 основных

вилов тестов, которые должны быть у модели и её процесса. >

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

>

Перед знакомством с тестами, научимся делать дебаг и поиск ошибок в коде на примере игры о pdb

>

Познакомимся с тестами через учебный проект по тестированию кода для теннисного турнира ссылка

>

Позапускаем тесты

>

Run

pytest /pathtotest_script.py

>

Run + html отчет

pytest /pathtotest_script.py --html=report.html --self-contained-html

>

Run + code coverage report (результат по каждой функции)

pytest /pathtotest_script.py --cov=src --verbose

>

Run + multi CPU

pytest /pathtotest_script.py -n 5

---

Объекты для которых нужны тесты

• Объекты извлекающие данные
• Объекты трансформирующие данные
• Модель (если она самописная)
• Обекты сохранения данных

---

Идея тестов

• Тестировать всегда один объект из процесса (кода). Т.е. один юнит за один тест
• Не использовать зависимости в тестируемом коде из других процессов (юнитов)
• Минимизировать запросы к API / Базе данных
• Тестировать свойства, не значения (для DS|DE)
• Тестировать типы и размерности (для DS|DE)
• Генерируйте фичи для теста (feature forge), а не создавайте руками (для DS|DE)
• Используйте базовые методы из Numpy и Pandas для тестирования (для DS|DE)

# тесты должны быть осмысленные
def mean(some_list):   # на примере:   # тест на пустой список и деление на 0   # TypeError тест   # OverFlow тест   return sum(somelist) / len(somelist)
dev тест
import pytest
def testdevmean():
  assert(sum(mean([1,2,3])) / len(3) == 2)
Но для DS
from hypothesis import given
import hypothesis.strategies as st
@given(st.lists(st.integers())) def testdsmean(some_list):   assert mean(somelist) == sum(somelist) / len(some_list)

# тесты с проверкой типов / размерности
@is_shape(5, 10)
@is_monotonic(strict = True)
@non_missing()
def my_funct(df, columns):
  pass

---

К основным видам тестов относятся:

1 Smoke test - тесты для опредления метода

import pytest from sklearn.linear_model import LinearRegression def test_smoke():     try:         assert LinearRegression() is not None     except NameError:         pytest.fail("Model does not exist")

2 Проверка коннекторов к БД (если нужны для проекта)

Ссылка на Хабр, с большим объяснением

import mock
def get_connect():     engine = sqlalchemy.createengine('connectionstring')
return df.read_sql("select  from ", con = engine)
@mock.patch('pytestex.textconnection.sqlalchemy.create_engine') def testconnection(enginemock, df):     newdf = getconnect(engine_mock)
pandas.testing.assertframeequal(new_df, df)
или тестировать функции чтения
def dffromcsv(filename):
    return pd.read_csv(filename)
@mock.patch.object(pd, 'read_csv') def testdffromcsv(readcsv_mock, df):     readcsvmock.return_value = df     actual = dffromcsv('file_name.csv')     # ожидаемый результат     expected = df     # assertions     pd.testing.assertframeequal(actual, expected)

3 Тесты на равенство / соответствие результатов после трансформации объекта

• Использовать pandas testing:

-- pandas.testing.assertframeequal (сравнение DataFrame)

-- pandas.testing.assertseriesequal (сравнение колонок)

-- pandas.testing.assertindexequal (сравнение строк по индексам)

-- pandas.testing.assertextensionarray_equal (сравнение любых массивов numpy)

-- Использование assert + numpy (методы сравнения объектов):

-- np.allclose

-- np.isclose

-- np.any

-- np.equal

-- Использование numpy testing методов:

-- (Ссылка на документацию numpy asserts)[https://numpy.org/doc/stable/reference/routines.testing.html]

4 Тестирование существования файлов

def dffromcsv(filename):     """читаем все DataFrame в формате csv"""     return pd.read_csv(filename)

def testdffrom_csv(filename):     assert dffromcsv(filename) is not FileNotFoundError

5 Тестирование API

import responses

@responses.activate def testapi404():     responses.add(         responses.GET,         'https://your_path',         json='ex_json',         status=404,     )
@responses.activate def testapi200():     responses.add(         responses.GET,         'https://your_path',         json={'ex_json'},         status=200,     )

6 Генерируйте данные для тестов на основе hypothesis

import pandas as pd
from hypothesis import given
from hypothesis import strategies
from hypothesis.extra.pandas import column, data_frames
import builder
@given(     # создавайте Pandas DataFrame для тестов     data_frames(     columns=[         column(             name='prog_start',             elements=strategies.datetimes(                 min_value=pd.Timestamp(2020, 1, 1),                 max_value=pd.Timestamp(2020, 1, 10)             )         , unique=True),         column(             name='code',             elements=strategies.just(float('nan'))         )     ]) ) def testfixnewboxesnanreplaced(rawprog):     prog = builder.fixnewboxes(raw_prog)     assert (prog.matcode == builder.NOMAT_CODE).all()     assert prog.shape == raw_prog.shape

Задание для самостоятельной работы

• Воспользуйтесь шаблонами для sklearn.pipeline и процедурного pipeline, переделайте свою модель из эксперементального jupyter notebook в 3 вида pipeline

Задание

---

Версионирование процесса

Разбор основных способов версионирования в GIT с проекцией на деятельность DS/DE.

GIT

Предложена мультиветвенная система версиониования для монолитного проекта. ! Определитесь с неймингом процессов!

О Git flow для DE/DS и основными шаблонами можно ознакомиться в другом репозитории

По ссылке вы найдете:

• шаблон для проекта

• шаблон для adhoc

• шаблон для Spark проекта

• набор готового кода для git

Ещё немного информации о git - больше волшебства

DVC

Рассмотрим инструмент dvc, отличный для экспериментов и фиксации процессов

dvc init
отключаем аналитику наших процессов (чтобы не собиралась статистика)
dvc config core.analytics false
устанавливаем наше хранилище для файлов
dvc remote add -d localremote /data/dvc-storage

---

Создаем params.yaml по шаблону:

# file params.yaml название модели (эксперимента):     параметры для эксперимента

---

Создаем шаги наших экспериментов (для трекинга):

dvc run -n STAGE_NAME \ -d все файлы от которых зависит процесс -O все файлы, которые будут являться результатами процесса ( но не будут версионироваться) -o все файлы, которые будут являться результатами процесса (будут версионироваться) -M файл с метрикой

---

Основные команды для процессинга

# воспроизведение процесса (повторение от шага 1 до финального)
dvc repro
сравнение параметров / метркиа
dvc params diff
dvc metrics diff
визуализация процесса
dvc dag

---

KEDRO

Отличный проект kedro, на основе которого будут выстроеные процессы на данном курсе

Проект является набором правил для разработки в МЛ. Но во время работы следует учитывать: * Не удалять (только дополнять) файл .gitignore * Работать в рамках конвенции DE разработки * Переменные и конфиги эфемерные * Конфиги в conf/local/

# создать структуру проекта
kedro new
в src/projectname/
нужно создать pipeline (для каждого типа процессов - свой пайплайн)
запуск jupyter notebook в kedro
kedro jupyter notebook

---

В проекте должны быть следующие pipelines:

• data engineering(etl + обработка данных)

• data science(обучение модели)

• predict_pipeline(предикт по модли и проверка качестве)

• predictapi_pipeline(предикт для работы через API)

# добавляем созданные pipeline
в hook для запуска
depipe = de.createpipeline()
dspipe = ds.createpipeline()
prpipe = pr.createpipeline()
prapipe = pra.createpipeline()
return {     "de": de_pipe,     "ds": ds_pipe,     "predict": pr_pipe,     "predictapi": prapipe,     "default": depipe + dspipe + pr_pipe, }

Это позволяем запускать код по имени pipeline

# в bash
kedro run --pipeline='predict_api'
в функции python
context.run(pipelinename='predictapi')

---

Pipeline модели в Kedro (у данного проекта)

Задание для самостоятельной работы

Задание

---

API для модели

Зачем DS/DE знать про API?

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

В данной работе мы сделаем выбор 1 из 3 основных реализаций API.

Мы будем использовать паттерны, которые поставляются с Kedro.

# подключаем контекст Kedro
from kedro.context import load_context
from kedro.extras.datasets.api import APIDataSet
import json
загружаем проект
context = load_context("../")
catalog = context.catalog
используя APIDataSet из Kedro
и устанавливаем конфиг
st = APIDataSet(
        url = "http://127.0.0.1:9876/data",
        headers = {"content-type": "application/json"}
).load()
записываем в DF для работы из json
df = pd.DataFrame.from_dict(st.json()['data'], orient='index').T
формируем результат для отправки назад по API
answer = {
        "predict_date": st.headers['date']
        "row_index": st.json()['index']
        "predict": model.predict(df)
}

---

Задание для самостоятельной работы

Задание

---

CI/CD

Разработка через тестирование для DS - реальность!

Рассмотрим, какие тесты нужно делать для DS/DE пайплайнов и как их делать. И немного погрузимся в методологию TDD.

Опять переделаем пайплайн, так как рассмотрев пример проблемы и поймем, что делали работу не по TDD методологии.

Тесты с hypothesis

Рассмотрим пример разработки теста для функции:

# функция для тестирования def index_creator(df, mind, maxd, freq):     by = 'Station'     newCol = 'Datetime'     return pd.MultiIndex \              .fromproduct([df[by].sortvalues().unique(),                             pd.Index(pd.date_range(mind,                                                    maxd,                                                    freq=freq))],                             names=[by, newCol])

Скрипт по разработке теста

import pandas as pd
import numpy as np
from hypothesis import given from hypothesis import strategies as st from hypothesis.extra.pandas import data_frames, column from scipy.special import expit
import matplotlib.pyplot as plt import seaborn as sns
создадим DF для работы (генерация случайного ДатаФРейма)
генерация данных для функции
получаем элементы из given
создаем случайный
import string
df = data_frames(         [             column('Station', dtype=str,                    elements=st.text(alphabet=f"{string.asciiletters}{string.asciilowercase}", min_size=5)),             column('DateOfPeriod',                    elements=st.datetimes(min_value=datetime.datetime.strptime('2019-01-01 00:00:00', '%Y-%m-%d %H:%M:%S'),                                          max_value=datetime.datetime.strptime('2021-01-01 00:00:00', '%Y-%m-%d %H:%M:%S'))),         ]     ).example()
создаем переменные для даты и время
mn = st.datetimes(min_value=datetime.datetime.strptime('2019-01-01 00:00:00', '%Y-%m-%d %H:%M:%S'),
                 max_value=datetime.datetime.strptime('2021-01-01 00:00:00', '%Y-%m-%d %H:%M:%S')).example()
mx = st.datetimes(min_value=datetime.datetime.strptime('2019-01-01 00:00:00', '%Y-%m-%d %H:%M:%S'),
                 max_value=datetime.datetime.strptime('2021-01-01 00:00:00', '%Y-%m-%d %H:%M:%S')).example()

что нам нужно получить для функции
переменные
mind = mn
maxd = mx
freq = 'H'
константы
by = 'Station'
newCol = 'Datetime'
pd.MultiIndex \   .fromproduct([df[by].sortvalues().unique(),                  pd.Index(pd.date_range(mind,                                         maxd,                                         freq=freq))],                  names=[by,                         newCol])
тесты
соответсвие класса (smoke)
index_creator(df, mn, mx, 'D').class == pd.core.indexes.multi.MultiIndex
правильный способы проверки класса
isinstance(index_creator(df, mn, mx, 'D'), pd.MultiIndex)
что будет в нужной структуре и не пустой индекс с уровнями, именами и определителями
try:
    pd.testing.assertindexequal(index_creator(df, mn, mx, 'D'),
                                      pd.core.indexes.multi.MultiIndex(levels = [['', '0'], []],
                                                                       names = ['Station', 'Datetime'],
                                                                       codes=[[], []]))
except AssertionError:
    True

with pytest.raises(AssertionError):     pd.testing.assertindexequal(index_creator(df, mn, mx, 'D'),                                   pd.core.indexes.multi.MultiIndex(levels = [['', '0'], []],                                                                    names = ['Station', 'Datetime'],                                                                    codes=[[], []]))

GitHub Actions

Actions создаются в формате YAML и должны находится в папке .github/workflows. В Actions учавствуют следующие элементы:

• event триггер к действияю
• machine место, где запускается джоб
• The jobs процесс выполняющий задачи
• The steps задачи для джоба

#.github/workflows/first_workflow.yml name: First Workflow                                                on: push                                                   jobs:                            first-job:                                name: Hello                              runs-on: ubuntu-latest                                steps:                                - name: Print a greeting                                    run: echo Hellow!

Использование Action

Action - это одна задача, которая:

• Вызывается в том же репозитории, где и хранится код

• Который хранится отдельным репозиторием

• Который выполняется в контейнере

# .github/workflows/first_workflow.yml
name: First Workflow
on: push                                                  
jobs:                         
  first-job:                           
    name: Hellow                  
    runs-on: ubuntu-latest                           
    steps:                           
      - name: Print a greeting                             
        run: echo Hello!
- name: Show ASCII greeting                                      uses: mscoutermarsh/ascii-art-action@master            with:                                          text: 'HELLO!'

Интеграция Python в Action

Для интеграции Python в GitHub Actions необходимо в контейнер установить Python, можно использовать специальный Action: setup-python и проверить доступность файлов через Action checkout.

# .github/workflows/first_workflow.yml
name: First Workflow
on: push                                                  
jobs:                         
  get-posts-job:                            
    name: Run                      
    runs-on: ubuntu-latest     
    steps:                             
      - name: Check-out the repo under $GITHUB_WORKSPACE                               
        uses: actions/checkout@v2
- name: Set up Python 3.8                                        uses: actions/setup-python@v2                                        with:                                            python-version: '3.8'
- name: Install Scrapy                                        run: pip install scrapy
- name: Run Python command                                       run: scrapy runspider posts_spider.py -o posts.json

Получить данные из Git с помощью Actions

Ваши процессы могут возвращать результат, вы можете получать результат и скачивать его из Git с помощью Actions upload-artifact и download-artifact.

# .github/workflows/first_workflow.yml
name: First Workflow
on: push                                                  
jobs:                         
  get-posts-job:                            
    name: Run          
    runs-on: ubuntu-latest     
    steps:                             
      - name: Check-out the repo under $GITHUB_WORKSPACE                               
        uses: actions/checkout@v2
- name: Set up Python 3.8                                        uses: actions/setup-python@v2                                        with:                                            python-version: '3.8'
- name: Install Scrapy                                        run: pip install scrapy
- name: Run Python                                       run: scrapy runspider posts_spider.py -o posts.json
- name: Upload artifact                               uses: actions/upload-artifact@v2                                 with:                                            name: posts                                            path: posts.json

Создать Action самостоятельно

Action создается, как Docker контейнер (про Docker), файл с описание парсится при с помощью TypeScript.

name: 'Любое имя для GitHub Action'
description: 'Описание его действия, что ему нужно на вход и какой будет результат'
inputs:
  user:  
    description: 'Описание входящих данных'
    required: true
    default: 'Что будет по умолчанию'
runs:
  using: "composite"
  steps:
    - run: команда, которая запускает процесс
      shell: bash/ЯП

Примеры GitHub actions

# Пример WorkFlow, который
запускает установку Python и зависимостей, запускает tests
name: Tests 
для каких действий запускать данный WorkFlow
 on:   push:     branches: [ main ]   pull_request:     branches: [ main ] jobs:   build:     runs-on: ubuntu-latest     steps:     - uses: actions/checkout@v2     - name: Set up Python 3.8       uses: actions/setup-python@v2       with:         python-version: 3.8     - name: Install dependencies       run: |         python -m pip install --upgrade pip         pip install flake8 pytest         if [ -f requirements.txt ]; then pip install -r requirements.txt; fi     - name: Lint with flake8       run: |         # stop the build if there are Python syntax errors or undefined names         flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics         # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide         flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics     - name: Test with pytest       run: |         pytest

# Action для CodeReview
name: Code review on:   push:   pull_request: jobs:   pycodestyle:     name: pycodestyle     runs-on: ubuntu-latest     steps:     - uses: actions/checkout@master     - name: wemake-python-styleguide       uses: wemake-services/wemake-python-styleguide@0.14.1       with:         reporter: 'github-pr-review'   # для репорта ошибок в PR комментарии       env:          GITHUBTOKEN: ${{ secrets.GITHUBTOKEN }}

Задание для самостоятельной работы

Задание

---

Feature Store

Сделаем шаг в сторону от модели и рассмотрим специальный тип хранилища, но для DS.

Great Expectations

Great Expectations - это специальный инструмент для тестирования данных и фиксации их профилирования.

GE встраиватся в pipeline для тестирования входных данных.

Создан для DE/DS:

• Помощь в работе с данными и мониторинге

• Помощь в нормализации данных

• Улучшение взаимодействия аналитиков и инженеров

• Построение автоматической верификации новых данных

• Ускорение поиска ошибок в данных

• Улучшение передачи данных между командами

• Построение документации для данных

Пример использования:

import datetime
import numpy as np
import pandas as pd
import great_expectations as ge
import greatexpectations.jupyterux
from great_expectations.datasource.types import BatchKwargs
создаем контекст
contextge = ge.datacontext.DataContext("..")
создаем набор параметров
expectationsuitename = "bike.table.trips.st" # имя набора
contextge.createexpectationsuite(expectationsuite_name)
определяем тип ресурсов (вот, что мы забыли - 2)
contextge.adddatasource('pandasdatasource', classname='PandasDatasource')
batchkwargs = {'dataset': catalog.load('tripdata'), 'datasource': "pandas_datasource"}
создаем батч с данными и передаем в него имя набора, которое будет наполнять тестами
batch = contextge.getbatch(batchkwargs, expectationsuite_name)

Для дальнейшей работы с GE используйте правила

# используем разные expectations для создания правил по данными
batch.expectcolumnvaluestobeunique('TripID')
зафиксируем правила по данным и сохраним их
batch.saveexpectationsuite(discardfailedexpectations=False)
зафиксируем время создание данного обзора
run_id = {
  "runname": "bike.train.parttrip",
  "run_time": datetime.datetime.utcnow()
}
results = contextge.runvalidationoperator("actionlistoperator", assetstovalidate=[batch], runid=run_id)
сделаем валиадацию данных
df.tail(100).validate(expectationsuite=batch.getexpectation_suite())  #result
создаем html документацию
contextge.builddata_docs()
contextge.opendata_docs()

GE сделаем ваши данные:

Использование GE в Kedro

def data_qual(df: pd.DataFrame):
    """
    Функция для тестирования данных в pipeline
	Одна функция - одная проверка
:param df:     :return: df     """     df = ge.from_pandas(df)
# создаем проверку     result = df.expectcolumnvaluestobeinset('Subscriber_Type',                                                   list(df['Subscriber_Type'].unique()),)                                                   #mostly=.99)
if not result['success']:         err = result["exception_info"]         raise Exception(f"You get unexpected data in Subscriber_Type column\n{err}")
return df

Добавляем в пайплан Kedro

# Создаем отдельный DataQuality pipeline
checkedpipe = Pipeline([node(q.dataqual,
                             "trip_data",
                             "tripdatacheked")
                        ])
return { 		# ставим на позицию перед основным pipeline         "de": checkedpipe + depipe,

Feature Store

Существует проблема передачи фичей для моделей между командами и DS

Feature Store - инструмент, который получает таблицу с данными и отправляет на хранение. Имеет 2 вида объектов: ключи и данные (ввиде колонок)

Разница между DWH и Feature Store

Задание для самостоятельной работы

Задание

---

Docker

В этой части я попытался ответить на вопрос: "Нужен ли Docker для DS/DE?". Это отличное решение для изоляции разработок и мгновенной передачи в прод..

Выбранные решения на базе Kedro + GitHub + GitHub Action позволяют пронести процесс от разработки до готового контейнера

Формируем Docker контейнер на базе файла Dockerfile:

• FROM - на какой базе
• ENV - установка окружения
• WORKDIR – создание директории для работы
• COPY / ADD– копируем с «машины» в Docker контейнер
• RUN - первые команды в Shell (выполнено / установлено)
• EXPOSE - указание портов
• CMD / ENTRYPOIN - что запустить в докере [команда, параметры,,,]

Пример:

FROM python:3.7
ENV PYTHONDONTWRITEBYTECODE 1 ENV PYTHONUNBUFFERED 1
RUN apt-get update \     && apt-get -y install gcc make \     && rm -rf /var/lib/apt/lists/*
RUN pip install --no-cache-dir --upgrade pip
WORKDIR /model
copy requirements.txt
COPY deployml_course/requirements.txt /model/requirements.txt
install project requirements
RUN pip install --no-cache-dir -r requirements.txt
copy project
COPY deployml_course/ /model
EXPOSE 6789
ENTRYPOINT [ "python" ] CMD [ "runner.py"]

Используейте .dockerignore (загружайте необходимый минимум в контейнер) + GitHub Action для деплоя ваших разработок.

Задание для самостоятельной работы

Задание

---

AirFlow

Рассмотрим ETL для DE части нашей моделе. Мы не сразу используем AirFlow, а сначала посмотрим простой инструмент для ETL - Bonobo

Файл docker-compose для деплоя AirFlow + PostgreSQL в виде Docker контейнеров.

В это части, мы попытаемся решить проблемы ETL процесса (и найдем инструмент, который это решит):

• Масштабируемость

• Работа с «неудачными» выполнениям

• Мониторинг

• Зависимости

• Сохранение историчности

Простой ETL (+ bonobo)

Пример Bonobo ETL для работы с HTTP API

import bonobo
import requests
from bonobo.config import usecontextprocessor
from bonobo.config import use
используем http из bonobo
@use('http')
def extract(http):
    """
        Получаем данные с сервиса
    """
    yield from http.get('http://IP:9876/data').json().get('data')
def get_services():     """        Указываем тип сервиса     """     http = requests.Session()     return {         'http': http     }               def withopenedfile(self, context):     """         Файл для записи результатов     """     with open('output.txt', 'w+',  encoding='utf-8') as f:         yield f
на основе контекста, который работыет с файлом
мы создаем контролиремый процесс записи
@usecontextprocessor(withopenedfile)
def writereprto_file(f, *row):
    """
       Записываем все полученные строки в файл
    """
    f.write(repr(row) + "\n")
def get_graph(**options):     """        создаем граф выполнения     """          graph = bonobo.Graph()          graph.add_chain(         extract,         writereprto_file,     )     return graph
выполним и посмотрим на результат
bonobo.run(getgraph(), services=getservices())

Bonobo + Kedro

from kedro.context import load_context
from kedro.extras.datasets.api import APIDataSet
def kedrodepipeline():     """         подключаем нужный pipeline     """     context = load_context("../")     context.run(pipeline_name='de')
def get_graph(**options):     """        создаем граф выполнения     """          graph = bonobo.Graph()          graph.add_chain(         kedrodepipeline             )     return graph
bonobo.run(get_graph())

ETL на AirFlow

AirFlow является отличным решением для сложных задач, этот инструмент решает ETL проблемы и имеет следующие приемущества:

• наличие наглядного веб-GUI для визуализации конвейеров обработки данных

• есть scheduler (у остальных, только cron)

• пакетный характер работы с данными (batch processing)

• популярность в области Big Data(много контекстов для BigData решений)

• есть историчность данных

• есть общее хранилище для переменных и параметров

• есть сенсоры, которые могут управлять DAG

AirFlow имеет набор приемуществ, которые позволяеют ему реализовывать:

• Sensors (на git):

* HdsfSensor - ожидает появления нового файла в таблице Hadoop * NamedHivePartitionSensor - проверяет доступность партиций в Hive таблице * DateTime / TimeDelta- зависимость от времени * Filesystem / FileSensor - если появился новый файл в заданной директории * Python - если какой-нибудь(указаный в условиях) python файл вернул True

!NB Вы всегда можете добавить свой собственный сенсор:

from airflow.sensors.base import BaseSensorOperator

class MySensor(BaseSensorOperator):          @apply_defaults     def init(self, , параметры для данного сенсор, *kwargs):         super().init(**kwargs)         ...
def poke(self, context):         ...

---

• Operators (на git):

* BashOperator * PythonOperator * HiveOperator * HttpOperator * Postgres/MySqlOperator * SubDag !NB Вы всегда можете добавить свой собственный оператор:

from airflow.models import BaseOperator
class MyBestOperator(BaseOperator):          @apply_defaults     def init(self, , параметры для данного оператора, *kwargs):         super().init(**kwargs)         ...              def execute(self, context):         ...

---

• Transfers:

* MySql/PostgresToHiveTransfes

---

Рассмотрим пример:

# Шаг 1
from datetime import datetime, timedelta
from airflow import DAG
from airflow.models import Variable
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
+ создаем функции
def my_func():
    return "Dream func - Всё необходимое в одной функции"
установка Apache Airflow Variables 
/admin/variable/ -> Create new | или загрузить json с переменными
AUTHOR = Variable.get("descdict", deserializejson=True)['dml']['author']

Шаг 2 (аргументы)
default_args = {
    "owner": "airflow",
    "dependsonpast": False,              # зависимость от прошлого результата
    "startdate": datetime(2020, 12, 21),  # первая дата выполнения -> airflow.utils.dates.daysago(#)
    # "end_date":                          # последняя дата (выполнять до)
    "retries": 1,                          # повторных запусков
    "retry_delay": timedelta(minutes=2),   # повторный запуск через кол-во минут
    # email | emailonfailure | emailonretry 
    # 'queue': 'bash_queue',
    # 'pool': 'backfill',
    # 'priority_weight': 10,
    # 'waitfordownstream': False,
    # 'dag': dag,
    # 'sla': timedelta(hours=2),
    # 'execution_timeout': timedelta(seconds=300),
    # 'onfailurecallback': some_function,
    # 'onsuccesscallback': someotherfunction,
    # 'onretrycallback': another_function,
    # 'slamisscallback': yetanotherfunction,
    # 'triggerrule': 'allsuccess'
}
Шаг 3 (описание)
dag = DAG(
    "stepsinDAG",                          # имя процесса
    description="Все шаги для создания DAG", # описание
    schedule_interval="0 0   *",           # аналогично, как в cron
    defaultargs=defaultargs,
    catchup=False                            # catchup - концепция
    
    # catchup - DAG разбивается на шаги, каждый шаг - это отдельный запуск. 
    # При параметре True, DAG будет выполнятся по отдельности, без очередности (каждый шаг в разный момент)
)
Шаг 4
task1 = DummyOperator(taskid="dummytask",
                      retries=3,
                      dag=dag)
Документирование каждого задани
task1.doc_md = """
Task 1
Здес описано задание 1 для Apache Airflow """
dag.docmd = doc
task2 = PythonOperator(taskid="myfunc_task",                        pythoncallable=myfunc,                        dag=dag)
можно добавить визуальный шаблон
templated_command = """
echo "{{ var.value.AUTHOR }}"
echo "{{ params.best_param }}"
"""
task3 = BashOperator(     task_id='templated',     dependsonpast=False,     bashcommand=templatedcommand,     params={'best_param': 'О, да! Ты подобрал лучшие параметры'},     dag=dag, )

Шаг 5 (установка последовательности)
установка последовательности
task1 >> task2 >> task3
равнозначно
task2.set_upstream(task1)
task1.set_downstream([task2, task3])
task1 >> [task2, task3]
в обратном порядке
[task2, task3] << task1

AirFlow + Kedro

import sys
from datetime import datetime, timedelta
from airflow import DAG from kedro_airflow.runner import AirflowRunner from kedro.framework.context import load_context
Установка аргументов
default_args = {
    "owner": "kedro",
    "start_date": datetime(2020, 12, 21),
    "dependsonpast": False,
    "waitfordownstream": True,
    "retries": 1,
    "retry_delay": timedelta(minutes=10),
}
Функция создания контекста и определения pipeline для запуска
def processcontext(datacatalog, **airflow_context):
    
    for key in ["dag", "conf", "macros", "task", "task_instance", "ti", "var"]:
        del airflow_context[key]
        
    datacatalog.addfeeddict({"airflowcontext": airflow_context}, replace=True)
parameters = data_catalog.load("parameters")     parameters["airflowdepipeline"] = airflow_context["de"]     data_catalog.save("parameters", parameters)
return data_catalog
Создаем DAG для Kedro
dag = DAG(
    "bike",
    description="DE pipeline для bike модели",
    defaultargs=defaultargs,
    schedule_interval=None,
    catchup=False
)
загружаем контекст kedro (как обычно, как в Bonobo)
context = loadcontext("")
datacatalog = context.catalog
pipeline = _context.pipeline
создаем airflow процесс
runner = AirflowRunner(
                       dag=dag,
                       processcontext=processcontext,
                       operator_arguments=dict(),
                       )
инит процесса == task1 >> task2 >> ...
runner.run(pipeline, data_catalog)

kedro может больше, kedro сам может создать себе готовый код для airflow:

# наберите в командной строке, там где у вас расположен проект 
kedro airflow create

Задание для самостоятельной работы

Задание

---

Мониторинг

Метрик много, но какие выбрать для мониторинга за моделью, а какие принять для определения качества моделей?

Не достаточно смотреть на кривые и площади под ними. Но можно с этого начать

Для классификации

Для регрессии

Не забудьте проверить входные параметры для модели

Самый эффективный способ - это следить за стабильностью данных, т.е. использовать PSI - population stability index

Результат при изменении данных

Мониторинг в прод

Метрики для Model Quality не всегда подохят для мониторинга в работе.

Разберем пример на примере MCC

И сравним его результаты с F1 для бинарной классификации

AB Test

Рассмотрим 2 варианта АБ тестирования:

• новую модель со старой моделью

• все модели со всеми

import pandas as pd
%matplotlib inline
import math import scipy.stats as stats
import numpy as np import matplotlib.pyplot as plt
from statsmodels.stats.proportion import proportions_ztest from statsmodels.stats.proportion import proportions_chisquare
import seaborn as sns sns.set(color_codes=True)
from datetime import timedelta import datetime import time import os

# Цвета для виз
colorhexlist = ['#88CCEE', '#CC6677', '#DDCC77', '#117733', '#332288', '#E58606',  '#999933', '#882255', '#661100', 
 '#5D69B1', '#52BCA3', '#99C945', '#CC61B0', '#24796C', '#DAA51B', '#2F8AC4', '#764E9F', '#ED645A']

---

НАСТРОЙКИ ДО ТЕСТА

Размер группы

Если вы не хотите считать сами, воспользуйтесь готовы решением на сайте: ссылка

# установка базовых переменных
baseline_probability = 0.10
beta = 0.2
alpha = 0.05
effectSize = 0.01
определяем one-side или two-side тест
one_sided = True

Рассчитаем стартовые данные

alphaZscore = stats.norm.ppf(1-alpha(1-0.5(1-onesided)))
beta_Zscore = stats.norm.ppf(1-beta)
print('Baseline Conversion (pr) = ', baseline_probability*100, '%') print('Confidence Level = ', (1-alpha)*100, '%') print('Alpha Z-score (za) = ', round(alpha_Zscore,2)) print('Power = ', (1-beta)*100, '%') print('Beta Z-score (zb) = ', round(beta_Zscore,2)) print('Effect Size (E) = ', effectSize*100, '%')

Baseline Conversion (pr) = 10.0 % Confidence Level = 95.0 % Alpha Z-score (za) = 1.64 Power = 80.0 % Beta Z-score (zb) = 0.84 Effect Size (E) = 1.0 %

Функция установки "траффика", который будет проходить через модель

def sample_size(E, za, zb, pr):
    variance = pr*(1-pr)  # эту часть можно заменить на baseline по модели
    z = (za+zb)**2
    top = 2variancez
    bottom = E**2
    n = top/bottom
    return n
n = samplesize(effectSize, alphaZscore, betaZscore, baselineprobability) print('Sample Size (per variation)=',round(n,0))

Sample Size (per variation)= 11129.0

Определяем минимальный значимый эффект

• расчет на основе alpha / beta / размер группы
• для обнаружения минимального эффекта, мы должны заранее определить размер группы (максимальный)

# функция оценки эффекта
def measurableeffectsize(n, za, zb, pr):
    variance = pr*(1-pr)  # эту часть можно заменить на baseline по модели
    z = (za+zb)**2
    top = 2variancez
    bottom = n
    E = math.sqrt(top/n)
    return E
минимальный эффект
measurableeffectsize(n, alphaZscore, betaZscore, baseline_probability)

0.01

Alpha Cutoff

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

Мы считаем - Familywise Error Rate (Alpha Inflation)

# количество сегментов для сравнения (кол-во значений в таргите)
segments = 2

# вероятность ошибки в тести, определеяем для действия - alpha cutoff
1 - (1-alpha)**segments

0.09750000000000003

print("Допустимые коррекции:")
print("Тестирование гипотез на уровне α∗ = (α/segments) = ({a}/{seg}) = {aseg} (где {seg} - кол-во сегментов в группе).".format(a=alpha, seg=segments, aseg = alpha/segments))
print("Гарантированная ошибка 1 типа не будет привышать α = {}".format(round((1 - (1-alpha)**segments),3)))
print("However, this adjustment may be too conservative.")

Допустимые коррекции: Тестирование гипотез на уровне α∗ = (α/segments) = (0.05/2) = 0.025 (где 2 - кол-во сегментов в группе). Гарантированная ошибка 1 типа не будет привышать α = 0.098 However, this adjustment may be too conservative.

---

Данные

Для теста надо иметь в данных следующее:

• id пользователя
• дата
• тип группы
• параметры для сравнения

file_name = 'Data/sd.csv'
datesort = 'datevisit'
загрузим
data = pd.readcsv(filename,
                  converters= {datesort: pd.todatetime} )
df = data.copy()
df = df.sortvalues(datesort, ascending = True)
df = df[df['version'].isin(['A1', 'A2'])]
Определение конверсий
traffic_label = 'clicked'
conversion_label = 'liked'
Группировка
userlabel = 'userid'
datelabel = 'datevisit'
segment_label = 'version'
segmentlist = list(df[segmentlabel].unique())
df.head()

Обзор сегментов

variations = len(segment_list)
print('Different segments: ',segment_list)
print('Number of segments: ', variations)

Different segments: ['A1', 'A2'] Number of segments: 2

Траффик по дням

# Агрегат по дню
dailyusers = pd.DataFrame(d2.groupby(datelabel)[userlabel].nunique()).resetindex()
dailytraffic = pd.DataFrame(d2.groupby(datelabel)[trafficlabel].sum()).resetindex()
dailyconversions = pd.DataFrame(d2.groupby(datelabel)[conversionlabel].sum()).resetindex()
Визуализируем
plt.subplots(figsize=(13, 6))
plt.plot(pd.todatetime(dailyusers[datelabel]), dailyusers[user_label], label = 'users')
plt.plot(pd.todatetime(dailytraffic[datelabel]), dailytraffic[trafficlabel], label = ('traffic: ' + trafficlabel))
plt.plot(pd.todatetime(dailyconversions[datelabel]), dailyconversions[conversionlabel], label = ('conversion: ' + conversionlabel))
plt.xlabel('Date', fontsize=15)
plt.xticks(fontsize=15, rotation=30)
plt.yticks(fontsize=15)
plt.title('Daily: Users, Traffic & Conversions', fontsize=18)
plt.legend(fontsize=15)
plt.show()

# Conversion Rate
round((dailyconversions[conversionlabel]/dailytraffic[trafficlabel]).mean()*100,2)

40.43

# Average Traffic / User
round((dailytraffic[trafficlabel]/dailyusers[userlabel]).mean(),2)

0.96

# Average Conversions / User
round((dailyconversions[conversionlabel]/dailyusers[userlabel]).mean(),2)

0.39

Таблица сегментов

# выделение колонок для агрегата'traffic' / 'conversion'
aggregationcolumn = [trafficlabel, conversion_label]
traffic = []
conversions = []
расчет агрегатов
for i in range(variations):
    v, c = df[df[segmentlabel] == segmentlist[i] ][aggregation_column[:2]].sum()
traffic.append(v)     conversions.append(c)           
новый DF   
 dfp_simple = pd.DataFrame({     "converted": conversions,     "traffic": traffic }, index = segment_list)

dfp_simple
#dfp = dfpsimple.copy().sortindex()

dfp = dfpsimple.copy().sortindex()
Сумма по всем колонкам
dfp.loc['Total'] = dfp.sum()
определение не сконвертируемых
dfp['not_converted'] = dfp['traffic'] - dfp['converted']
отношение конвертируемых к траффику
proportion = dfp.converted/dfp.traffic
dfp['converted_proportion'] = proportion
STD
dfp['standard_error'] = ((proportion  (1-proportion))/dfp.traffic)*(.5)
% траффика
n = dfp.loc['Total']['traffic']
dfp['population_percentage'] = dfp['traffic']/n
dfp

Plot

# визуализируем BarPlot
labels = dfp.index[:-1].tolist()
label_index = np.arange(len(labels))
values = (round(dfp['converted_proportion'][:-1]*100,2)).tolist()
halfci = (round(dfp['standarderror'][:-1]1.96100,2)).tolist()
plt.subplots(figsize=(13, 6))
plt.bar(labelindex, values, yerr=halfci, alpha=0.75,
       color= colorhexlist[0:len(labels)],
       error_kw=dict(ecolor='black', lw=2, capsize=5, capthick=2))
plt.xlabel('Segment', fontsize=15)
plt.ylabel('% Conversion', fontsize=15)
plt.xticks(label_index, labels, fontsize=13, rotation=30)
plt.yticks(fontsize=15)
plt.title('Conversion Rate / Segment', fontsize=18)
plt.show()

ТЕСТИРОВАНИЕ МОДЕЛИ ЛОКАЛЬНО

2 Sample

Тип теста: z-test

• H0: Нет разницы между сегментами и уровнем конверсии
• H1: Есть разница между сегментами и уровнем конверсии

!-> A1 - основная, старая модель, все остальные это новые "челленджеры"

variation1 = 'A1'
variation2 = 'A2'

Расчет параметров по каждому сегменту

x1 = dfp.loc[variation1]['converted']
n1 = dfp.loc[variation1]['traffic']
x2 = dfp.loc[variation2]['converted']
n2 = dfp.loc[variation2]['traffic']
print(variation1, 'Converted:', x1) print(variation1, 'Traffic:', n1) print(variation2, 'Converted:', x2) print(variation2, 'Traffic:', n2)

A1 Converted: 42.0 A1 Traffic: 86.0 A2 Converted: 37.0 A2 Traffic: 88.0

Сравнение результатов

p1 = x1/n1
p2 = x2/n2
p = (x1+x2)/(n1+n2)
print('Кол-во (', variation1 ,'): {0:.2f}%'.format(100*p1))
print('Кол-во (', variation2 ,'): {0:.2f}%'.format(100*p2))
print('Среднее кол-во по всем группам: {0:.2f}%'.format(100*p))
print('% разница между группами: {0:.2f}%'.format(100*(p2-p1)))
print('% относительная разница между группами: {0:.2f}%'.format(100*(p2-p1)/p1))
var = p(1-p)(1/n1+1/n2) se = math.sqrt(var) print('\nVariance: {0:.4f}%'.format(100*var)) print('Standard Error: {0:.2f}%'.format(100*se))
z = (p1-p2)/se pvalue = 1-stats.norm.cdf(abs(z)) pvalue *= 2-one_sided print('\nz-stat: {z}'.format(z=z)) print('p-value: {p}'.format(p=pvalue))

Кол-во ( A1 ): 48.84% Кол-во ( A2 ): 42.05% Среднее кол-во по всем группам: 45.40% % разница между группами: -6.79% % относительная разница между группами: -13.91% Variance: 0.5699% Standard Error: 7.55% z-stat: 0.8996463724803273 p-value: 0.184154235386902

Фиксирование разниц

m_proportion = abs(min((1-p1),(1-p1))-1)
mindetectableeffectsize = measurableeffectsize(min([n1, n2]), alphaZscore, betaZscore, mproportion)
print("Для текущего размера группы {n} и с baseline {m:.2f},\nмы можем получить разницу в {e:.2f}%."
      .format(n = min([n1, n2]), m = mproportion, e=mindetectableeffectsize*100))
print('\n')
n1 = samplesize(p2-p1, alphaZscore, betaZscore, baselineprobability) print("Разница {d:.2f}%, которая требуется для минимальной группы {n_needed}"       .format(d=abs(100*(p2-p1)), n_needed = round(n1,0))) print('\n')
print("Alpha - {a:.2f}\nBeta - {b:.2f}".format(a = alphaZscore, b = betaZscore))

Для текущего размера группы 86.0 и с baseline 0.49, мы можем получить разницу в 18.95%. Разница 6.79%, которая требуется для минимальной группы 241.0 Alpha - 1.64 Beta - 0.84

Нужно ли больше данных?

abs(p2-p1) >= mindetectableeffect_size
if abs(p2-p1) >= mindetectableeffect_size:     print("Нет, данных достаточно для определения значимости") else:     print("Да, мы должны получить больше данных для определения значимости изменений")

Да, мы должны получить больше данных для определения значимости изменений

Интерпритация результатов

! -> Только, если больше не нужно больше данных

print('p-value {p} меньше, чем alpha, {alpha}?\n'.format(p=round(pvalue,5), alpha=alpha))
if (pvalue < alpha):
    print('p-value меньше, чем alpha, 0-гипотеза отвергается (null-hypothesis = no difference)')
else: 
    print('Нет, 0-гипотеза не может быть отвергнута')

p-value 0.18415 меньше, чем alpha, 0.05? Нет, 0-гипотеза не может быть отвергнута

Доверительный интервал - для эффективного размера группы

# z-statistics
zcritical = stats.norm.ppf(1-alpha(1-0.5(1-onesided)))
верхний и нижний уровень значимости
cilow = (p2-p1) - zcritical*se
ciupp = (p2-p1) + zcritical*se
print(' 95% Confidence Interval = ( {0:.2f}% , {1:.2f}% )'       .format(100cilow, 100ciupp))

95% Confidence Interval = ( -19.21% , 5.63% )

Stats Model Formula

2х хвостовой тест

counts = np.array([x1, x2])
nobs = np.array([n1, n2])
stat, pval = proportions_ztest(counts, nobs, alternative = 'smaller') print('z-stat: {0:.4f}'.format(stat)) print('p-value: {0:.8f}'.format(pval))

z-stat: -4.6325 p-value: 0.00000181

Extra

# 2х хвостовой тест результат
pvalue = 1-stats.norm.cdf(abs(z))
pvalue *= 2-False
print('p-value: {p}'.format(p=pvalue))

p-value: 0.368308470773804

# 1 хвостовой тест
pvalue = 1-stats.norm.cdf(abs(z))
pvalue *= 2-True
print('p-value: {p}'.format(p=pvalue))

p-value: 0.184154235386902

2+ Sample Proportion

Тип теста: Chi Square

Гипотезы

• H0: Нет разницы между сегментами и уровнем конверсии
• H1: Есть разница между сегментами и уровнем конверсии

Main Question:

Есть разница между моделями (сегментами)?

# Повторю загрузку для добавления сегментов
file_name = 'Data/sd.csv'
datesort = 'datevisit'
data = pd.readcsv(filename,
                  converters= {datesort: pd.todatetime} )
df = data.copy() df = df.sortvalues(datesort, ascending = True) df = df[df['version'].isin(['A1', 'A2','B', 'C'])]
traffic_label = 'clicked' conversion_label = 'liked' userlabel = 'userid' datelabel = 'datevisit' segment_label = 'version' segmentlist = list(df[segmentlabel].unique()) variations = len(segment_list)
aggregationcolumn = [trafficlabel, conversion_label] traffic = [] conversions = []
for i in range(variations):     v, c = df[df[segmentlabel] == segmentlist[i] ][aggregation_column[:2]].sum()          traffic.append(v)     conversions.append(c)      dfp_simple = pd.DataFrame({     "converted": conversions,     "traffic": traffic},      index = segment_list)
dfp = dfpsimple.copy().sortindex() dfp.loc['Total'] = dfp.sum() dfp['not_converted'] = dfp['traffic'] - dfp['converted'] proportion = dfp.converted/dfp.traffic dfp['converted_proportion'] = proportion dfp['standard_error'] = ((proportion  (1-proportion))/dfp.traffic)*(.5) n = dfp.loc['Total']['traffic'] dfp['population_percentage'] = dfp['traffic']/n dfp

dfpTo = dfp[['converted', 'not_converted', 'traffic']].T
dfpTo

% пропорций (вероятности)

dfpTe = dfpTo.copy()
заменим вероятностями
for i in range(variations):
    for j in range(0, 2):
        dfpTe.iloc[j,i] = (dfpTo.loc['traffic'][i]*dfpTo['Total'][j])/n

Chi Square / p-value

o = dfpTo.drop(dfpTo.columns[-1], axis = 1)[:-1]
e = dfpTe.drop(dfpTe.columns[-1], axis = 1)[:-1]
ss = (o-e)**2/e ch2 = ss.values.sum() ch2

1.9666036399234188

# P-value (степень свободы - 1)
pvalue_chi = 1 - stats.chi2.cdf(ch2, variations-1)
pvalue_chi

0.5793670974852936

Интерпретация результатов

```python print('p-value {p} меньше, чем alpha, {alpha}?\n'.format(p=round(pvalue_chi,5), alpha=alpha)) if (pvalue < alpha): print('p-value меньше, чем alpha, 0-гипотеза отвергается (null-hypothesis = no difference)') else: print('Нет, 0-гипотеза не може

---

README truncated. View on GitHub