Оптимізація обробки даних ETL на Talent.com за допомогою Amazon SageMaker

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

читають: 0

Співавторами цієї публікації є Анатолій Хоменко, інженер машинного навчання, та Абденур Беззу, головний технічний директор Talent.com.

Заснована в 2011, Talent.com збирає оплачувані списки вакансій від своїх клієнтів і загальнодоступні списки вакансій, а також створив уніфіковану платформу для легкого пошуку. Охоплюючи понад 30 мільйонів списків вакансій у більш ніж 75 країнах і охоплюючи різні мови, галузі та канали розповсюдження, Talent.com задовольняє різноманітні потреби шукачів роботи, ефективно з’єднуючи мільйони шукачів з можливостями роботи.

Місія Talent.com полягає в тому, щоб сприяти глобальним зв’язкам робочої сили. Щоб досягти цього, Talent.com збирає списки вакансій з різних джерел в Інтернеті, пропонуючи шукачам доступ до широкого пулу з понад 30 мільйонів вакансій, адаптованих до їхніх навичок і досвіду. Відповідно до цієї місії Talent.com співпрацював з AWS, щоб розробити передову систему рекомендацій щодо роботи на основі глибокого навчання, спрямовану на допомогу користувачам у просуванні їх кар’єри.

Щоб забезпечити ефективну роботу цього механізму рекомендацій щодо вакансій, надзвичайно важливо запровадити широкомасштабний конвеєр обробки даних, відповідальний за вилучення й уточнення функцій зі зведених списків вакансій Talent.com. Цей конвеєр здатний обробляти 5 мільйонів щоденних записів менш ніж за 1 годину та дозволяє одночасно обробляти записи за кілька днів. Крім того, це рішення забезпечує швидке розгортання у виробництві. Основним джерелом даних для цього конвеєра є формат рядків JSON, який зберігається в Служба простого зберігання Amazon (Amazon S3) і розділені за датою. Щодня це призводить до генерації десятків тисяч файлів JSON Lines із поступовими оновленнями, що відбуваються щодня.

Основна мета цього конвеєра обробки даних – сприяти створенню функцій, необхідних для навчання та розгортання системи рекомендацій щодо роботи на Talent.com. Варто зазначити, що цей конвеєр має підтримувати поступові оновлення та задовольняти складні вимоги щодо вилучення функцій, необхідних для модулів навчання та розгортання, необхідних для системи рекомендацій щодо роботи. Наш конвеєр належить до загального сімейства процесів ETL (видобування, перетворення та завантаження), які об’єднують дані з багатьох джерел у велике центральне сховище.

Щоб отримати додаткові відомості про те, як Talent.com і AWS спільно створили передову обробку природної мови та навчальні методи моделі глибокого навчання, використовуючи Amazon SageMaker щоб створити систему рекомендацій щодо роботи, див Від текстових повідомлень до роботи мрії: створюйте на Talent.com рекомендацію про роботу на основі НЛП за допомогою Amazon SageMaker. Система включає розробку функцій, розробку архітектури моделі глибокого навчання, оптимізацію гіперпараметрів і оцінку моделі, де всі модулі запускаються за допомогою Python.

У цій публікації показано, як ми використовували SageMaker для створення широкомасштабного конвеєра обробки даних для підготовки функцій для механізму рекомендацій щодо роботи на Talent.com. Отримане рішення дає змогу досліднику даних створювати ідеї щодо вилучення функцій у блокноті SageMaker за допомогою бібліотек Python, таких як scikit-learn or PyTorch, а потім швидко розгорнути той самий код у конвеєрі обробки даних, виконуючи вилучення функцій у масштабі. Рішення не вимагає перенесення коду вилучення функцій для використання PySpark, як це вимагається під час використання Клей AWS як рішення ETL. Наше рішення може бути розроблено та розгорнуто виключно Data Scientist наскрізно, використовуючи лише SageMaker, і не потребує знання інших рішень ETL, таких як Пакет AWS. Це може значно скоротити час, необхідний для розгортання конвеєра машинного навчання (ML) у виробництві. Конвеєр керується через Python і бездоганно інтегрується з робочими процесами вилучення функцій, що робить його адаптованим до широкого спектру програм аналітики даних.

