Pandas concat и merge: объединение двух батчей данных

Интеграция двух батчей: concat → дедупликация → агрегация → merge → финальная схема

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

Что уже дано

Проверяющая система вызывает integrate_batches(batch1, batch2, dim_table), передавая три готовых pd.DataFrame:

• batch1 и batch2 — два батча событий одинаковой схемы: 'event_id' (int), 'source' (str), 'amount' (float). Оба батча могут содержать пересекающиеся бизнес-ключи ('event_id', 'source'): более поздняя версия события находится в batch2.
• dim_table — таблица-измерение со столбцами 'source' (str) и 'channel' (str). Уникальна по 'source'.

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

Что нужно сделать

Реализуйте функцию integrate_batches(batch1, batch2, dim_table), выполняющую ровно пять шагов.

Шаг 1. Склейка батчей.
Объедините batch1 и batch2 через pd.concat([batch1, batch2], ignore_index=True). Сохраните в combined.

Шаг 2. Дедупликация по бизнес-ключу.
Удалите дубли по составному ключу ['event_id', 'source'], оставив последнее вхождение (keep='last') — оно соответствует более актуальной записи из batch2. Сбросьте индекс через reset_index(drop=True). Сохраните в deduped.

Шаг 3. Агрегация.
Сгруппируйте deduped по 'source' с as_index=False и вычислите за один вызов .agg():

• event_count=('event_id', 'count') — число уникальных событий по источнику;
• total_amount=('amount', 'sum') — суммарная сумма.

Сохраните в agg.

Шаг 4. Merge с validate.
Соедините agg и dim_table через pd.merge с параметрами:

• on='source'
• how='left'
• validate='many_to_one'

Сохраните в result.

Шаг 5. Финальная спецификация.
Оставьте в result ровно эти столбцы в указанном порядке:

['source', 'channel', 'event_count', 'total_amount']

Отсортируйте по 'total_amount' по убыванию и сбросьте индекс через reset_index(drop=True).

Верните итоговый result.

Ввод и вывод

stdin не используется. Печатать через print в stdout ничего не нужно.

Проверяющая система вызывает integrate_batches(batch1, batch2, dim_table) и проверяет возвращаемое значение. Функция принимает три аргумента — готовые pd.DataFrame. Функция должна вернуть pd.DataFrame с четырьмя столбцами в порядке спецификации, отсортированный по 'total_amount' по убыванию, с чистым RangeIndex.

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

import pandas as pd

batch1 = pd.DataFrame({
 'event_id': [1, 2, 3],
 'source': ['web', 'app', 'web'],
 'amount': [100.0, 200.0, 50.0],
})
batch2 = pd.DataFrame({
 'event_id': [2, 4],
 'source': ['app', 'app'],
 'amount': [250.0, 300.0],
})
dim_table = pd.DataFrame({
 'source': ['web', 'app'],
 'channel': ['organic', 'paid'],
})
result = integrate_batches(batch1, batch2, dim_table)
print(list(result.columns))
print(result[['source', 'total_amount']].values.tolist())

Результат:

['source', 'channel', 'event_count', 'total_amount']
[['app', 550.0], ['web', 150.0]]