Пишем простой счетчик для сайта или приложения с помощью Google Cloud Functions и AWS Lambda/Snowflake

Как работают популярные счетчики веб или мобильной аналитики, например, Google Analytics или AppsFlyer? На сайт устанавливаются их коды или в приложение интегрируется мобильное SDK. Потом при каждом действии клиента отправляется http запрос на сервер аналитики. Например, для Google Analytics в простейшем случае это будет: https://analytics.google.com/g/collect?v=2&tid=G-XXXXXXXXXX&cid=1415663337.1642776362&en=order , где:

• tid — уникальный идентификатор потока(счетчика)

• cid — уникальный идентификатор браузера

• en — название события

У использования стандартных счетчиков/пикселей есть минусы:

• некоторые посетители используют анонимайзеры, которые блокируют такие запросы;

• их сложно кастомизировать под себя.

В этой статье напишем собственный счетчик, который будет решать эти проблемы. Счетчик встроим в PowerBI отчеты. Но принцип одинаков, его можно будет использовать и на веб-сайте, и в приложении, и в других  устройствах с доступом к интернету. Попробуем две точки сбора событий, чтобы изучить больше технологий: Google Cloud Function, которая будет писать события в Google BigQuery, и Amazon Lambda Functions с записью событий в Snowflake.

1)  Заходим на страницу Google Cloud Function — Create function. Убеждаемся, что установлен триггер HTTP, не установлена галочка напротив Require authentication. Желательно, чтобы Region совпадал с регионом в вашем датасете в BigQuery.

URL на скриншоте – это тот адрес, куда нужно будет отправлять события. Например, https://us-central1-project-name.cloudfunctions.net/powerbi 

Через GET параметры можно передавать любые данные. Например, при использовании на веб-сайте будет полезно передавать referrer. В нашем же случае это будет название отчета в PowerBI и имя страницы. Тогда адрес для отправки будет иметь вид: https://us-central1-project-name.cloudfunctions.net/powerbi?report=MyReport&page=Page1

2) Далее попадаем на страницу редактирования кода функции. Выбираем Python. В файле requirements.txt прописываем библиотеки, которые нужно подгрузить в проект:

requests>=2.27.1 google.cloud>=0.34.0 google-cloud-bigquery>=2.32.0

В веб-интерфейсе создаем файл config.py следующего содержания: 

# id чата в телеграме куда будут отправляться ошибки. чтобы узнать свой id - напишите боту @userinfobot  tg_chat_id = 'XXXXXXX'  # токен бота. чтобы его получить, создайте бота написав @BotFather tg_bot_token = 'YYYYYYY:YYYYYYYYYYYYYYYYYYYYYYYYYY'  # название таблицы в BigQuery куда будут писаться события. Ее нужно создать заранее. bq_table_name = 'project-name.dataset.powerbi_views'

Файл main.py: 

import datetime import requests from google.cloud import bigquery from config import tg_chat_id, tg_bot_token, bq_table_name   def main(request, report='', page='', ip='', user_agent=''):     if 'report' in request.args:         report = request.args['report']     if 'page' in request.args:         page = request.args['page']        #for headers in request.headers:  header = header + str(headers)  # получить все заголовки     ip = request.headers['X-Forwarded-For']     user_agent = request.headers['User-Agent']       if report != '':         return stream_bq(report, page, ip, user_agent)     else:         url = 'https://api.telegram.org/bot' + tg_bot_token + '/sendMessage?chat_id=' + tg_chat_id + '&text=google cloud functions powerbi error: event without parameters received'         response = requests.get(url)         return 'event without parameters received'        def stream_bq(report, page, ip, user_agent):     bq_client = bigquery.Client()     bq_table = bq_client.get_table(bq_table_name)     row = [{'report': report,             'page': page,             'ip': ip,             'timestamp': datetime.datetime.utcnow(),             'user_agent': user_agent  }]     bq_client.insert_rows(bq_table, row)     return 'ok'

Нажимаем Deploy, проверяем работу: открываем в браузере URL из пункта 1. Должны увидеть текст “ок”. Если это не так – ищем ошибки в Logs.

3) В BigQuery данные будут выглядеть следующим образом:

Далее обогатим эти данные:

3.1) По ip  определим страну, город, почтовый индекс, используя данные GeoLite2 Free Geolocation Data:

SELECT *  FROM ( SELECT ip FROM `project-name.dataset.powerbi_views` GROUP BY ip ) AS q  CROSS JOIN (    SELECT country_name, network, network_bin, city_name, postal_code, mask    FROM `fh-bigquery.geocode.201806_geolite2_city_ipv4_locs`  ) AS g  WHERE network_bin = NET.IP_FROM_STRING(ip) & NET.IP_NET_MASK(4, mask)

