Как я создаю ИИ для службы аналитики

В этой статье я хочу поделиться своим опытом разработки ИИ-сервиса для платформы веб-аналитики под названием Swetrix.

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

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

2. Требования и архитектура

Во время планирования было принято решение перейти к микросервисной архитектуре с брокером сообщений RabbitMQ для связи между сервисами AI и API.

Прежде всего, нам нужно собрать данные с помощью ежечасной задачи cron в отдельную базу данных. Мы решили выбрать ClickHouse, так как в нем хранятся исходные данные с сайтов на Swetrix. Подробности о формате будут рассмотрены в следующих разделах.

RabbitMQ был выбран в качестве брокера сообщений из-за его простоты, и нам необходимо установить связь между сервисами AI и API. Давайте разберем все и проверим основную логику

Служба Swetrix-API:

• Ежечасно собирает статистику данных с помощью задачи Cron и отправляет необработанные данные в службу искусственного интеллекта.
• Вставляет и получает предварительно обработанные данные из ClickHouse.

Служба Светрикс-АИ:

• Обрабатывает необработанные данные и выбранные предпочтения (интервал и подкатегория) для прогнозирования.
• Преобразует данные прогноза в формат JSON и отправляет их обратно в службу API через RabbitMQ.

Служба Swetrix-AI будет использовать платформу NestJs для серверной части и скрипты Python для предварительной обработки данных и прогнозирования моделей.

3. Предварительная обработка

Собираем следующие данные о проектах в analytics таблицу. Вы уже видели визуализированную версию этих данных в первом разделе статьи.

