В этой статье мы покажем, как развернуть open-source сервис Airbyte на вашем компьютере и с его помощью подключиться к Facebook Marketing API.

Что такое Airbyte?

Airbyte — это платформа интеграции данных с открытым исходным кодом для создания ELT пайплайнов, которая помогает вам реплицировать данные в ваших хранилищах, озерах и базах данных. Кроме того, она имеет удобный интерфейс, который очень красиво оформлен (вы только посмотрите на эту осьминожку)!

Разворачиваем Airbyte

Установка Airbyte на машину выполняется достаточно просто. В этом можно убедиться, заглянув в официальную документацию. Для этого выполнить несколько шагов:

1. Установить Docker (см. Инструкции) и убедиться, что у вас стоит последняя версия docker-compose.
2. Выполнить следующие команды в терминале:

• git clone
• cd airbyte
• docker-compose up

Как только в выводе терминала появится баннер Airbyte, UI станет доступен по адресу http://localhost:8000.

Подключаем Facebook

С полной документацией по работе с Facebook Marketing из Airbyte можно ознакомиться по ссылке. Мы же дадим выжимку из нее. Итак, для того, чтобы подключить Facebook в Airbyte, надо учесть следующие условия:

• Наличие Facebook Ad Account ID (где его найти?)
• Наличие Facebook App с включенным Marketing API (подробнее)
• Наличие Facebook Marketing API Access Token (подробнее)
• Запросить увеличение лимита на количество запросов: Facebook сильно ограничивает токены API, сгенерированные из приложений Facebook с уровнем «Стандартный доступ» (по умолчанию для новых приложений именно такой уровень доступа), что делает невозможным использование токена для синхронизации с Airbyte. Вам нужно будет запросить апгрейд до расширенного доступа для вашего приложения со следующими разрешениями (подробнее):
  • Ads Management Standard Access
  • ads_read
  • Ads_management
• Перед этим вам нужно будет пройти процесс проверки вашей компании (подробнее).

После этого можно подключить Facebook в Airbyte.

1. Заходим в UI на вкладку Sources и нажимаем ‘+ new source’

2. Выбираем в качестве источника Facebook Marketing и прописываем конфигурации подключения:
   • Название источника, под которым он будет отображаться в Airbyte
   • Facebook Ad Account ID
   • Дата начала — с этого момента будут грузиться данные из Facebook (⚠️ВАЖНО: Данные по инсайтам можно получить только за последние 37 месяцев)
   • Дата окончания — до этого момента будут грузиться данные из Facebook (можно оставить пустым, чтобы получать последние данные из аккаунта)
   • Facebook Marketing API Access Token

3. Также имеются несколько дополнительных конфигурационных опций:
   • Включение удаленных данных Campaigns, Ads и AdSets
   • Сохранение миниатюр изображений
   • Настройка кастомных Insignts

4. После заполнения полей нажимаем ‘Set up source’ и ждем, пока процесс подключения не завершится

5. При необходимости настройки подключения к источнику можно поменять на его вкладке ‘Settings’

Может возникнуть ошибка в связи с тем, что версия фреймворка устарела. Тогда надо будет перейти в основную вкладку ‘Settings’ (шестеренка в левом нижнем углу), там выбрать ‘Sources’, в списке найти Facebook Marketing и нажать кнопку ‘Change’ (или нажать ‘Upgrade all’, чтобы загрузить все доступные обновления).

Подключаем хранилище данных

После того, как мы подключили Facebook Marketing, нам необходимо связать этот источник с нашим хранилищем данных.

1. Нажимаем ‘add destination’ на вкладке с настроенным источником (см. пункт 4 предыдущего раздела) либо идем на вкладку ‘Destinations’ и там нажимаем ‘+ new destination’.

2. Далее выбираем тип хранилища, с которым будем работать. Airbyte может подключаться к большому количеству популярных СУБД. Опишем необходимые шаги конфигурации на примере работы с PostgreSQL. Заполняем следующие поля:
   • Название хранилища, под которым оно будет отображаться в Airbyte
   • Хост
   • Порт
   • База данных
   • Схема
   • Пользователь
   • Пароль
   • Нужно ли использовать SSL
   • Параметры SSH туннеля (опционально)

3. После заполнения полей нажимаем ‘Set up destination’ и ждем, пока процесс подключения не завершится
4. Если вы выполняли подключение к хранилищу отдельно с вкладки ‘Destinations’, то надо связать с ним источник, нажав на ‘add source’