Огляд рішення

Огляд конвеєра ETL за допомогою обробки SageMaker

Трубопровід складається з трьох основних фаз:

Використовуйте an Обробка Amazon SageMaker завдання для обробки необроблених файлів JSONL, пов’язаних із вказаним днем. Кілька днів даних можна обробляти окремими завданнями обробки одночасно.
Працівник Клей AWS для сканування даних після обробки даних за кілька днів.
Завантажте оброблені функції для вказаного діапазону дат за допомогою SQL із Амазонка Афіна таблиці, а потім навчіть і розгорніть модель рекомендованої роботи.

Обробляти необроблені файли JSONL

Ми обробляємо необроблені файли JSONL за визначений день за допомогою завдання обробки SageMaker. Завдання реалізує вилучення функцій і стиснення даних, а також зберігає оброблені функції у файли Parquet з 1 мільйоном записів на файл. Ми використовуємо переваги розпаралелювання ЦП, щоб паралельно виконувати вилучення функцій для кожного необробленого файлу JSONL. Результати обробки кожного файлу JSONL зберігаються в окремому файлі Parquet у тимчасовому каталозі. Після обробки всіх файлів JSONL ми виконуємо стиснення тисяч маленьких файлів Parquet у кілька файлів з 1 мільйоном записів на файл. Ущільнені файли Parquet потім завантажуються в Amazon S3 як результат завдання обробки. Ущільнення даних забезпечує ефективне сканування та SQL-запити на наступних етапах конвеєра.