Результат будет выглядеть примерно так:

3.2) Из user_agent выделим  операционную систему и браузер, используя JS библиотеку woothee:

CREATE OR REPLACE FUNCTION `project-name.dataset.decode_user_agent`(ua STRING)    RETURNS STRING    LANGUAGE js AS """  return JSON.stringify(woothee.parse(ua));    """OPTIONS(library="gs://project-name-bucket/woothee.js");

SELECT JSON_VALUE(json, '$.os') AS os, JSON_VALUE(json, '$.name') AS browser, json, user_agent, FROM  (    SELECT user_agent, `bproject-name.dataset.decode_user_agent`(user_agent) AS json    FROM `bi-analytics-318415.Others.powerbi_views`    GROUP BY user_agent )

Результат:

3.3) Теперь мы знаем не только названия отчетов и страницы, которые смотрят пользователи, но и информацию о местоположении, устройстве и операционную систему. Можно сформировать отчетность на свой вкус.

4) Интегрируем наш счетчик в PowerBI. Для этого в редакторе создаем таблицу с единственным значением – нашим URL:

Затем выбираем эту колонку справа и в меню Column tools указываем Data category = Image URL:

После этого вставляем в отчет визуальный элемент “таблица” с этой колонкой где-нибудь в малозаметном углу отчета. Теперь при каждом его открытии будет уходить http запрос, который через Cloud Functions будет фиксироваться в BigQuery. Наш счетчик готов!

5) Чтобы не быть ограниченным одним облачным вендором, проделаем аналогичное на AWS Lambda. В качестве хранилища будем использовать Snowflake, но вы можете использовать любое, например, Redshift. Скорость записи (по логам) в таком случае будет медленней чем в первом примере Google Cloud Function -> Google BigQuery. Но у нас еще побочная цель – изучить разные подходы, чтобы каждый мог для себя выбрать оптимальный вариант.

Заходим в AWS Lambda. Создаем новую функцию powerbi . Вариант создания – “Start with a simple Hello World example”. Runtime – указываем версию Python, которая установлена у вас на локальном компьютере, например, Python 3.8. Далее нужно добавить триггер типа API Gateway — HTTP API. Security указываем Open. После выполненных действий в консоли, функция должна выглядеть примерно так:

URL для отправки событий будет иметь вид: https://your-id.execute-api.us-east-2.amazonaws.com/default/powerbi?report=MyReport&page=Page1

По умолчанию Timeout для функции (время ее работы) составляет 3 секунды. Для нас этого недостаточно, т.к. будем писать во внешний сервис Snowflake. Особенно много времени нужно, если события будут приходить редко, т.к. много времени уходит на запуск Warehouse в Snowflake после его остановки. Заходим в Configuration -> General configuration -> Edit -> Timeout = 12 sec.

6) Python код нашей функции будет использовать библиотеки requests и snowflake.connector. В AWS Lambda библиотеки подключаются не так как в Google Cloud Functions. Поэтому сначала создадим проект на локальном компьютере, потом загрузим в Lambda zip-архив. Инструкция. По моему опыту, самым оптимальным является вариант Using a virtual environment.

На локальном компьютере создаем пустую папку, где будут лежать файлы проекта. 

Создаем файл config.py следующего содержания: 

# id чата в телеграме куда будут отправляться ошибки. чтобы узнать свой id - напишите боту @userinfobot  tg_chat_id = 'XXXXXXX'  # токен бота. чтобы его получить, создайте бота написав @BotFather tg_bot_token = 'YYYYYYY:YYYYYYYYYYYYYYYYYYYYYYYYYY'   # ниже реквизиты доступа от snowflake. таблицу нужно создать заранее sn_user = 'user_name' sn_password = 'user_password' sn_account = 'youproject.us-east-2.aws' sn_database = 'test' sn_schema = 'PUBLIC' sn_table = 'powerbi_views' sn_warehouse = 'COMPUTE_WH' sn_role_name = 'ACCOUNTADMIN'

Создаем файл lambda_function.py :

