Автоматизация импорта данных в Google BigQuery с помощью Google Cloud Functions

Мы постоянно работаем с Google BigQuery – импортируем данные о пользователях, их заказах и расходах на рекламу из разных источников, чтобы иметь возможность объединять их между собой. Что нам это дает? Например, если у вас интернет-магазин и клиент делает заказ по телефону, а потом авторизовывается на сайте, то с помощью Google BigQuery можно связать все его действия задним числом. Можно отслеживать весь путь клиента по маркетинговой воронке – от первого попадания на сайт до покупки в brick and mortar магазине, и оценивать рекламные кампании с учетом таких офлайн-продаж.

В этой подборке у нас 6 python-скриптов для автоматизации импорта данных в Google BigQuery с помощью Google Cloud функции: FTP; FTPS; HTTP(s); Intercom; MySQL и SFTP. Принцип работы один: с помощью HTTP POST запроса вызывается Cloud-функция, которая получает данные из источника и загружает их в таблицу Google BigQuery. Если таблица уже существует в выбранном датасете, то она будет перезаписана.

Основные требования

1. Проект в Google Cloud Platform с активированным биллингом.
2. Доступ на редактирование (роль «Редактор» данных BigQuery) и выполнение заданий (роль Пользователь заданий BigQuery) для сервисного аккаунта Cloud-функции в проекте BigQuery, куда будет загружена таблица;
3. HTTP-клиент для выполнения POST запросов, вызывающих Cloud-функцию.

Этапы настройки

• Перейдите в Google Cloud Platform Console и авторизуйтесь с помощью Google аккаунта, или зарегистрируйтесь, если аккаунта еще нет.
• Перейдите в проект с активированным биллингом или создайте новый биллинг-аккаунт для проекта.
• Перейдите в раздел Cloud Functions и нажмите «Создать функцию». Обратите внимание, что за использование Cloud-функций взимается плата.
• Заполните следующие поля:

Название: например, ftp-bq-integration или любое другое подходящее название;

Выделенный объем памяти: 2 ГБ или меньше, в зависимости от размера обрабатываемого файла;

Триггер: HTTP;

Исходный код: Встроенный редактор;

Среда выполнения: Python 3.X.

• Скопируйте содержимое файла main.py в встроенный редактор, вкладка main.py.
• Скопируйте содержимое файла requirements.txt в встроенный редактор, вкладка requirements.txt.
• В качестве вызываемой функции укажите ftp/ftps/https и так далее, в зависимости от модуля, который вы используете.
• В дополнительных параметрах увеличьте время ожидания с 60 сек. до 540 сек. или меньшее, в зависимости от размеров обрабатываемого файла.
• Завершите создание Cloud-функции, нажав на кнопку «Создать».

FTP / FTPS / SFTP

Этот модуль предназначен для передачи файлов с FTP (FTPS, SFTP) — серверов в Google BigQuery с помощью Google Cloud функции. Решение позволяет автоматически выполнять загрузку данных в Google BigQuery из файла, который регулярно обновляется на FTP-сервере.

Файл, который нужно получить с соответствующего сервера, может иметь любое подходящее расширение (.json, .txt, .csv), но должен быть в одном из следующих форматов: JSON (newline-delimited) или Comma-separated values (CSV).

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

from urllib import urlencode from httplib2 import Http  trigger_url = "https://REGION-PROJECT_ID.cloudfunctions.net/ftp/" headers = { "Content-Type": "application/json" } payload = {                "ftp":                          {                           "user": "ftp.user_name",                           "psswd": "ftp.password",                           "path_to_file": "ftp://server_host/path/to/file/"                         },                "bq":                          {                           "project_id": "my_bq_project",                           "dataset_id": "my_bq_dataset",                           "table_id": "my_bq_table",                           "delimiter": ",",                           "source_format": "NEWLINE_DELIMITED_JSON",                           "location": "US"                         }             } Http().request(trigger_url, "POST", urlencode(payload), headers = headers)

HTTP(s)

Модуль для передачи файлов с HTTPS-серверов в Google BigQuery.

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

from urllib import urlencode from httplib2 import Http  trigger_url = "https://REGION-PROJECT_ID.cloudfunctions.net/https/" headers = { "Content-Type": "application/json" } payload = {                 "https":                          {                             "path_to_file": "https://server_host/path/to/file/",                             "user": "https.user_name",                             "psswd": "https.password"                         },                 "bq":                         {                             "project_id": "my_bq_project",                             "dataset_id": "my_bq_dataset",                             "table_id": "my_bq_table",                             "delimiter": ",",                             "source_format": "CSV",                             "location": "US"                         }             }  Http().request(trigger_url, "POST", urlencode(payload), headers = headers)

Intercom

Модуль для автоматизации передачи данных из Intercom в Google BigQuery с помощью Google Cloud функции. В настоящее время модуль позволяет импортировать из Intercom такие сущности, как: users, companies, contacts, admins, conversations, teams, tags, segments. При этом модуль не поддерживает custom attributes.

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

from urllib import urlencode from httplib2 import Http  trigger_url = "https://REGION-PROJECT_ID.cloudfunctions.net/intercom/" headers = { "Content-Type": "application/json" } payload = {           "intercom": {             "accessToken": "INTERCOM ACCESS TOKEN",             "entities": [                     "users",                      "companies",                      "contacts",                      "admins",                     "conversations",                     "teams",                     "tags",                     "segments"                     ]                 },             "bq": {                 "project_id": "YOUR GCP PROJECT",                 "dataset_id": "YOUR DATASET NAME",                 "location": "US"              } } Http().request(trigger_url, "POST", urlencode(payload), headers = headers)

MySQL

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

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

from urllib import urlencode from httplib2 import Http  trigger_url = "https://REGION-PROJECT_ID.cloudfunctions.net/mysql/" headers = { "Content-Type": "application/json" } payload = {             "mysql":                      {                         "user": "mysql.user",                         "psswd": "mysql.password",                         "host": "host_name",                         "port”: 3306,                         "database": "database_name",                         "table_id": "table_name",                         "query": "SELECT * FROM table_name"                     },             "bq":                      {                         "project_id": "my_bq_projec",                         "dataset_id": "my_bq_dataset",                         "table_id": "my_bq_table"                     } }  Http().request(trigger_url, "POST", urlencode(payload), headers = headers)

Более подробную документацию для каждого модуля можно найти в файлах readme в каждом из разделов.

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