Нижче наведено приклад коду для планування завдання обробки SageMaker на певний день, наприклад 2020-01-01, за допомогою SDK SageMaker. Завдання читає необроблені файли JSONL з Amazon S3 (наприклад, з s3://bucket/raw-data/2020/01/01) і зберігає стислі файли Parquet в Amazon S3 (наприклад, у s3://bucket/processed/table-name/day_partition=2020-01-01/).

### install dependencies %pip install sagemaker pyarrow s3fs awswrangler import sagemaker
import boto3 from sagemaker.processing import FrameworkProcessor
from sagemaker.sklearn.estimator import SKLearn
from sagemaker import get_execution_role
from sagemaker.processing import ProcessingInput, ProcessingOutput region = boto3.session.Session().region_name
role = get_execution_role()
bucket = sagemaker.Session().default_bucket() ### we use instance with 16 CPUs and 128 GiB memory
### note that the script will NOT load the entire data into memory during compaction
### depending on the size of individual jsonl files, larger instance may be needed
instance = "ml.r5.4xlarge"
n_jobs = 8 ### we use 8 process workers
date = "2020-01-01" ### process data for one day est_cls = SKLearn
framework_version_str = "0.20.0" ### schedule processing job
script_processor = FrameworkProcessor( role=role, instance_count=1, instance_type=instance, estimator_cls=est_cls, framework_version=framework_version_str, volume_size_in_gb=500,
) script_processor.run( code="processing_script.py", ### name of the main processing script source_dir="../src/etl/", ### location of source code directory ### our processing script loads raw jsonl files directly from S3 ### this avoids long start-up times of the processing jobs, ### since raw data does not need to be copied into instance inputs=[], ### processing job input is empty outputs=[ ProcessingOutput(destination="s3://bucket/processed/table-name/", source="/opt/ml/processing/output"), ], arguments=[ ### directory with job's output "--output", "/opt/ml/processing/output", ### temporary directory inside instance "--tmp_output", "/opt/ml/tmp_output", "--n_jobs", str(n_jobs), ### number of process workers "--date", date, ### date to process ### location with raw jsonl files in S3 "--path", "s3://bucket/raw-data/", ], wait=False
)

Наведена нижче схема коду для основного сценарію (processing_script.py), який запускає завдання обробки SageMaker, виглядає так:

import concurrent
import pyarrow.dataset as ds
import os
import s3fs
from pathlib import Path ### function to process raw jsonl file and save extracted features into parquet file from process_data import process_jsonl ### parse command line arguments
args = parse_args() ### we use s3fs to crawl S3 input path for raw jsonl files
fs = s3fs.S3FileSystem()
### we assume raw jsonl files are stored in S3 directories partitioned by date
### for example: s3://bucket/raw-data/2020/01/01/
jsons = fs.find(os.path.join(args.path, *args.date.split('-'))) ### temporary directory location inside the Processing job instance
tmp_out = os.path.join(args.tmp_output, f"day_partition={args.date}") ### directory location with job's output
out_dir = os.path.join(args.output, f"day_partition={args.date}") ### process individual jsonl files in parallel using n_jobs process workers
futures=[]
with concurrent.futures.ProcessPoolExecutor(max_workers=args.n_jobs) as executor: for file in jsons: inp_file = Path(file) out_file = os.path.join(tmp_out, inp_file.stem + ".snappy.parquet") ### process_jsonl function reads raw jsonl file from S3 location (inp_file) ### and saves result into parquet file (out_file) inside temporary directory futures.append(executor.submit(process_jsonl, file, out_file)) ### wait until all jsonl files are processed for future in concurrent.futures.as_completed(futures): result = future.result() ### compact parquet files
dataset = ds.dataset(tmp_out) if len(dataset.schema) > 0: ### save compacted parquet files with 1MM records per file ds.write_dataset(dataset, out_dir, format="parquet", max_rows_per_file=1024 * 1024)

Масштабованість є ключовою особливістю нашого трубопроводу. По-перше, кілька завдань SageMaker Processing можна використовувати для обробки даних протягом кількох днів одночасно. По-друге, ми уникаємо одночасного завантаження всіх оброблених або необроблених даних у пам’ять, обробляючи дані кожного визначеного дня. Це дозволяє обробляти дані за допомогою типів екземплярів, які не можуть вмістити дані за цілий день у основній пам’яті. Єдина вимога полягає в тому, щоб тип екземпляра міг одночасно завантажувати в пам’ять N необроблених файлів JSONL або оброблених Parquet, причому N означає кількість робочих процесів, що використовуються.

Сканувати оброблені дані за допомогою AWS Glue

Після обробки всіх необроблених даних за кілька днів ми можемо створити таблицю Athena з усього набору даних за допомогою сканера AWS Glue. Ми використовуємо AWS SDK для панд (awswrangler) бібліотеку для створення таблиці за допомогою наступного фрагмента:

import awswrangler as wr ### crawl processed data in S3
res = wr.s3.store_parquet_metadata( path='s3://bucket/processed/table-name/', database="database_name", table="table_name", dataset=True, mode="overwrite", sampling=1.0, path_suffix='.parquet',
) ### print table schema
print(res[0])

Завантажте оброблені функції для навчання

Оброблені функції для визначеного діапазону дат тепер можна завантажити з таблиці Athena за допомогою SQL, а потім ці функції можна використовувати для навчання моделі рекомендацій для роботи. Наприклад, наступний фрагмент завантажує оброблені функції за один місяць у DataFrame за допомогою awswrangler бібліотека:

import awswrangler as wr query = """ SELECT * FROM table_name WHERE day_partition BETWEN '2020-01-01' AND '2020-02-01' """ ### load 1 month of data from database_name.table_name into a DataFrame
df = wr.athena.read_sql_query(query, database='database_name')

Крім того, використання SQL для завантаження оброблених функцій для навчання можна розширити, щоб відповідати іншим варіантам використання. Наприклад, ми можемо застосувати подібний конвеєр для підтримки двох окремих таблиць Athena: однієї для зберігання показів користувачів, а іншої для зберігання кліків користувачів за цими показами. Використовуючи оператори з’єднання SQL, ми можемо отримати покази, на які користувачі клацнули або не клацнули, а потім передати ці покази в завдання навчання моделі.

Переваги рішення

Впровадження запропонованого рішення дає декілька переваг нашому існуючому робочому процесу, зокрема:

Спрощена реалізація – Рішення дозволяє реалізовувати вилучення функцій у Python за допомогою популярних бібліотек ML. І для цього не потрібно переносити код у PySpark. Це спрощує вилучення функцій, оскільки той самий код, розроблений Data Scientist у блокноті, виконуватиметься цим конвеєром.
Швидкий шлях до виробництва – Рішення може бути розроблено та розгорнуто спеціалістом із обробки даних для масштабного вилучення функцій, що дає їм змогу розробити рекомендаційну модель ML на основі цих даних. У той же час, те саме рішення може бути розгорнуто у виробництві інженером ML з невеликими змінами.
Багаторазовість – Рішення надає багаторазовий шаблон для вилучення функцій у масштабі та може бути легко адаптований для інших випадків використання, окрім створення рекомендованих моделей.
Ефективність – Рішення забезпечує хорошу продуктивність: обробка одного дня Talent.comдані зайняли менше 1 години.
Поступове оновлення – Рішення також підтримує поступове оновлення. Нові щоденні дані можна обробляти за допомогою завдання SageMaker Processing, а розташування S3, що містить оброблені дані, можна повторно сканувати, щоб оновити таблицю Athena. Ми також можемо використовувати завдання cron для оновлення поточних даних кілька разів на день (наприклад, кожні 3 години).

Ми використали цей конвеєр ETL, щоб допомогти Talent.com обробляти 50,000 5 файлів на день, що містять 90 мільйонів записів, і створили навчальні дані, використовуючи функції, отримані з необроблених даних за 450 днів із Talent.com — загалом 900,000 мільйонів записів у 2 8.6 файлів. Наші розробки допомогли Talent.com створити та розгорнути систему рекомендацій лише за XNUMX тижні. Рішення виконувало всі процеси ML, включаючи ETL, на Amazon SageMaker без використання інших служб AWS. Завдяки системі рекомендацій щодо вакансій рейтинг кліків під час онлайн-тестування A/B збільшився на XNUMX% у порівнянні з попереднім рішенням на основі XGBoost, що допомогло мільйонам користувачів Talent.com отримати кращу роботу.

Висновок

У цьому дописі описано конвеєр ETL, який ми розробили для обробки функцій для навчання та розгортання моделі рекомендованих вакансій на Talent.com. Наш конвеєр використовує завдання SageMaker Processing для ефективної обробки даних і вилучення функцій у великих масштабах. Код вилучення функцій реалізовано на Python, що дозволяє використовувати популярні бібліотеки ML для виконання вилучення функцій у масштабі без необхідності переносити код для використання PySpark.

Ми заохочуємо читачів вивчити можливість використання конвеєра, представленого в цьому блозі, як шаблону для своїх випадків використання, де потрібне вилучення функцій у масштабі. Спеціаліст з обробки даних може використати цей конвеєр для створення моделі ML, а потім інженер ML може використати цей конвеєр для запуску у виробництво. Це може значно скоротити час, необхідний для створення наскрізного рішення МЛ, як це було у випадку з Talent.com. Читачі можуть звернутися до посібник із налаштування та запуску завдань обробки SageMaker. Також пропонуємо читачам переглянути публікацію Від текстових повідомлень до роботи мрії: створюйте на Talent.com рекомендацію про роботу на основі НЛП за допомогою Amazon SageMaker, де ми обговорюємо використання навчальних методів моделі глибокого навчання Amazon SageMaker створити систему рекомендацій щодо роботи Talent.com.

Про авторів

Дмитро Беспалов є старшим прикладним науковим співробітником Amazon Machine Learning Solutions Lab, де він допомагає клієнтам AWS у різних галузях прискорити впровадження ШІ та хмарних технологій.

І Сян є Applied Scientist II в Amazon Machine Learning Solutions Lab, де вона допомагає клієнтам AWS у різних галузях прискорити впровадження ШІ та хмарних технологій.

Тонг Ван є старшим прикладним науковим співробітником Amazon Machine Learning Solutions Lab, де він допомагає клієнтам AWS у різних галузях прискорити впровадження ШІ та хмарних технологій.

Анатолій Хоменко є старшим інженером машинного навчання в Talent.com з пристрастю до обробки природної мови, підбираючи хороших людей для хорошої роботи.

Абденур Беззух є керівником із понад 25-річним досвідом створення та надання технологічних рішень, які масштабуються для мільйонів клієнтів. Абденур займав посаду головного технічного директора (CTO) у Talent.com коли команда AWS розробила та виконала це рішення для Talent.com.

Яньцзюнь Ці є старшим менеджером із прикладних наук у Amazon Machine Learning Solution Lab. Вона впроваджує інновації та застосовує машинне навчання, щоб допомогти клієнтам AWS прискорити впровадження ШІ та хмари.