5. При необходимости настройки подключения к хранилищу можно поменять на его вкладке ‘Settings’
6. Версию образа для работы с хранилищем можно обновить при необходимости (по аналогии с тем, что мы писали в конце предыдущего раздела)

Настраиваем связь источника и  хранилища данных

После всех проделанных шагов у нас открывается вкладка ‘Set up connection’, на которой мы можем настроить то, как и какие данные мы будем грузить с помощью Airbyte:

• Как часто будет происходить загрузка
• Куда именно мы будем помещать данные в хранилище
• Какие данные из источника мы будем брать и каким образом (Full refresh | Increment, Overwrite | Append)
• Надо ли нормализовывать данные (сырые данные все равно останутся в хранилище)
• Настройка кастомных трансформаций (опционально)

После заполнения полей, нажимаем ‘Set up connection’. Далее откроется страница настроенного подключения, где мы можем наблюдать историю выгрузок данных, а также вносить изменения в установленную связь Источник-Хранилище.

Если нажать на Sync, то будут показаны логи исполнения.

Результаты

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

Сразу можно обратить внимание на количество сгенерированных таблиц.

Так как мы указали, что нам необходима нормализация данных, то у нас в результате получилось чуть меньше 400 таблиц для streams [activities, ‘ad_account’, ‘ad_creatives’, ‘ad_sets’, ‘ads’, ‘ads_insights’, ‘campaigns’], причем многие из них оказались пустыми (так сгенерировал Airbyte).

Сырые данные содержатся в таблицах с именами ‘_airbyte_raw_*’. В таких таблицах имеется всего 3 поля — ‘_airbyte_ab_id’, ‘_airbyte_data’ и ‘_airbyte_emmited_at’. Столбец ‘_airbyte_data’ имеет тип jsonb.
Пример для ‘_airbyte_raw_ads_insights’:

В процессе нормализации каждое поле из json выносится в отдельную колонку. Пример: часть списка колонок из таблицы ‘ads_insignts’:

После получения основной нормализованной таблицы для каждого stream создаются различные breakdowns, например, для тех же ‘ads_insights’:

Давайте узнаем, как получить информацию о затратах, кликах и показах рекламных кампаний из Facebook.

from facebook_business.api import FacebookAdsApi
from facebook_business.exceptions import FacebookRequestError
from facebook_business.adobjects.adaccount import AdAccount
from facebook_business.adobjects.adreportrun import AdReportRun
from facebook_business.adobjects.adsinsights import AdsInsights
from facebook_business.adobjects.campaign import Campaign
from facebook_business.adobjects.adset import AdSet
from facebook_business.adobjects.adaccountuser import AdAccountUser as AdUser
from facebook_business import adobjects
from matplotlib import pyplot as plt
from pandas import DataFrame
import time

Получение ключа

Первое, что необходимо сделать для работы с Facebook API — создать приложение. Для этого заходим на https://developers.facebook.com — My Apps — Create App. Заполняем название приложения, контактный Email и жмём на «Create App ID»

Перед нами открывается Dashboard приложения. Переходим в Settings — Basic и записываем себе App ID и App Secret. Эти данные понадобятся для авторизации.

Теперь переходим в Tools — Graph API Explorer и оказываемся в меню для создания нового access token.

Токены выдаются под разные нужды, и нам нужно задать определенные права для нашего. Понадобится только на ads_managment — это право позволяет получать информацию о рекламных кампаниях вашего аккаунта Facebook. Добавляем его и нажимаем на «Generate Access Token».

Осторожно — user access token изначально выдается как short-lived, уже через 1-2 часа он перестанет действовать. Для получения long-lived Token можно нажать на синюю кнопку, чтобы открыть Access Token Info, а затем на «Open in Access Token Tool». Откроется новая страница, внизу которой появится синяя кнопка «Extend Access Token». После этого сгенерируется новый long-lived Access Token, действующий 60 дней.

Написание скрипта

Создадим три переменных, в которые запишем access token, App ID и App Secret. Авторизуемся через метод init() класса FacebooksAdsApi и добавляем пользователя. Метод get_ad_accounts() вернёт нам данные по всем нашим рекламным аккаунтам в виде словаря. По этим же данным можем получить информацию о кампаниях методом get_campaigns().

my_access_token = ''
my_app_id = ''
my_app_secret = ''
FacebookAdsApi.init(my_app_id, my_app_secret, my_access_token)

me = AdUser(fbid='me')
my_accounts = list(me.get_ad_accounts())
my_accounts