Мне удалось добиться этого (почти приемлемого) результата с помощью следующего запроса:

 @Cron(CronExpression.EVERY_HOUR) async insertHourlyProjectData(): Promise<void> { const gatherProjectsData = ` INSERT INTO analytics.hourly_projects_data (UniqueID, projectID, statisticsGathered, br_keys, br_vals, os_keys, os_vals, lc_keys, lc_vals, ref_keys, ref_vals, so_keys, so_vals, me_keys, me_vals, ca_keys, ca_vals, cc_keys, cc_vals, dv_keys, dv_vals, rg_keys, rg_vals, ct_keys, ct_vals) SELECT generateUUIDv4() as UniqueID, pid as projectID, toStartOfHour(now()) as statisticsGathered, groupArray(br) as br_keys, groupArray(br_count) as br_vals, groupArray(os) as os_keys, groupArray(os_count) as os_vals, ... groupArray(ct) as ct_keys, groupArray(ct_count) as ct_vals FROM ( SELECT pid, br, count(*) as br_count, os, count(*) as os_count, ... ct, count(*) as ct_count FROM analytics.analytics GROUP BY pid, br, os, lc, ref, so, me, ca, cc, dv, rg, ct ) GROUP BY pid; ` try { await clickhouse.query(gatherProjectsData).toPromise() } catch (e) { console.error( `[CRON WORKER] Error whilst gathering hourly data for all projects: ${e}`, )

Запуск функции запланирован каждый час с использованием задания Cron. Он собирает и вставляет аналитические данные в Clickhouse analytics.hourly_projects_data .

Выход

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

В частности, я использовал Python, чтобы сделать следующее:

3.1 Объединение ключей и значений

Объедините ключи и значения, относящиеся к одной категории, в одно поле JSON, например объединив ключи и значения устройств в один объект как таковой.

 os_keys = {“Windows”, ”MacOS”, ”MacOS”, ”MacOS”, ”Linux”} os_values = {1, 2, 2, 1, 5}

В:

 os = {“Windows”: 1, “MacOS”: 5, “Linux”: 5}

Прикрепляем код и вывод:

 def format_data(keys_list, vals_list, threshold): """ Format data by converting string representations of lists to actual lists, then sums up the counts for each key. Keys with counts below a specified threshold are aggregated into 'Other'. """ counts = defaultdict(int) for keys_str, vals_str in zip(keys_list, vals_list): keys = ast.literal_eval(keys_str) vals = ast.literal_eval(vals_str) for key, val in zip(keys, vals): counts[key] += val final_data = defaultdict(int) for value, count in counts.items(): final_data[value] = count return dict(final_data) def process_group(group): """ Combine specific groups by a group clause, and make a """ result = {} for col in group.columns: if col.endswith('_keys'): prefix = col.split('_')[0] # Extract prefix to identify the category (eg, 'br' for browsers) threshold = other_thresholds.get(prefix, 1) # Get the threshold for this category, default to 1 vals_col = col.replace('_keys', '_vals') keys_list = group[col].tolist() vals_list = group[vals_col].tolist() result[col.replace('_keys', '')] = format_data(keys_list, vals_list, threshold) return pd.Series(result)

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

Выход
3.2 Объединение ключей и значений

Чтобы обучить адекватную модель, я решил определить другие группы для разных категорий. Это означает, что если глобальное количество экземпляров группы в определенной категории ниже определенного процента (%), она будет добавлена как часть другой.

Например, в категории os у нас есть:

 {“MacOS”: 300, “Windows”: 400, “Linux”: 23 and “TempleOS”: 10}

Поскольку и Linux, и TempleOS в данном случае встречаются крайне редко, их объединят в другую группу , следовательно, конечный результат будет таким:

 {“MacOS”: 300, “Windows”: 400, “other”: 33}.

Причем «редкость» определяется по-разному в зависимости от категории и исходя из установленного для этой категории порога.

Его можно настроить на основе предпочтений и желаемых данных для клиента.

 other_thresholds = { 'br': 0.06, 'os': 0.04, 'cc': 0.02, 'lc': 0.02, 'ref': 0.02, 'so': 0.03, 'me': 0.03, 'ca': 0.03, 'cc': 0.02, 'dv': 0.02, 'rg': 0.01, 'ct': 0.01 }

Для достижения этой цели были реализованы 2 функции.

 def get_groups_by_treshholds(df,column_name): """Calculate total values for all columns""" if column_name in EXCLUDED_COLUMNS: return counter = count_dict_values(df[column_name]) total = sum(counter.values()) list1 = [] for key, value in counter.items(): if not (value / total) < other_thresholds[column_name]: list1.append(key) return list1 def create_group_columns(df): column_values = [] for key in other_thresholds.keys(): groups = get_groups_by_treshholds(df, key) if not groups: continue for group in groups: column_values.append(f"{key}_{group}") column_values.append(f"{key}_other") return column_values column_values = create_group_columns(df) column_values

Выход

 ['br_Chrome', 'br_Firefox', 'os_Mac OS', 'os_other', 'cc_UA', 'cc_GB', 'cc_other', 'dv_mobile', 'dv_desktop', 'dv_other']

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

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

Я создал функцию create_exploded_df где каждый объект представлен в виде отдельного столбца, а строки содержат соответствующие числовые значения. (Это еще не идеально, но это было лучшее решение, которое я смог создать)

 def create_exploded_df(df): """ Function which creates a new data set, iterates through the old one and fill in values according to their belongings (br_other, etc..) """ new_df = df[['projectID', 'statisticsGathered']] for group in column_values: new_df[group] = 0 new_df_cols = new_df.columns df_cols = df.columns for column in df_cols: if column in ['projectID', 'statisticsGathered']: continue for index, row in enumerate(df[column]): if column in EXCLUDED_COLUMNS: continue for key, value in row.items(): total = 0 if (a:=f"{column}_{key}") in new_df_cols: new_df[a][index] = value else: total += value new_df[f"{column}_other"][index] = total return new_df new_df = create_exploded_df(df) new_df.to_csv("2-weeks-exploded.csv") new_df

Выход

3.3 Заполните часы

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

Поэтому я написал скрипт, который будет находить пропущенные часы и вставлять пустые строки, если час пропущен.

3.4 Добавление и изменение целевых столбцов

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

 def sort_df_and_assign_targets(df): df = df.copy() df = df.sort_values(by=['projectID', 'statisticsGathered']) for column_name in df.columns: if not column_name.endswith('target'): continue df[column_name] = df.groupby('projectID')[column_name].shift(-1) return df new_df = sort_df_and_assign_targets(new_df)

Выход

3.5 Разделить statisticsGathered в отдельные столбцы.

Основная причина такого подхода заключается в том, что statisticsGathered была объектом datetime , модели, которые я пытался использовать (проверьте последующие разделы), не могли его обработать и определить правильный шаблон.

Это привело к ужасным показателям MSE/MRSE . Поэтому во время разработки было принято решение разделить функции на day , month и hour , что значительно улучшило результаты.

 def split_statistic_gathered(df): df['Month'] = df['statisticsGathered'].dt.month.astype(int) # as int df['Day'] = df['statisticsGathered'].dt.day.astype(int) # as int df['Hour'] = df['statisticsGathered'].dt.hour df = df.drop('statisticsGathered', axis = 1) return df new_df = split_statistic_gathered(new_df) new_df

Выход

Вот и все! Перейдем к самому обучению! 🎉🎉🎉

4. Линейная регрессия

Что ж, я думаю, самое сложное при создании этого приложения было собственно прогнозирование.

Первое, что я хотел попробовать, это использовать модель LinearRegression :

Я реализовал следующие функции:

 def create_model_for_target(train_df, target_series):    X_train, x_test, Y_train, y_test = train_test_split(train_df, target_series, test_size=0.3, shuffle=False)    reg = LinearRegression()    reg.fit(X_train, Y_train)    y_pred = reg.predict(x_test)    return {"y_test": y_test, "y_pred": y_pred} def create_models_for_targets(df):    models_data = dict()    df = df.dropna()    train_df = clear_df(df)    for target_name in df[[column_name for column_name in df.columns if column_name.endswith("target")]]:        models_data[target_name] = create_model_for_target(train_df, df[target_name])    return models_data

Объяснение

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

Чтобы оценить правильность результатов, я добавил функцию, которая собирает необходимые показатели и выдает выходные данные.

 def evaluate_models(data):    evaluation = []    for target, results in data.items():        y_test, y_pred = results['y_test'], results['y_pred']        mse = mean_squared_error(y_test, y_pred)        rmse = mean_squared_error(y_test, y_pred) ** 0.5        mae = mean_absolute_error(y_test, y_pred)        mean_y = y_test.mean()        median_y = y_test.median()        evaluation.append({'target': target, 'mse': mse, 'rmse': rmse, 'mae': mae, 'mean_y': mean_y, 'median_y': median_y})    return pd.DataFrame(evaluation)

Выход

Я написал сценарий, который генерировал выходные данные и сохранял их в файл Excel, учитывая значения mse , rmse , mae mean_y .

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

Поэтому я принял решение спрогнозировать общее количество посетителей в час, чтобы были созданы следующие функции:

 def add_target_column(df, by):  totals_series = df.apply(lambda x: sum(x[[column for column in df.columns if column.startswith(by)]]), axis=1)  df['total'] = totals_series  df[f'total_{by}_target'] = totals_series  return df def shift_target_column(df, by):  df = df.sort_values(by=['projectID', 'statisticsGathered'], ignore_index=True)  df['total_target'] = df.groupby('projectID')[f'total_{by}_target'].shift(-1)  return df new_df = add_target_column(new_df, 'br') new_df = shift_target_column(new_df, 'br') new_df[['total_br_target']]

Выход

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

При таком подходе модель показала в 10 раз лучшие результаты, чем была раньше .

5. Вывод

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

Прогнозы отклоняются от фактических значений примерно на 2,45 посетителей (поскольку RMSE = √MSE ) . Что не может иметь никакого негативного решающего влияния на маркетинговые нужды.

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

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

И удачи!