import datetime import json import requests import snowflake.connector from config import tg_chat_id, tg_bot_token, sn_user, sn_password, sn_account, \                    sn_warehouse, sn_database, sn_table, sn_schema   def lambda_handler(event={}, context='', report='', page='', ip='', user_agent=''):     if event.get('queryStringParameters') is not None:         if 'report' in event['queryStringParameters']:             report = event['queryStringParameters']['report']         if 'page' in event['queryStringParameters']:             page = event['queryStringParameters']['page']         ip = event['headers']['x-forwarded-for']         user_agent = event['requestContext']['http']['userAgent']         time = datetime.datetime.utcnow()         return stream_snowflake(report, page, ip, time, user_agent)     else:         url = 'https://api.telegram.org/bot' + tg_bot_token + '/sendMessage?chat_id=' + tg_chat_id + '&text=aws lambda powerbi error: event without parameters received. event = ' + str(event)         response = requests.get(url)         return 'event without parameters received'     def stream_snowflake(report, page, ip, time, user_agent):     error = False     try:         conn = snowflake.connector.connect(user=sn_user, password=sn_password, account=sn_account, warehouse=sn_warehouse,                                            database=sn_database, schema=sn_schema)         cs = conn.cursor()         cs.execute(             "INSERT INTO " + sn_table + "(report, page, ip, time, user_agent) VALUES " +             "    (%s, %s, %s, %s, %s) ", (report, page, ip, time, user_agent))     except Exception as e:         error = True         url = 'https://api.telegram.org/bot' + tg_bot_token + '/sendMessage?chat_id=' + tg_chat_id + '&text=aws lambda powerbi error:' + str(e)         response = requests.get(url)     finally:         try:             cs.close()         except Exception as u:             pass     if error == False:         return 'Write in snowflake successfully'     else:         return 'Was error with Snowflake. Send message to telegram'   # это код для запуска на локальном компьютере # if __name__== "__main__": #     lambda_handler()

Проверим на локальном компьютере, что наш скрипт работает. Для этого нужно раскомментировать блок снизу и в функции lambda_handler прописать вызов stream_snowflake без условий (то есть с пустыми значениями report и т.д.) Если все норм, загружаем наш проект в AWS Lambda по инструкции в начале этого пункта.

Открываем в браузере URL из пункта 5 для запуска Lambda функции, проверяем что все работает. Логи находятся в Monitor -> Logs.

7) В Snowflake собираемы данные должны выглядеть следующим образом:

Далее обогатим эти данные:

7.1) Стандартного общедоступного набора данных по geo-ip в Snowflake нет (или я не нашел). Но на Snowflake Data Marketplace есть возможность получить 2-х недельный триал к IPINFO: IP GEOLOCATION . После получения доступа скопируем данные себе. После окончания триала они перестанут обновляться. На мою первую заявку не было никакой реакции, поэтому пришлось зарегистрировать второй аккаунт – с ним все получилось.

После получения доступа, если мы попробуем скопировать таблицы командой CLONE — но там покажется ошибка: SQL compilation error: Cannot clone from a table that was imported from a share.

Поэтому скопируем двумя командами:

CREATE TABLE test.public.location_ip LIKE ipinfo_snowflake_myproject_trial.public.location; INSERT INTO test.public.location_ip SELECT * FROM ipinfo_snowflake_myproject_trial.public.location;

Далее для получения региона и индекса (не влез на скриншот) можно использовать этот SQL-запрос:

SELECT *  FROM  (     SELECT ip,  PARSE_IP(ip, 'inet'):ipv4 AS ipv4      FROM ( SELECT ip FROM test.public.powerbi_views WHERE ip='X.X.X.X' GROUP BY ip )  ) AS q  LEFT JOIN test.public.location_ip AS l ON q.ipv4<=l.end_ip_int AND q.ipv4>=l.start_ip_int

7.2) Для парсинга user_agent не получиться использовать JavaScript UDF функцию, т.к. в Snowflake они не поддерживаются. Поддержка Java UDF функций экспериментальна и доступна только для проектов на AWS. Будем использовать библиотеку Yauaa.     

Перейдем в БД TEST и создадим Stage. Назовем его, например, YAUAA:

В случае, если выбрали Snowflake Managed, то загрузить файл туда можно с помощью SnowSQL (CLI Client) . Неочевидный момент: в файле ./snowflake/config region нужно указывать вместе с вендором, например us-east-2.aws. После запуска утилиты файл можно загрузить командой PUT: put file:///home/anton/yauaa-snowflake-6.9-udf.jar @YAUAA;

Проверим командой: LIST @YAUAA; 

Теперь можно создать UDF функцию. Будьте внимательны, директива imports регистрозависима:

use test;  create or replace function parse_useragent(useragent VARCHAR) returns object language java imports = ('@YAUAA/yauaa-snowflake-6.9-udf.jar') handler = 'nl.basjes.parse.useragent.snowflake.ParseUserAgent.parse';

SQL-код для определения операционной системы и браузера:

SELECT    parse_useragent(user_agent):AgentName::string AS browser,    parse_useragent(user_agent):OperatingSystemName::string AS os,    user_agent FROM ( SELECT user_agent FROM test.public.powerbi_views GROUP BY user_agent )

Таким образом мы реализовали в Snowflake такой же функционал, как в BigQuery в пунктах 3 — 4. Выбор за вами. Счетчик можно интегрировать не только в PowerBI, но и в другие BI решения, установить на сайт, в приложение или программу.