Автоматизация импорта данных в 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. Если таблица уже существует в выбранном датасете, то она будет перезаписана.

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

Проект в Google Cloud Platform с активированным биллингом.
Доступ на редактирование (роль «Редактор» данных BigQuery) и выполнение заданий (роль Пользователь заданий BigQuery) для сервисного аккаунта Cloud-функции в проекте BigQuery, куда будет загружена таблица;
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, потому что видим, что они самые востребованные среди наших клиентов. Поделитесь, какие способы вы используете для объединения данных и каких интеграций вам не хватает для этого.