my_account = my_accounts[0]
campaigns = my_account.get_campaigns()
print(campaigns)

Попробуем получить amount spent через my_account. Для этого воспользуемся методом api_get(), передав в параметр fields поле AdAccount.Field.amount_spent. Теперь, чтобы получить желаемые данные, выведем поле у переменной my_account, поделив на 100, чтобы обрубить копейки. Расходы получаем в валюте аккаунта, в нашем случае это RUB. То, ради чего мы всё это затеваем — получить данные о расходах на рекламные кампании для последующего анализа.

my_account.api_get(fields=[AdAccount.Field.amount_spent])
print(int(my_account[AdAccount.Field.amount_spent])/100)

Объявим переменную fields — в этом списке будут храниться поля, которые мы хотим получать: id кампании, количество кликов, затрат и просмотров. Теперь опишем две функции. Первая будет асинхронно отправлять запросы к Facebook и возвращать результаты. Вторая — формирует эти запросы и передает в первую функцию. В результате будем получать список словарей.

fields = [
    AdsInsights.Field.campaign_id,
    AdsInsights.Field.clicks,
    AdsInsights.Field.spend,
    AdsInsights.Field.impressions]

count = 0

def wait_for_async_job(async_job):
    global count
    async_job = async_job.api_get()
    while async_job[AdReportRun.Field.async_status] != 'Job Completed' or async_job[
        AdReportRun.Field.async_percent_completion] < 100:
        time.sleep(2)
        async_job = async_job.api_get()
    else:
        print("Job " + str(count) + " completed")
        count += 1
    return async_job.get_result(params={"limit": 1000})

def get_insights(account, date_preset='last_3d'):
    account = AdAccount(account["id"])
    i_async_job = account.get_insights(
        params={
            'level': 'ad',
            'date_preset': date_preset,
            'time_increment': 1},
            fields=fields,
            is_async=True)
    results = [dict(item) for item in wait_for_async_job(i_async_job)]
    return results

Следующий шаг — получение искомых данных о затратах. Будем собирать данные за всё время, поэтому заведём переменную date_preset, значение которой поставим lifetime. И для каждого аккаунта вызовем функцию get_insights(), а список, который она возвращает, положим в insights_lists.
Создадим DataFrame и вытащим из insights_lists интересующие данные — это id кампании, количество кликов, затраты и просмотры.

elem_insights = []
insights_lists = []
date_preset = 'last_year'
for elem in my_accounts:
            elem_insights = get_insights(elem, date_preset)
            insights_lists.append(elem_insights)

insight_campaign_id_list = []
insight_clicks_list = []
insight_spend_list = []
insight_impressions_list = []
insight_date_start_list = []
insight_date_stop_list = []
for elem1 in insights_lists:
    for elem2 in elem1:
        insight_campaign_id_list.append(int(elem2['campaign_id']))
        insight_clicks_list.append(int(elem2['clicks']))
        insight_spend_list.append(float(elem2['spend']))
        insight_impressions_list.append(int(elem2['impressions']))
        insight_date_start_list.append(elem2['date_start'])
        insight_date_stop_list.append(elem2['date_stop'])

df = DataFrame()
df['campaign_id'] = insight_campaign_id_list
df['clicks'] = insight_clicks_list
df['spend'] = insight_spend_list
df['impressions'] = insight_impressions_list
df['date_start'] = insight_date_start_list
df['date_stop'] = insight_date_stop_list

Получим такой DataFrame:

Давайте подытожим эти данные — сгруппируем по кампаниям и посчитаем суммы каждой колонки для групп. Для этого в pandas реализованы методы groupby() и sum() — достаточно указать, по какой колонке проводить группировку.

df.groupby(['campaign_id']).sum()

Теперь построим два графика — по количеству кликов и просмотров относительно дат. Воспользуемся атрибутом rcParams, чтобы задать размеры графиков.

plt.rcParams['figure.figsize'] = [20, 5]
plt.plot(df.date_start.str.replace('2019-', ''), df.clicks)

plt.rcParams['figure.figsize'] = [20, 5]
plt.plot(df.date_start.str.replace('2019-', ''), df.impressions)

Вот и всё! Получили всю информацию о затратах, просмотрах и кликах. В следующем материале посмотрим, как выгрузить её в формате json и передать в Redash для последующего анализа и визуализации.

Пост написан при участии команды Valiotti Analytics: Елисей Рыков (младший аналитик), Георгий Ушаков (технический консультант)