В новой статье CleverData мы расскажем о проектировании ML-пайплайна предсказания целевого действия с помощью Yandex Cloud. Пайплайн необходим для автоматического обмена данными с CDP CleverData Join — использования информации с платформы для обучения ML-моделей и формирования прогнозов поведения каждого пользователя. На примерах рассмотрим использование API сервисов Yandex Cloud, коснемся алгоритмов обработки данных и обучения ML-модели, а также расскажем о возникших проблемах. Под катом делимся кодом.
Что такое CDP и зачем нужны предиктивные атрибуты
Платформа управления клиентскими данными (Customer Data Platform) — это инструмент, помогающий компаниям собирать, анализировать и использовать информацию о своих клиентах. CDP объединяет данные из таких источников, как сайты, мобильные приложения, CRM-системы и т.п., чтобы сформировать полное представление о портрете клиента и истории его действий.
CDP CleverData Join содержит два основных типа данных: клиентские профили и события, привязанные к этим профилям. В состав профиля входят идентификаторы клиента и атрибуты, содержащие информацию о пользователе (его имя, демографические характеристики, предпочтения и многое другое). Атрибуты событий представляют собой информацию о конкретных действиях клиентов. Например, посещения страниц сайта, факты покупки товаров, реакции на маркетинговые кампании. Но CDP — это не только база данных. Это инструмент, который помогает делать полезные выводы о клиентах и повышать эффективность маркетинга.
Основная часть данных собирается из внешних систем, интегрированных с CDP. Однако при наличии такого массива данных возникает желание получить новую, предиктивную характеристику профиля, сформированную методом машинного обучения на собранных сведениях. Знать не то, что человек уже сделал, а предугадывать его следующие шаги.
В CDP предиктивные атрибуты, как правило, используются для прогнозирования поведения клиентов, определения их потребностей и предсказания вероятности совершения определенных действий. Например, для маркетологов может быть полезно иметь прогноз того, что клиент возьмет кредит или оформит подписку. Это можно вычислить при помощи ML-модели, обученной на исторических данных о поведении клиента. Предиктивные атрибуты играют важную роль в персонализации маркетинговых кампаний, улучшении обслуживания клиентов.
Конечно же, нужно понимать, что для моделирования предиктивных атрибутов требуется:
-
иметь качественные, структурированные данные для моделирования,
-
обеспечить достаточный объем таких данных,
-
обладать качественными алгоритмами предобработки и обучения.
Не менее важным вопросом является обслуживание моделей — их надо регулярно обновлять и дообучать на свежих данных, иначе качество прогнозирования будет неизбежно падать. Построению пайплайна, который обеспечивает автоматизированное взаимодействие с CDP, и посвящена эта статья.
MVP и его цели
Мы решили построить MVP пайплайна, реализующего передачу данных для обучения, само обучение и передачу предсказаний обратно в CDP CleverData Join, используя готовые инструменты, предоставляемые Yandex Cloud.
В качестве бизнес-задачи мы для себя определили предсказание вероятности совершения клиентом целевого действия на основании данных о переходах по страницам сайта — кликстриме. Это типовая задача бинарной классификации (классификация по определенным признакам в две группы) массива клиентов. Под целевым действием подразумевается посещение нужного нам URL (например, страницы оформления покупки или кредита). Поскольку кликстрим является цепочкой переходов пользователей по ссылкам, в качестве архитектуры модели была выбрана рекуррентная нейронная сеть (вид нейронных сетей, где связи между элементами образуют направленную последовательность).
Целью проекта стало построение полностью автоматизированного пайплайна, который приносил бы ценность заказчику в течение длительного времени. Для этого нам нужно было реализовать следующие цепочки операций.
-
Предобработка новых данных для обучения и инференса (непрерывной работы нейронной сети на устройствах конечного пользователя) и преобразование их из формата хранения в CDP в формат, требуемый для ML-модели.
-
Первичное обучение модели под задачу конкретного заказчика на кликстриме его клиентов.
-
Периодическое дообучение модели на новых данных, поступающих в CDP. Дообучение на новых данных крайне важно, поскольку с течением времени модель может потерять свою актуальность, а, значит, точность прогнозов значительно уменьшится.
-
Периодический инференс (обеспечение непрерывной работы нейронной сети на устройствах конечного пользователя) для клиентов с обновлениями в кликстриме и отгрузку результатов обратно в CDP.
Архитектура компонентов MVP в Yandex Cloud
Итак, нам потребовалось разработать и развернуть в Yandex Cloud сервисы, реализующие все необходимые для построения MVP цепочки операций. После анализа требований была предложена следующая архитектура решения:
На стороне CleverData Join находятся сервисы Model Manager, отвечающие за автоматизированное создание инфраструктуры под пайплайн в Yandex Cloud, и Data Processor, обеспечивающий периодическую первичную обработку новых данных и загрузку их в бакет S3. По сути Model Manager является Control Plane, который автоматизирует операции по развертыванию компонентов в Yandex Cloud при появлении потребности создать новый предиктивный атрибут. Остальные компоненты находятся в инфраструктуре Yandex Cloud.
-
В бакете Yandex Object Storage накапливаются данные, представляющие собой первично обработанный кликстрим. Там же должны располагаться артефакты и метаданные, полученные при обучении моделей, а также результаты инференса моделей.
-
В сервисе Yandex DataSphere автоматически создаётся проект, содержащий ноутбуки для обучения и инференса. При поступлении новых данных в Yandex Object Storage должен запускаться необходимый ноутбук. В ноутбуке обучения реализованы вторичная обработка данных на кластере Yandex DataProc, обучение модели непосредственно в сервисе, загрузка артефактов модели на S3. В ноутбуке инференса — получение предсказаний по новым данным, отгруженным на S3.
-
В сервисе Yandex Cloud Functions располагаются функции, необходимые для запуска ноутбуков, и триггеры, которые привязаны к обновлению данных в Yandex Object Storage по определенным при их создании префиксам. Таким образом, после отгрузки новых данных на бакет производится, в зависимости от расположения данных, обучение/дообучение модели, инференс модели и отгрузка предсказаний в REST API CleverData Join.
Так как важно обеспечить одновременную работу нескольких моделей совершения целевого действия для разных тенантов в системе CD Join, для каждой задачи нужен свой отдельный проект в DataSphere со своими шаблонными ноутбуками, облачными функциями и триггерами. Чтобы процесс был максимально автоматизирован, необходимо, чтобы каждый проект был преднастроен:
-
в него были загружены шаблонные ноутбуки,
-
был активирован готовый Docker-образ для проекта DataSphere.
Забегая вперёд, отметим, что с автоматизацией преднастройки проекта возникли некоторые затруднения, но об этом мы расскажем позднее.
Немного о предобработке данных
Data Processor запускается по расписанию и производит первичную обработку данных и записывает результат в хранилище Yandex Object Storage.
События кликстрима хранятся в CleverData Join в формате AVRO. Предобработка происходит в два этапа.
-
На первом этапе из событий кликстрима извлекаются идентификатор клиента, метка времени и URL перехода. Ссылки очищаются от технических доменов и обрезаются до верхнеуровневых доменов. Данные сохраняются в формате Parquet.
-
Далее для каждого идентификатора клиента формируется цепочка из посещенных URL в хронологическом порядке и дельт, то есть временных интервалов, прошедших между посещениями каждого URL.
Процесс, организованный в два этапа, позволяет избегать повторной первичной обработки старых событий. Старые события хранятся в формате Parquet в S3 бакете и периодически дополняются первично обработанными событиями за новые периоды. После этого происходит пересборка цепочек для клиентов, по которым есть обновления.
Автоматизированное создание инфраструктуры
У сервисов Yandex Cloud есть API, позволяющий взаимодействовать с ними без использования UI, что критически важно для реализации управления процессом в Model Manager. В этом разделе мы пройдём по всем автоматизированным через API операциям, а именно:
-
созданию сервисного аккаунта;
-
созданию и настройке сети VPC для кластера Data Proc;
-
созданию проекта в DataSphere;
-
созданию S3 бакета и получению статического ключа доступа;
-
созданию и настройке Cloud Functions и их триггеров.
Сначала создадим сервисный аккаунт, от имени которого будет происходит взаимодействие с ресурсами Yandex Cloud:
def create_service_account(folder_id: str, name: str, iam_token: str, description: str = ''): url = 'https://iam.api.cloud.yandex.net/iam/v1/serviceAccounts' headers = {"Authorization": f"Bearer {iam_token}"} data = { "folderId": folder_id, "name": name, "description": description, } response = requests.post(url, headers=headers, json=data) return response sa_response = create_service_account(folder_id = folder_id, name = 'ta-service-account', iam_token = iam, description = 'ta-service-account')Теперь, чтобы у сервисного аккаунта хватало прав на взаимодействие с необходимыми сервисами Yandex Cloud (DataSphere, Object Storage, Data Proc, Cloud Functions, VPC), выдадим ему необходимые роли. Для каталога это:
-
editor — нужна для чтения, записи и управления бакетом S3;
-
vpc.user, vpc.admin — необходимы для настройки доступа кластеру в интернет;
-
dataproc.agent — для использования кластера Data Proc;
-
dataproc.admin — для создания и удаления кластера Data Proc.
def update_access_bindings(resource_id : str, folder_id: str, roles: list = ['iam.serviceAccounts.user', 'editor', 'vpc.user', 'vpc.admin', 'dataproc.admin', 'dataproc.agent'], iam_token: str = iam): url = f'https://resource-manager.api.cloud.yandex.net/resource-manager/v1/folders/{folder_id}:updateAccessBindings' headers = {"Authorization": f"Bearer {iam_token}"} access_binding_deltas = [ { "action": "ADD", "accessBinding": { "roleId": role, "subject": { "id": resource_id, "type": "serviceAccount" } } } for role in roles ] payload = { "accessBindingDeltas": access_binding_deltas } response = requests.post(url, json=payload, headers=headers) return response response_update_access_bindings = update_access_bindings( resource_id=service_account_id, folder_id=folder_id)Также необходимо назначить роль сервисному аккаунту, чтобы можно было им управлять. Роль должна быть назначена как на ресурс:
def update_access_bindings_1(resource_id: str, roles: list = ['datasphere.community-projects.editor', 'iam.serviceAccounts.user', 'functions.functionInvoker'], iam_token: str = iam): url = f"https://iam.api.cloud.yandex.net/iam/v1/serviceAccounts/{resource_id}:updateAccessBindings" headers = {"Authorization": f"Bearer {iam_token}"} access_binding_deltas = [ { "action": "ADD", "accessBinding": { "roleId": role, "subject": { "id": resource_id, "type": "serviceAccount" } } } for role in roles ] payload = { "accessBindingDeltas": access_binding_deltas } response = requests.post(url, json=payload, headers=headers) return response update_access_response_1 = update_access_bindings_1(resource_id=service_account_id)Для того, чтобы кластер Data Proc мог общаться с внешним миром, нам нужно создать сеть VPC, указать шлюз и сформировать таблицу маршрутизации. В этой сети необходимо выделить подсеть для настройки доступа в интернет при помощи созданного шлюза.
Итак, создаём сеть:
def create_network(folder_id: str, network_name: str, description: str, iam_token: str = iam, ): url = 'https://vpc.api.cloud.yandex.net/vpc/v1/networks' headers = {"Authorization": f"Bearer {iam_token}"} data = { "folderId": folder_id, "name": network_name, "description": description, } response = requests.post(url, headers=headers, json=data) return response create_network_response = create_network(folder_id = folder_id, network_name = 'network-1', description = '' )…, шлюз:
def create_gateway(folder_id: str, gateway_name: str, description: str, iam_token: str = iam, ): url = 'https://vpc.api.cloud.yandex.net/vpc/v1/gateways' headers = {"Authorization": f"Bearer {iam_token}"} data = { "folderId": folder_id, "name": gateway_name, "description": description, "sharedEgressGatewaySpec": {} } response = requests.post(url, headers=headers, json=data) return response response_create_gateway = create_gateway(folder_id = folder_id, gateway_name = 'gateway', description = '' )…, таблицу маршрутизации:
def create_route_table(folder_id: str, route_table_name: str, description: str, gateway_id: str, network_id: str, destination_prefix: str = '0.0.0.0/0', iam_token: str = iam, ): url = 'https://vpc.api.cloud.yandex.net/vpc/v1/routeTables' headers = {"Authorization": f"Bearer {iam_token}"} data = { "folderId": folder_id, "name": route_table_name, "description": description, "networkId": network_id, "staticRoutes": [ { "destinationPrefix": destination_prefix, "gatewayId": gateway_id, } ] } response = requests.post(url, headers=headers, json=data) return response response_create_route_table = create_route_table(folder_id = folder_id, route_table_name = 'route-table', description = '', gateway_id = gateway_id, network_id = network_id, destination_prefix = '0.0.0.0/0' )…, и подсеть с доступом в интернет:
def create_subnet(folder_id: str, subnet_name: str, network_id: str, route_table_id: str, description: str, ip_range: str, zone_id: str, dhcp_options: dict = {}, iam_token: str = iam, ): url = 'https://vpc.api.cloud.yandex.net/vpc/v1/subnets' headers = {"Authorization": f"Bearer {iam_token}"} data = { "folderId": folder_id, "name": subnet_name, "description": description, "networkId": network_id, "zoneId": zone_id, "v4CidrBlocks": [ ip_range ], "routeTableId": route_table_id, "dhcpOptions": dhcp_options } response = requests.post(url, headers=headers, json=data) return response response_create_subnet_a = create_subnet(folder_id = folder_id, subnet_name = 'subnet-ru-central1-a', network_id = network_id, route_table_id = route_table_id, description = 'default subnet for zone ru-central1-a', ip_range = '10.128.0.0/24', zone_id = 'ru-central1-a', dhcp_options = {}, )В процессе работы мы будем создавать кластер Data Proc, на котором производится вторичная обработка данных.
При создании кластера важно не забыть передать идентификатор подсети, для которой мы ранее настраивали доступ в интернет. Нам нужно создать объект конфигурации с информацией о подкластерах, их ролях, типе и размере диска, количестве хостов. Этот объект будет передан в функцию создания кластера.
def create_cluster(folder_id: str, cluster_name: str, description: str, config_spec: dict, service_account_id: str, zone_id: str, bucket: str, ui_proxy: bool, security_group_ids: str, host_group_ids: str, deletion_protection: bool, log_group_id: str, iam_token: str = iam, ): url = 'https://dataproc.api.cloud.yandex.net/dataproc/v1/clusters' headers = {"Authorization": "Bearer {}".format(iam_token)} data = { "folderId": folder_id, "name": cluster_name, "description": description, "configSpec": config_spec, "zoneId": zone_id, "serviceAccountId": service_account_id, "uiProxy": ui_proxy, "deletionProtection": deletion_protection, } response = requests.post(url, headers=headers, json=data) return response config_spec = { "versionId": "2.0", "hadoop": { "services": [ 'HDFS', 'LIVY', 'MAPREDUCE', 'SPARK', 'TEZ', 'YARN', 'ZEPPELIN' ], "sshPublicKeys": [ public_key ], }, "subclustersSpec": [ { "name": "master", "role": "MASTERNODE", "resources": { "resourcePresetId": "s2.micro", "diskTypeId": "network-ssd", "diskSize": "137438953472" }, "subnetId": a_subnet_id, "hostsCount": "1", "assignPublicIp": False, }, { "name": "data", "role": "DATANODE", "resources": { "resourcePresetId": "s2.small", "diskTypeId": "network-hdd", "diskSize": "274877906944" }, "subnetId": a_subnet_id, "hostsCount": "1", "assignPublicIp": False, }, ] } response_create_cluster = create_cluster(folder_id = folder_id, cluster_name = 'xs-proc', description = '', config_spec = config_spec, service_account_id = service_account_id, zone_id = zone_id, bucket = '', ui_proxy = False, security_group_ids = '', host_group_ids = '', deletion_protection = False, log_group_id = '') cluster_id = response_create_cluster.json()['metadata']['clusterId']Так как кластер будет считывать данные с s3, нужно предоставить ему Credentials для доступа к S3:
#!spark --cluster=xs-proc --session preparing_session --variables key --variables sct sc._jsc.hadoopConfiguration().set("fs.s3a.endpoint", "storage.yandexcloud.net") sc._jsc.hadoopConfiguration().set("fs.s3a.signing-algorithm", "") sc._jsc.hadoopConfiguration().set("fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider") sc._jsc.hadoopConfiguration().set("fs.s3a.access.key",key) sc._jsc.hadoopConfiguration().set("fs.s3a.secret.key",sct)Теперь мы готовы создать проект в DataSphere, в нем будут происходить все операции, связанные с обучением и инференсом ML-моделей:
def create_project(project_name: str, community_id: str, service_account_id: str, description: str, subnet_id: str, commit_mode: str, default_folder_id: str, vm_inactivity_timeout: str, default_dedicated_spec: str, data_proc_cluster_id: str = '', security_group_ids: list = [], limits: dict = {}, early_access: bool = False, ide: str = 'JUPYTER_LAB', iam_token: str = iam): url = 'https://datasphere.api.cloud.yandex.net/datasphere/v2/projects' headers = {"Authorization": f"Bearer {iam_token}"} data = { "communityId": community_id, "name": project_name, "description": description, "settings": { "serviceAccountId": service_account_id, "subnetId": subnet_id, "dataProcClusterId": data_proc_cluster_id, "commitMode": commit_mode, "securityGroupIds": security_group_ids, "earlyAccess": early_access, "ide": ide, "defaultFolderId": default_folder_id, "vmInactivityTimeout": vm_inactivity_timeout, "defaultDedicatedSpec": default_dedicated_spec }, "limits": limits } response = requests.post(url, headers=headers, json=data) return response response_create_project = create_project(project_name = 'ta-service-project', community_id = community_id, service_account_id = service_account_id, subnet_id = a_subnet_id, description = 'ta model project', commit_mode = 'AUTO', default_folder_id = folder_id, vm_inactivity_timeout = '1800s', default_dedicated_spec = 'c1.4')Сервисный аккаунт, который из Cloud Functions будет запускать ноутбук, должен быть в участниках проекта DataSphere. Выдадим ему соответствующую роль:
def update_access_bindings_2(resource_id : str, project_id: str, roles: list = ['datasphere.community-projects.editor'], iam_token: str = iam): url = f"https://datasphere.api.cloud.yandex.net/datasphere/v2/projects/{project_id}:updateAccessBindings" headers = {"Authorization": f"Bearer {iam_token}"} access_binding_deltas = [ { "action": "ADD", "accessBinding": { "roleId": role, "subject": { "id": resource_id, "type": "serviceAccount" } } } for role in roles ] payload = { "accessBindingDeltas": access_binding_deltas } response = requests.patch(url, json=payload, headers=headers) return response response_update_access_bindings_2 = update_access_bindings_2(resource_id=service_account_id, project_id=project_id)Вы можете использовать для обмена данными уже существующий бакет S3. Но, если его нет, то его создание также не представляет сложности:
def create_bucket(bucket_name: str, folder_id: str, default_storage_class: str, max_size: str, anonymous_access_flags: dict, acl: dict, versioning: str, tags: list, iam_token: str = iam, ): url = 'https://storage.api.cloud.yandex.net/storage/v1/buckets' headers = {"Authorization": f"Bearer {iam_token}"} data = { "name": bucket_name, "folderId": folder_id, "defaultStorageClass": default_storage_class, "maxSize": max_size, "anonymousAccessFlags": anonymous_access_flags, "versioning": versioning, "acl": acl, "tags": tags } response = requests.post(url, headers=headers, json=data) return response bucket_name = 'test-storage-042' response_create_bucket = create_bucket(bucket_name = bucket_name, folder_id = folder_id, default_storage_class = 'STANDARD', max_size = '3221225472', anonymous_access_flags = {'read': False, 'list': False}, acl = {}, versioning = 'VERSIONING_DISABLED', tags = [])Однако крайне важно создать статический ключ доступа к бакету. Ключ позволяет читать данные с S3 как кластеру Data Proc, так и при помощи SDK boto3. Конечно же, у сервисного аккаунта service_account_id, указанного в функции, должна быть роль в каталоге с правами чтения и записи S3 бакета, в нашем случае это editor.
def create_static_key(service_account_id: str, description: str, iam_token: str = iam, ): url = 'https://iam.api.cloud.yandex.net/iam/aws-compatibility/v1/accessKeys' headers = {"Authorization": f"Bearer {iam_token}"} data = { "serviceAccountId": service_account_id, "description": description } response = requests.post(url, headers=headers, json=data) return response response_create_static_key = create_static_key( service_account_id = service_account_id, description = '') key = response_create_static_key.json()['accessKey']['keyId'] secret = response_create_static_key.json()['secret']Обязательно сохраняем идентификатор статического ключа и секретный ключ в защищенном месте. Мы передадим их через параметр запроса InputVariables метода execute при запуске ноутбуков.
Создаем функцию в Cloud Functions, которая будет запускать обучение:
def create_cloud_function(folder_id: str, function_name: str, description: str, iam_token: str = iam): url = 'https://serverless-functions.api.cloud.yandex.net/functions/v1/functions' headers = {"Authorization": f"Bearer {iam_token}"} data = { "folderId": folder_id, "name": function_name, "description": description, } response = requests.post(url, headers=headers, json=data) return response response_create_cloud_function = create_cloud_function(folder_id = folder_id, function_name = 'fit', description = 'start fit')Далее нужно собрать версию функции, в которую передадим сам код скрипта в виде строки. Чтобы получить эту строку, нужно заархивировать код функции в zip-архив и закодировать в base64. Строка передается в переменную content:
def create_cloud_function_version(function_id: str, resources: dict, runtime: str, entrypoint: str, log_options: dict, description: str, execution_timeout: str, service_account_id: str, content: str, iam_token: str = iam ): url = 'https://serverless-functions.api.cloud.yandex.net/functions/v1/versions' headers = {"Authorization": f"Bearer {iam_token}"} data = { "functionId": function_id, "runtime": runtime, "description": description, "entrypoint": entrypoint, "resources": resources, "executionTimeout": execution_timeout, "serviceAccountId": service_account_id, "logOptions": log_options, "content": content, } response = requests.post(url, headers=headers, json=data) return response response_create_cloud_function_version = create_cloud_function_version( function_id = fit_cf_id, resources = {'memory': '134217728'}, runtime = 'python312', entrypoint = 'index.handler', log_options = {'disabled': False, 'folderId': folder_id}, description = 'the function will execute fit notebook', execution_timeout = '1s', service_account_id = service_account_id, content = str(encoded)[2:-1], iam_token = iam )Функция и ее версия для запуска инференса добавляется полностью аналогичным образом.
Осталось создать триггеры, которые после появления новых данных в S3 запустят функции, которые в свою очередь запустят обучение, инференс или же отправку предсказаний в CleverData Join.
Мы будем отслеживать появление нового объекта на S3 с заданным префиксом, поэтому используем соответствующий event_type и передадим нужный prefix:
def create_trigger(folder_id: str, trigger_name: str, description: str, event_type: str, bucket_id: str, prefix: str, cloud_function_id: str, service_account_id: str, retry_attempts: str, interval: str, function_tag: str = '$latest', iam_token: str = iam ): url = 'https://serverless-triggers.api.cloud.yandex.net/triggers/v1/triggers' headers = {"Authorization": f"Bearer {iam_token}"} data = { "folderId": folder_id, "name": trigger_name, "description": description, "rule": { "objectStorage": { "eventType": [ event_type ], "bucketId": bucket_id, "prefix": prefix, "invokeFunction": { "functionId": cloud_function_id, "functionTag": function_tag, "serviceAccountId": service_account_id, "retrySettings": { "retryAttempts": retry_attempts, "interval": interval }, }, }, } } response = requests.post(url, headers=headers, json=data) return response response_create_fit_trigger = create_trigger( folder_id = folder_id, trigger_name = 'fit-test-trigger', description = '', event_type = 'OBJECT_STORAGE_EVENT_TYPE_CREATE_OBJECT', bucket_id = bucket_id, prefix = fitting_trigger_prefix, cloud_function_id = fit_cf_id, service_account_id = service_account_id, retry_attempts = '5', interval = '20s', function_tag = '$latest')Поговорим немного о возврате данных в CDP. В CleverData Join имеются готовые адаптеры на пакетную загрузку данных. При этом, все операции взаимодействия с CDP также превосходно автоматизируются через API.
Мы создаем экземпляр входящего адаптера для того, чтобы передавать и записывать в атрибут клиентского профиля CDP предсказания модели:
def create_adapter_instance_input(adapter_instance_id: str, source: str, cid: str, access_token: str, value: dict={}): headers = {"Authorization": "Bearer {}".format(access_token), "x-dmpkit-onbehalf-of": cid} url = f'https://domain_name.cleverdata.ru/integration-manager/v1/adapters/1/instances/{adapter_instance_id}/inputs' data = { "value":value, "source":source, "adapterInstanceId":adapter_instance_id } response = requests.post(url, json=data, headers=headers) return response response = create_adapter_instance_input(adapter_instance_id=adapter_instance_id, source='predictions_task.json', access_token=access_token, cid=cid)Все, что останется сделать, — реализовать в теле функции загрузку данных в файловый менеджер CleverData Join. Это тоже можно реализовать через API. Загруженные данные будут обработаны ранее созданным адаптером.
Обучение и дообучение модели
После запуска ноутбука обучения или инференса в проекте DataSphere происходит вторичная обработка с использованием кластера Dataproc. Новые сведения о событиях кликстрима добавляются в цепочки действий клиента. Далее на полученных цепочках происходит обучение модели.
Код вторичной обработки и код обучения во время первичного и повторного запуска ноутбуков немного отличается. Один и тот же код в зависимости от состояния объектов в бакете обрабатывает следующие несколько ситуаций. Рассмотрим их по отдельности.
В ситуации, когда в бакете находится только файл с событиями за текущий период, полученный в результате первичной обработки, мы имеем дело с первичным обучением модели. Следовательно, после обработки данных ноутбук выполнит последовательность первичного обучения модели. Для ноутбука инференса реализована та же логика. Если это первые данные для инференса в бакете, то модель вычислит предсказания для всех профилей.
В случае, если в бакете уже находились данные за предыдущие периоды, произойдет объединение строк и построение новых цепочек. Далее — дообучение модели.
Схема для повторного инференса выглядит схожим образом. Отличие заключается в том, что в данных за предыдущие периоды будут оставлены только события клиентов текущего периода. За счёт этого цепочки будут перестроены только для новых клиентов — именно тех, по которым значения предиктивных атрибутов ещё не сформированы.
В процессе дообучения создается новая модель-кандидат. Для того, чтобы определить, лучше ли эта модель, чем предыдущая, сравнивается AUC-ROC score новой модели на тестовой выборке с аналогичным показателем предыдущей модели, который сохраняется в бакете S3. При отсутствии увеличения метрики несколько итераций дообучения подряд — процесс дообучения заканчивается. Если же после дообучения модель показала лучшее качество, чем предыдущая лучшая модель, то происходит обновление лучшей модели (а также токенизатора, значения порога предсказания, файла с AUC-ROC score).
Ограничения в MVP версии проекта
В процессе реализации решения обнаружили некоторые ограничения, проработка которых может помочь расширить функциональность этого сервиса в будущем.
Например, в настоящее время отсутствует метод для загрузки ноутбуков в проект DataSphere. Поэтому при реализации мы изменили логику и ввели концепцию сервисного ноутбука, который однократно создается в DataSphere вручную. Это -разовая операция, нет необходимости выполнять ее для каждой новой задачи предсказания.
При заведении новой задачи выполнения целевого действия Model Manager размещает рабочие ноутбуки в S3 бакете. Сервисный ноутбук скачивает их в файловую систему DataSphere и последовательно запускает, если для задачи, соответствующей ноутбуку, есть новые первично обработанные данные.
Переход к данной концепции позволил ускорить и удешевить вторичную обработку. Теперь она сопровождается единоразовым поднятием кластера Data Proc и единой вторичной обработкой файлов всех задач клиента, под которые есть свежие данные.
Подводя итог, важно отметить, что в совокупности с концепцией сервисного ноутбука добавление метода загрузки ноутбуков в API позволит добиться еще более эффективной автоматизации развертывания пайплайна в Yandex Cloud, исключив необходимость в использовании UI.
Отключение режима запуска ноутбуков в Serverless в DataSphere
MVP изначально проектировалось в Serverless-парадигме — планировалось выделять ресурсы только в момент запуска вычислительных задач. Разумеется, мы с огромным удовольствием воспользовались Serverless-режимом запуска ноутбуков в DataSphere — это позволяло распределять рабочую нагрузку на CPU- и GPU-ресурсы в зависимости от характера нагрузки, хотя и приходилось взамен жертвовать временем на сериализацию, десериализацию переменных в процессе перехода между инстансами. Однако данный режим более не поддерживается, и, следовательно, нами был осуществлен переход к Dedicated.
Cтоит подчеркнуть, что переход от режима Serverless к Dedicated не повлиял на увеличение расходов на ресурсы вследствие простаивания инстансов, поскольку выделенная ВМ на запуск ноутбука через API после завершения выполнения инструкций моментально удаляется.
В рамках доработки текущего ограничения мы должны будем подобрать наиболее оптимальную конфигурацию ВМ при запуске в режиме Dedicated.
Ручная активация docker-образа в проекте
Пока что нет возможности взаимодействовать с ресурсом Docker-образ проекта DataSphere посредством API. Однако в ближайшем будущем этот функционал будет добавлен в API, что также обеспечит большую гибкость при развертывании пайплайна.
Для повышения функциональности ML-сервиса в CDP все перечисленные ограничения будут проработаны при разработке его продакшн-версии, в том числе некоторые из них будут усовершенствованы в сотрудничестве с Yandex Cloud.
Итоги MVP
Подводя итоги, можно с уверенностью сказать, что цели MVP были достигнуты.
-
Мы проверили гипотезу о возможности практически полностью автоматизированного развертывания инфраструктуры сервиса в облаке.
-
Построили эффективный конвейер предобработки данных.
-
Научились запускать код обучения и инференса по внешним триггерам.
-
Организовали процесс дообучения модели.
Использование ML-сервисов Yandex Cloud позволяет значительно упростить разработку решений по предиктивному анализу данных. Разработанный функционал является отличным дополнением к уже существующим широким возможностям платформы CleverData Join в части накопления знаний о едином профиле пользователя и построении его клиентского пути и позволяет обогащать данные о клиентской базе и открывает возможности использования предиктивных атрибутов для сегментации аудитории и организации адресных взаимодействий с клиентом.
ссылка на оригинал статьи https://habr.com/ru/articles/830446/
Добавить комментарий