9 заметок с тегом

Data engineering

Семантический анализ мнений о поправках к Конституции на основе данных ВКонтакте

Сегодня поработаем с открытыми данными из ВКонтакте и получим семантическую оценку на популярное и актуальное событие — поправки к Конституции Российской Федерации.

Обзор методов API

Воспользуемся методом newsfeed.search: он позволяет получить до тысячи последних постов из новостной ленты по ключевому слову. В результате приходит много полей: среди них идентификаторы записи и пользователя или сообщества, текст поста, количество лайков, комментарии, приложения, геопозиция и прочее. Нас интересуют только идентификаторы и текст.
Для аналитики пригодится расширенная информация об авторе поста: его город, пол и возраст можно получить методом users.get, причём в запросе будем отправлять сразу до тысячи пользователей.

Создаём таблицы в Clickhouse

Данные нужно будет где-то хранить, в качестве СУБД подойдёт Clickhouse. Создадим две таблицы: для постов и для пользователей. В первой будем хранить идентификаторы и текст поста, во второй — данные о пользователе: его id, пол, возраст и город. Движок ReplacingMergeTree() будет удалять дубликаты.

Мы уже писали о том, как установить Clickhouse на бесплатную машину AWS, создавать в нём внешние словари и материализованные представления

CREATE TABLE vk_posts(
   post_id UInt64,
   post_date DateTime,
   owner_id UInt64,
   from_id UInt64,
   text String
) ENGINE ReplacingMergeTree()
ORDER BY post_date

CREATE TABLE vk_users(
   user_id UInt64,
   user_sex Nullable(UInt8),
   user_city String,
   user_age Nullable(UInt16)
) ENGINE ReplacingMergeTree()
ORDER BY user_id

Сбор постов через API ВКонтакте

Перейдём к написанию скрипта. Импортируем библиотеки и задаём несколько константных значений:

В материале «Собираем данные по рекламным кампаниям ВКонтакте» подробно описан процесс получения токена пользователя для VK API

from clickhouse_driver import Client
from datetime import datetime
import requests
import pandas as pd
import time

token = 'your_token'
version = 5.103
client = Client(host='ec1-23-456-789-1011.us-east-2.compute.amazonaws.com', user='default', password='', port='9000', database='default')      
data_list = []
start_from = 0
query_string = 'конституция'

Опишем функцию get_and_insert_info_by_user — она будет принимать список идентификаторов пользователей, получать расширенную информацию о них и отправлять в таблицу vk_users. Так как параметр user_ids метода принимает список как строку, переводим структуру в тип str и отсекаем квадратные скобки. Многие пользователи скрывают пол, возраст или город — в таком случае вставляет Nullable значения. Для получения возраста берём текущий год и вычитаем год из даты рождения, если он представлен — проверку делаем регулярным выражением по четырём цифрам.


Функция get_and_insert_info_by_user

def get_and_insert_info_by_user(users):
    try:
        r = requests.get('https://api.vk.com/method/users.get', params={
            'access_token':token,
            'v':version,
            'user_ids':str(users)[1:-2],
            'fields':'sex, city, bdate'
        }).json()['response']
        for user in r:
            user_list = []
            user_list.append(user['id'])
            if client.execute(f"SELECT count(1) FROM vk_users where user_id={user['id']}")[0][0] == 0:
                print(user['id'])
                try:
                    user_list.append(user['sex'])
                except Exception:
                    user_list.append('cast(Null as Nullable(UInt8))')
                try:
                    user_list.append(user['city']['title'])
                except Exception:
                    user_list.append('')
                try:
                    now = datetime.now()
    			    year = item.split('.')[-1]
    			    if re.match(r'\d\d\d\d', year):
        		        age = now.year - int(year)
			    	   user_list.append(age)
                except Exception:
                    user_list.append('cast(Null as Nullable(UInt16))')
                user_insert_tuple = tuple(user_list)
                client.execute(f'INSERT INTO vk_users VALUES {user_insert_tuple}')
    except KeyError:
        pass


Наш скрипт будет работать в вечном цикле, чтобы постоянно добирать новые данные, ведь мы можем получать только тысячу последних. Метод newsfeed.search за раз возвращает двести постов, так что нужно вызывать его пять раз подряд и собирать все ответы.


Цикл сбора новых постов

while True:
    for i in range(5):
        r = requests.get('https://api.vk.com/method/newsfeed.search', params={
            'access_token':token,
            'v':version,
            'q':query_string,
            'count':200,
            'start_from': start_from
        })
        data_list.append(r.json()['response'])
        try:
            start_from = r.json()['response']['next_from']
        except KeyError:
            pass

Полученные в ответе данные можно распарсить. В ВКонтакте у пользователей id всегда положительный, а у сообществ идёт со знаком минус. Чтобы получить данные только от пользователей, будем собирать только те, где from_id больше нуля. Следующая проверка — на отсутствие текста в посте, такие нам тоже не нужны. Наконец, будем собирать данные только если таких ещё нет — для этого обращаемся к таблице vk_posts по текущему id. В конце приостановим скрипт на 180 секунд, чтобы дождаться новых постов и не столкнуться с ограничениями по запросам VK API.


Занесение новых данных в Clickhouse

user_ids = []
    for data in data_list:
        for data_item in data['items']:
            if data_item['from_id'] > 0:
                post_list = []
                if not data_item['text']:
                    continue
                if client.execute(f"SELECT count(1) FROM vk_posts WHERE post_id={data_item['id']} AND from_id={data_item['from_id']}")[0][0] == 0:
                    user_ids.append(data_item['from_id'])
                    date = datetime.fromtimestamp(data_item['date'])
                    date = datetime.strftime(date, '%Y-%m-%d %H:%M:%S')
                    post_list.append(date)
                    post_list.append(data_item['id'])
                    post_list.append(data_item['owner_id'])
                    post_list.append(data_item['from_id'])
post_list.append(data_item['text'].replace("'","").replace('"','').replace("\n",""))
                    post_list.append(query_string)
                    post_tuple = tuple(post_list)
                    print(post_list)
                    try:
                        client.execute(f'INSERT INTO vk_posts VALUES {post_tuple}')
                    except Exception as E:
                        print('!!!!! try to insert into vk_post but got', E)
    try:
        get_and_insert_info_by_user(user_ids)
    except Exception as E:
        print("Try to insert user list:", user_ids, "but got:", E)
    time.sleep(180)

Анализ постов через Dostoevsky

Этот скрипт мы оставили работать на неделю: за это время он набрал почти 20000 постов из ВКонтакте, в которых упоминается ключевое слово «конституция». Напишем второй скрипт — для аналитики и визуализации данных. Для начала соберём данные из таблицы, сформируем DataFrame и для каждого поста получим значения тональности: насколько он положителен, отрицателен и нейтрален. Для оценки тональности текста будем использовать библиотеку Dostoevsky.

from dostoevsky.tokenization import RegexTokenizer
from dostoevsky.models import FastTextSocialNetworkModel
from clickhouse_driver import Client
import pandas as pd
client = Client(host='ec1-23-456-789-1011.us-east-2.compute.amazonaws.com', user='default', password='', port='9000', database='default')

Простым запросом содержимое всей таблицы с постами занесём в переменную vk_posts. Пройдём все посты, выберем те посты, где есть текст помимо пробелов и положим их в DataFrame.

vk_posts = client.execute('SELECT * FROM vk_posts')
list_of_posts = []
list_of_ids = []
for post in vk_posts:
    if str(post[-2]).replace(" ", ""):
        list_of_posts.append(str(post[-2]).replace("\n",""))
        list_of_ids.append(int(post[2]))
df_posts = pd.DataFrame()
df_posts['post'] = list_of_posts
df_posts['id'] = list_of_ids

Обходим моделью весь список постов с текстом и получаем к оценку тональности для каждой записи.

tokenizer = RegexTokenizer()
model = FastTextSocialNetworkModel(tokenizer=tokenizer)
sentiment_list = []
results = model.predict(list_of_posts, k=2)
for sentiment in results:
    sentiment_list.append(sentiment)

Для каждой строки в DataFrame заведём ещё три колонки: насколько запись положительна, отрицательна и нейтральна. В случае, если по одному из трёх параметров ничего не вернулось, будем заносить ноль.

neutral_list = []
negative_list = []
positive_list = []
speech_list = []
skip_list = []
for sentiment in sentiment_list:
    neutral = sentiment.get('neutral')
    negative = sentiment.get('negative')
    positive = sentiment.get('positive')
    if neutral is None:
        neutral_list.append(0)
    else:
        neutral_list.append(sentiment.get('neutral'))
    if negative is None:
        negative_list.append(0)
    else:
        negative_list.append(sentiment.get('negative'))
    if positive is None:
        positive_list.append(0)
    else:
        positive_list.append(sentiment.get('positive'))
df_posts['neutral'] = neutral_list
df_posts['negative'] = negative_list
df_posts['positive'] = positive_list

Посмотрим, как выглядит наш DataFrame теперь:

Можем посмотреть примеры самых негативных постов:

df_posts[df_posts.negative > 0.9]

Нашей таблице не хватает данных об авторах постов. Возьмём их из таблицы vk_users и сольём обе таблицы по полю «id».

vk_users = client.execute('SELECT * FROM vk_users')
vk_user_ids_list = []
vk_user_sex_list = []
vk_user_city_list = []
vk_user_age_list = []
for user in vk_users:
    vk_user_ids_list.append(user[0])
    vk_user_sex_list.append(user[1])
    vk_user_city_list.append(user[2])
    vk_user_age_list.append(user[3])
df_users = pd.DataFrame()
df_users['id'] = vk_user_ids_list
df_users['sex'] = vk_user_sex_list
df_users['city'] = vk_user_city_list
df_users['age'] = vk_user_age_list
df = df_posts.merge(df_users, on='id')

Теперь таблица выглядит так:

Анализируем графики от plotly

В материале «Как построить красивый waterfall chart в Python?» мы уже строили графики библиотекой plotly

Для начала посчитаем процентное соотношение постов с положительной, отрицательной и нейтральной тональностью: пройдём все три столбца и подсчитаем для каждого случая строки, отличные от нуля. Затем проделаем то же самое для разных возрастных категорий и половых принадлежностей.

Из графика следует, что 46% постов по запросу «конституция» за последнюю неделю имеют негативный окрас. Другие 52% высказываются нейтрально. Чуть позже узнаем, насколько мнения в интернете совпадают с официальными результатами голосования.

Заметно, что доля положительных постов среди мужской аудитории составляет 2%, среди женской — вдвое больше, 4%. Впрочем, негативных постов в обоих группах практически поровну: 47% среди мужской и 44% среди женской.

Наконец, оценка постов по возрастным группам: больше всего доля позитивного текста наблюдается в группе 18 — 25 лет, это 3%. Меньше всего позитивных постов в группе до 18 лет, но это может происходить и в связи с тем, что многие пользователи моложе 18 лет предпочитают скрывать возраст, и точные данные по такой группе получить не удастся. Негативных постов во всех группах кроме 18 — 25 поровну: 46%.
Заметно, что на всех трёх графиках данные распределены приблизительно одинаково. Это говорит о том, что за последнюю неделю практически половина всех постов по ключевому слову «конституция» в новостной ленте ВКонтакте имела негативный окрас.

 Нет комментариев    51   5 ч   Data analytics   Data engineering   plotly

Обрабатываем нажатие кнопки в Selenium

В материале «Парсим данные, используя Buetiful Soup и Selenium» мы уже рассмотрели, как быть, когда данные на странице динамически подгружаются при скролле страницы. Но бывают ситуации, когда новые данные можно получить, только нажав на кнопку «Показать ещё» — сегодня узнаем, как через Selenium сымитировать нажатие кнопки для полного открытия страницы, соберём идентификаторы пива, оценки к каждому продукту и отправим данные в Clickhouse

Структура страницы

Возьмём случайную пивоварню — у неё 105 чекинов, то есть, отзывов. Страница с чекинами пивоварни показывает не более 25 чекинов и выглядит так:

Если попробуем промотать в самый низ, столкнёмся с той самой кнопкой, мешающей нам взять все 105 за раз:

Мы поступим так: выясним, к какому классу относится элемент кнопки и будем на неё нажимать, пока это возможно. Так как Selenium запускает браузер, следующая кнопка «Показать ещё» может не успеть прогрузиться, поэтому между нажатиями поставим интервал в пару секунд. Как только страница раскроется полностью — мы возьмём её содержимое и распарсим нужные данные из чекинов. Зайдём в код страницы и найдём кнопку — она относится к классу more_checkins.

У кнопки есть свойства стиля, а именно — display. В случае, если кнопка должна отображаться, display принимает значение block. Но когда промотаем страницу до самого конца, кнопку не нужно будет показывать, ведь открывать больше нечего — поэтому display кнопки примет значение none. В случае, если мы запросим у кнопки display и вернётся none будем знать, что открывать больше нечего и можно перестать жать на кнопку.

Пишем код

Начнём с импорта библиотек:

import time
from selenium import webdriver
from bs4 import BeautifulSoup as bs
import re
from datetime import datetime
from clickhouse_driver import Client

Chromedriver, необходимый для запуска браузера через Selenium, можно установить с официальной страницы

Подключимся к базе данных, зададим cookies:

client = Client(host='ec1-23-456-789-10.us-east-2.compute.amazonaws.com', user='', password='', port='9000', database='')
count = 0
cookies = {
    'domain':'untappd.com',
    'expiry':1594072726,
    'httpOnly':True,
    'name':'untappd_user_v3_e',
    'path':'/',
    'secure':False,
    'value':'your_value'
}

О том, как запускать Selenium с cookies можно прочитать в материале «Парсим данные каталога сайта, используя Beautiful Soup и Selenium». Нам нужен параметр untappd_user_v3_e.

Так как мы планируем работать с пивоварнями, у которых более сотни тысяч чекинов, может оказаться так, что страница будет чересчур тяжёлой, и нагрузка на машину будет огромна. Чтобы этого избежать, отключим всё лишнее, а затем подключим cookie для авторизации:

options = webdriver.ChromeOptions()
prefs = {'profile.default_content_setting_values': {'images': 2, 
                            'plugins': 2, 'fullscreen': 2}}
options.add_experimental_option('prefs', prefs)
options.add_argument("start-maximized")
options.add_argument("disable-infobars")
options.add_argument("--disable-extensions")
driver = webdriver.Chrome(options=options)
driver.get('https://untappd.com/TooSunnyBrewery')
driver.add_cookie(cookies)

Напишем функцию, которая принимает ссылку, переходит по ней, полностью раскрывает страницу и возвращает нам soup, который можно будет распарсить. Получим display кнопки и запишем в переменную more_checkins: пока он не равен none будем нажимать на кнопку и снова получать её display. Сделаем интервал в две секунды между нажатиями, чтобы подождать прогрузку страницы. Как только будет получена вся страница, переведём её в soup библиотекой bs4.

def get_html_page(url):
    driver.get(url)
    driver.maximize_window()
    more_checkins = driver.execute_script("var more_checkins=document.getElementsByClassName('more_checkins_logged')[0].style.display;return more_checkins;")
    print(more_checkins)
    while more_checkins != "none":
        driver.execute_script("document.getElementsByClassName('more_checkins_logged')[0].click()")
        time.sleep(2)
        more_checkins = driver.execute_script("var more_checkins=document.getElementsByClassName('more_checkins_logged')[0].style.display;return more_checkins;")
        print(more_checkins)
    source_data = driver.page_source
    soup = bs(source_data, 'lxml')
    return soup

Напишем следующую функцию: она тоже будет принимать url страницы, передавать его в get_html_page, получать soup и парсить его. Функция вернёт запакованные списки с идентификатором пива и оценкой к нему.

О том, как парсить элементы страницы мы уже говорили в материале «Парсим данные каталога сайта, используя Beautiful Soup».

def parse_html_page(url):
    soup = get_html_page(url)
    brewery_id = soup.find_all('a', {'class':'label',
                                     'href':re.compile('https://untappd.com/brewery/*')})[0]['href'][28:]
    items = soup.find_all('div', {'class':'item',
                                  'id':re.compile('checkin_*')})
    checkin_rating_list = []
    beer_id_list = []
    count = 0
    print('Заполняю списки')
    for checkin in items:
        print(count, '/', len(items))
        try:
            checkin_rating_list.append(float(checkin.find('div', {'class':'caps'})['data-rating']))
        except Exception:
            checkin_rating_list.append('cast(Null as Nullable(Float32))')
        try:
            beer_id_list.append(int(checkin.find('a', {'class':'label'})['href'][-7:]))
        except Exception:
            beer_id_list.append('cast(Null as Nullable(UInt64))')
        count += 1 
    return zip(checkin_rating_list, beer_id_list)

Наконец, напишем вызов функций по пивоварням. В материале «Использование словарей в Clickhouse на примере данных Untappd» мы уже рассмотрели, как получить список идентификаторов российских пивоварен — обратимся к нему через таблицу в Clickhouse

brewery_list = client.execute('SELECT brewery_id FROM brewery_info')

Если посмотрим на brewery_list, то узнаем, что данные вернулись в неудобном формате: это список кортежей.

Небольшое лямбда-выражение позволит его «выпрямить»:

flatten = lambda lst: [item for sublist in lst for item in sublist]
brewery_list = flatten(brewery_list)

Работать с таким списком значительно комфортнее:

Для каждой пивоварни в списке сформируем url — он состоит из стандартной ссылки и идентификатора пивоварни в конце. Отправим url в функцию parse_html_page, которая сама вызовет get_html_page и вернёт списки с beer_id и rating_score. Так как два списка вернутся упакованными можем пройти по ним итератором, сформировав кортеж и отправив его в Clickhouse.

for brewery_id in brewery_list:
    print('Беру пивоварню с id', brewery_id, count, '/', len(brewery_list))
    url = 'https://untappd.com/brewery/' + str(brewery_id)
    returned_checkins = parse_html_page(url)
    for rating, beer_id in returned_checkins:
        tuple_to_insert = (rating, beer_id)
        try:
            client.execute(f'INSERT INTO beer_reviews VALUES {tuple_to_insert}')
        except errors.ServerException as E:
            print(E)
    count += 1

С кнопками на этом всё. Постепенно мы формируем отличный датасет для последующего анализа, который рассмотрим в следующем цикле статей, как только завершим сбор данных — возможно, через месяц.

 Нет комментариев    55   13 дн   Amazon Web Services   AWS   clickhouse   Data engineering   python

Использование словарей в Clickhouse на примере данных Untappd

В Clickhouse реализована возможность использования внутренних и внешних словарей, которые могут быть альтернативой JOIN (которые, к сожалению, не всегда здорово работают). Словари хранят информацию в памяти и к ним можно обратиться командой dictGet. Рассмотрим как создать словарь в Clickhouse и как его можно использовать в запросах.

Будем изучать функционал на примере данных из API Untappd. Untappd — социальная сеть любителей крафтового пива. Мы сфокусируемся на чекинах российких крафтовых пивоварен, начнем собирать информацию о них, чтобы в следующих постах проанализировать данные и сделать некоторые выводы. В рамках этого поста разберем получение мета-информации о российских пивоварнях на Untappd, а полученные данные сохраним в словаре Clickhouse.

Собираем данные с Untappd

Для обращений к API нужны client_id и  client_secret_key — их можно получить, создав приложение. Для этого переходим в раздел создания приложения в документации и указываем некоторые данные:

После отправления заявки нужно будет подождать некоторое время: от 1 до 3 недель.

import requests
import pandas as pd
import time

Отправлять запросы к API будем через requests, а в pandas посмотрим на результаты и выгрузим в csv, чтобы отправить в словарь Clickhouse. У Untappd строгие ограничения на количество запросов: всего в час можно отправить 100 запросов, поэтому будем библиотекой time ставить скрипт в ожидание на 38 секунд, чтобы число запросов в час не превосходило 100.

client_id = 'ваш_client_id'
client_secret = 'ваш_client_secret'
all_brewery_of_russia = []

Мы хотим собрать всю тысячу российских пивоварен. Один запрос к методу Brewery Search позволяет получить до 50 пивоварен. При поиске вручную на сайте Untappd по слову «Russia» сайт выдаст 3369 пивоварен:

Проверим это: пролистаем страницу до самого низа и откроем код страницы.

Каждая полученная пивоварня в поиске находится в классе beer-item. Значит, можем в поиске посчитать количество упоминаний beer-item:

И выясняем, что на самом деле их здесь ровно 1000, а не 3369. По запросу Russia в выборку попадают и американские пивоварни, а некоторые были удалены. Значит, придётся отправить 20 запросов, будем получать по 50 пивоварен за раз:

for offset in range(0, 1000, 50):
    try:
        print('offset = ', offset)
        print('осталось:', 1000 - offset, '\n')
        response = requests.get(f'https://api.untappd.com/v4/search/brewery?client_id={client_id}&client_secret={client_secret}',
                               params={
                                   'q':'Russia',
                                   'offset':offset,
                                   'limit':50
                               })
        item = response.json()
        print(item, '\n')
        all_brewery_of_russia.append(item)
        time.sleep(37)
    except Exception:
        print(Exception)
        continue

В параметрах метод Brewery Search принимает q — строку, по которой будем осуществлять поиск на сервисе. Укажем в ней «Russia», чтобы получить все пивоварни, связанные с Россией. Другой параметр — offset — отвечает за смещение. Получив первые 50 пивоварен мы смещаемся на 50 строк в поиске, чтобы получить следующие 50 пивоварен. limit отвечает за количество получаемых пивоварен и не может быть больше 50.
Преобразовываем ответ в формат json и добавляем полученные данные в список all_brewery_of_russia. Объект item будет содержать такие данные:

Но в полученных данных могли затесаться и пивоварни других стран. Отфильтруем их: пройдём итератором по всему списку all_brewery_of_russia и добавим в итоговый только те пивоварни, у которых параметр country_name принимает значение Russia.

brew_list = []
for element in all_brewery_of_russia:
    brew = element['response']['brewery']
    for i in range(brew['count']):
        if brew['items'][i]['brewery']['country_name'] == 'Russia':
            brew_list.append(brew['items'][i])

Посмотрим на первый элемент списка brew_list:

print(brew_list[0])

Соберём из списка DataFrame с колонками brewery_id, beer_count, brewery_name, brewery_slug, brewery_page_url, brewery_city, lat и  lng. Получим в отдельные списки данные из  brewery_list:

df = pd.DataFrame()
brewery_id_list = []
beer_count_list = []
brewery_name_list = []
brewery_slug_list = []
brewery_page_url_list = []
brewery_location_city = []
brewery_location_lat = []
brewery_location_lng = []
for brewery in brew_list:
    brewery_id_list.append(brewery['brewery']['brewery_id'])
    beer_count_list.append(brewery['brewery']['beer_count'])
    brewery_name_list.append(brewery['brewery']['brewery_name'])
    brewery_slug_list.append(brewery['brewery']['brewery_slug'])
    brewery_page_url_list.append(brewery['brewery']['brewery_page_url'])
 brewery_location_city.append(brewery['brewery']['location']['brewery_city'])
    brewery_location_lat.append(brewery['brewery']['location']['lat'])
    brewery_location_lng.append(brewery['brewery']['location']['lng'])

И отправим их в DataFrame:

df['brewery_id'] = brewery_id_list
df['beer_count'] = beer_count_list
df['brewery_name'] = brewery_name_list
df['brewery_slug'] = brewery_slug_list
df['brewery_page_url'] = brewery_page_url_list
df['brewery_city'] = brewery_location_city
df['brewery_lat'] = brewery_location_lat
df['brewery_lng'] = brewery_location_lng

Посмотрим, как выглядит наша таблица:

df.head()

Отсортируем значения по  brewery_id и выгрузим таблицу в формате csv без столбца с индексами и заголовков колонок:

df = df.sort_values(by='brewery_id')
df.to_csv('brewery_data.csv', index=False, header=False)

Создаём словарь Clickhouse

Словари для Clickhouse можно создавать по-разному. Мы попробуем задать его структуру в xml-файле, настроить конфигурационные файлы сервера и обращаться к нему через клиент. Наш xml будет иметь следующую структуру:

Со всеми способами создания словарей можно ознакомиться в документации

<yandex>
<dictionary>
        <name>breweries</name>
        <source>
                <file>
                        <path>/home/ubuntu/brewery_data.csv</path>
                        <format>CSV</format>
                </file>
        </source>
        <layout>
                <flat />
        </layout>
        <structure>
                <id>
                        <name>brewery_id</name>
                </id>
                <attribute>
                        <name>beer_count</name>
                        <type>UInt64</type>
                        <null_value>Null</null_value>
                </attribute>
                <attribute>
                        <name>brewery_name</name>
                        <type>String</type>
                        <null_value>Null</null_value>
                </attribute>
                <attribute>
                        <name>brewery_slug</name>
                        <type>String</type>
                        <null_value>Null</null_value>
                </attribute>
                <attribute>
                        <name>brewery_page_url</name>
                        <type>String</type>
                        <null_value>Null</null_value>
                </attribute>
                <attribute>
                        <name>brewery_city</name>
                        <type>String</type>
                        <null_value>Null</null_value>
                </attribute>
                <attribute>
                        <name>lat</name>
                        <type>String</type>
                        <null_value>Null</null_value>
                </attribute>
                <attribute>
                        <name>lng</name>
                        <type>String</type>
                        <null_value>Null</null_value>
                </attribute>
        </structure>
        <lifetime>300</lifetime>
</dictionary>
</yandex>

Под  идёт имя словаря. В  указываем свойства колонок. Под тегом идёт ключевое поле, а под тегом укажем путь и формат файла. Скоро мы положим его в папку /home/ubuntu, поэтому так и укажем.

Загрузим нашу csv-таблицу и xml-файл на сервер, это можно сделать, например, по ftp через FileZilla. В одном из материалов мы учились ставить Clickhouse на бесплатную машину от Amazon, в этот раз будем работать там же. В FileZilla заходим в настройки SFTP и добавляем файл с ключом:

И подключаемся к серверу по адресу, который указан в консоли EC2 машины на AWS. Укажем протокол SFTP, свой Host и в качестве User — Ubuntu:

В случае перезагрузки машины через консоль Public DNS мог измениться

После подсоединения мы попадём в папку /home/ubuntu сервера. Положим файлы туда же. Теперь подключимся по SSH через Termius. Чтобы Clickhouse увидел файл со структурой словаря, его нужно положить в папку /etc/clickhouse-server:

О том, как подключаться в серверу на AWS через SSH-клиент мы рассказывали в материале «Устанавливаем Clickhouse на AWS»

sudo mv breweries_dictionary.xml /etc/clickhouse server/

Переходим в конфигурационный файл:

cd /etc/clickhouse-server
sudo nano config.xml

Нам нужен тег  — он указывает путь к файлу, который описывает структуру словарей. Укажем путь к нашему xml:

<dictionaries_config>/etc/clickhouse-server/breweries_dictionary.xml</dictionaries_config>

Сохраняем файл и запускаем клиент Clickhouse:

clickhouse client

Проверим, что наш словарь действительно загрузился:

SELECT * FROM system.dictionaries\G

В случае успеха получим подобное:

Напишем запрос к функции dictGet, чтобы получить название пивоварни под ID 999. Указываем первым аргументом наименование словаря, затем поле, значение которого хотим получить и ID.

SELECT dictGet('breweries', 'brewery_name', toUInt64(999))

Если сделаем всё правильно, то выясним, что под ID 999 находится Балтика:

Аналогичным образом удобно использовать функцию, когда в таблице с фактами хранится только ID измерения для получения понятного наименования.

 Нет комментариев    95   22 дн   Amazon Web Services   clickhouse   Data analytics   Data engineering   python
 Нет комментариев    72   1 мес   Data engineering   sql

Собираем данные по рекламным кампаниям ВКонтакте

В пятничном лонгриде проделаем большую работу: возьмём информацию по рекламным кампаниям ВКонтакте и сопоставим их с данными Google Analytics в Redash. Чтобы снова не поднимать сервер, будем передавать данные через Google Docs, используя Spreadsheet API.

Получение access token
Для получение пользовательского ключа ВКонтакте нужно создать приложение. Идём в раздел «Разработчики» по https://vk.com/apps?act=manage, жмём на кнопку «Создать приложение». В поле «Тип приложения» выбираем «Standalone-приложение» и даём любое название. После этого в меню слева идём в настройки и сохраняем себе ID приложения.

Актуальную информацию о ключах можно посмотреть в статье «Получение ключа доступа»

Теперь копируем себе эту ссылку:

https://oauth.vk.com/authorize?client_id=YourClientID&scope=ads&response_type=token

Но вместо YourClientID вставляем ID своего созданного приложения. В scope у этой ссылки только ads, так что с этим ключом можно будет получать только информацию о рекламном кабинете. Вставляем её в браузер и нас скидывает на другую страницу — в адресе этой странице будет указан ваш сгенерированный access token.

Срок жизни токена — 86400 секунд: ровно сутки. Чтобы получить токен без временных ограничений можно добавить в scope параметр offline. Если токен понадобилось отозвать — смените пароль от страницы или в настройках безопасности завершите активные сессии.

Ещё для запросов к API нам пригодится ID рекламного кабинета — проходим по https://vk.com/ads?act=settings и копируем «номер кабинета».

Сбор данных через запросы к API
Напишем скрипт, который обращается к серверу ВКонтакте с нашим access token и номером рекламного кабинета и берёт информацию о всех кампаниях пользователя: количество просмотров на рекламах, кликов и затрат. Затем скрипт будет формировать из него DataFrame и отправлять в Google Docs.

from oauth2client.service_account import ServiceAccountCredentials
from pandas import DataFrame
import requests
import gspread
import time

Зададим несколько константных значений: access token, ID рекламного кабинета и версию API ВКонтакте, которую будем использовать. Актуальной является версия 5.103.

token = 'fa258683fd418fafcab1fb1d41da4ec6cc62f60e152a63140c130a730829b1e0bc'
version = 5.103
id_rk = 123456789

За получение статистики по рекламе отвечает метод ads.getStatistics, но один из обязательных параметров при его вызове — ’ids’, ID рекламного объявления, статистику по которому мы хотим получить. Так как ID у нас пока нет, придётся сначала воспользоваться методов ads.getAds, который возвращает ID объявлений и кампаний.

Подробнее со всеми методами ВКонтакте API можно ознакомиться в документации

Библиотекой requests отправляем запрос к серверу и передаём свои параметры. Полученный ответ сразу переведём в формат json


campaign_ids = []
ads_ids = []
r = requests.get('https://api.vk.com/method/ads.getAds', params={
    'access_token': token,
    'v': version,
    'account_id': id_rk
})
data = r.json()['response']

Вот, как выглядит объект data: нам вернулся обычный список словарей, с которым мы уже имели дело в материале “Передаём и анализируем собранные данные по рекламным капманиям в Redash”.

Заполняем словарь ad_campaign_dict. Ключом будет ID объявления, а значением — ID кампании, к которой принадлежит объявление. Так будет удобнее присваивать к объявлению ID кампании, к которой оно принадлежало.

ad_campaign_dict = {}
for i in range(len(data)):
    ad_campaign_dict[data[i]['id']] = data[i]['campaign_id']

Теперь, имея ID каждого нужного объявления, можно обратиться к методу ads.getStatistics. Мы будем собирать количество просмотров, кликов, затрат и даты начала и конца объявления, поэтому заблаговременно заведём пустые списки.

ads_campaign_list = []
ads_id_list = []
ads_impressions_list = []
ads_clicks_list = []
ads_spent_list = []
ads_day_start_list = []
ads_day_end_list = []

Вызывать getStatistics нужно отдельно для каждого объявления — будем делать это в итераторе по ad_campaign_dict. Отправляем запрос, передавая в ‘period’ значение ‘overall’ — берём данные за всё время. У некоторых объявлений могут отсутствовать данные по полю «Просмотры» или «Клики» если они не были запущены, и, потребовав их, мы словим KeyError — во избежание этого добавим обработчик try — except, который заставит скрипт не обращать внимания на эту ошибку.

for ad_id in ad_campaign_dict:
        r = requests.get('https://api.vk.com/method/ads.getStatistics', params={
            'access_token': token,
            'v': version,
            'account_id': id_rk,
            'ids_type': 'ad',
            'ids': ad_id,
            'period': 'overall',
            'date_from': '0',
            'date_to': '0'
        })
        try:
            data_stats = r.json()['response']
            for i in range(len(data_stats)):
                for j in range(len(data_stats[i]['stats'])):
                    ads_impressions_list.append(data_stats[i]['stats'][j]['impressions'])
                    ads_clicks_list.append(data_stats[i]['stats'][j]['clicks'])
                    ads_spent_list.append(data_stats[i]['stats'][j]['spent'])
                    ads_day_start_list.append(data_stats[i]['stats'][j]['day_from'])
                    ads_day_end_list.append(data_stats[i]['stats'][j]['day_to'])
                    ads_id_list.append(data_stats[i]['id'])
                    ads_campaign_list.append(ad_campaign_dict[ad_id])
        except KeyError:
            continue

Теперь сформируем из списков DataFrame и выведем первые 5 элементов:

df = DataFrame()
df['campaign_id'] = ads_campaign_list
df['ad_id'] = ads_id_list
df['impressions'] = ads_impressions_list
df['clicks'] = ads_clicks_list
df['spent'] = ads_spent_list
df['day_start'] = ads_day_start_list
df['day_end'] = ads_day_end_list
print(df.head())

Экспорт данных в Google Docs
Для экспорта DataFrame в таблицу Google Sheets необходим ключ доступа Google API. Пройдём по https://console.developers.google.com и создадим новый проект. Даём ему любое имя и в Dashboard жмём на кнопку “Подключить API и сервисы”. Нужно включить два API — Google Drive API и Google Sheets API. Ищем первый в поиске, нажимаем на “Включить API”, затем ищем второй и проделываем то же самое.

После включения нас отправят на панель управления API. Жмём на «Создать учётные данные» — по ним будем проводить авторизацию в скрипте. Отмечаем, что используем Google Sheets API из веб-сервера и обращаемся к данным пользователя. Нажимаем на «Выбрать тип учётных данных» и создаем сервисный аккаунт. В поле «Роль» выбираем Проект — Редактор, а тип ключа оставим JSON.

После этого нам отправят файл в формате JSON с нашими учетными данными — назовём его «credentials.json» — и перенаправят на страницу с сервисными аккаунтами. Ниже будет поле с почтой — копируем её себе.

Переходим по https://docs.google.com/spreadsheets и создаем пустой файл с названием data, в который будут отправляться данные из DataFrame. В настройках доступа даём доступ по почте, скопированной ранее из сервисных аккаунтов — от неё будут приходить данные из скрипта.

Закинем файл credentials.json в директорию со скриптом и продолжим писать код. Перечисляем область видимости в виде ссылок:

scope = ['https://spreadsheets.google.com/feeds', 'https://www.googleapis.com/auth/drive']

И при помощи библиотек oauth2client и gspread проводим авторизацию методами ServiceAccountCredentials.from_json_keyfile_name и gspread.authorize, указывая в параметрах первого наш файл и переменную scope. Через переменную sheet будем обращаться к нашему файлу в Google Docs.

creds = ServiceAccountCredentials.from_json_keyfile_name('credentials.json', scope)
client = gspread.authorize(creds)
sheet = client.open('data').sheet1

Для ввода значений в ячейку таблички есть метод update_cell. Важно: нумерация индексов ячеек при обращении начинается не с нуля, а с единицы. Первым циклом пройдём по первой строке и перенесем туда заголовки нашего DataFrame. Во втором будем идти по каждой ячейке и вставлять соответствующие значения DataFrame. По умолчанию стоит ограничение — 100 запросов в 100 секунд. Это ограничение может остановить наш скрипт на полпути: чтобы избежать ошибки пропишем time.sleep, чтобы после каждой вставки скрипт секунду выжидал.

count_of_rows = len(df)
count_of_columns = len(df.columns)
for i in range(count_of_columns):
    sheet.update_cell(1, i + 1, list(df.columns)[i])
for i in range(1, count_of_rows + 1):
    for j in range(count_of_columns):
        sheet.update_cell(i + 1, j + 1, str(df.iloc[i, j]))
        time.sleep(1)

Если всё сделаем правильно — получим таблицу такого вида:

Экспорт данных в Redash

Подключение Google Analytics к Redash описано в статье «Как подключить Google Analytics как Redash?».

Имея в Redash таблицу с Google Analytics и рекламным кампаниям ВКонтакте, можем сопоставить их друг другу. Напишем такой запрос:

SELECT
    query_50.day_start,
    CASE WHEN ga_source LIKE '%vk%' THEN 'vk.com' END AS source,
    query_50.spent,
    query_50.impressions,
    query_50.clicks,
    SUM(query_49.ga_sessions) AS sessions,
    SUM(query_49.ga_newUsers) AS users
FROM query_49
JOIN query_50
ON query_49.ga_date = query_50.day_start
WHERE query_49.ga_source LIKE '%vk%' AND DATE(query_49.ga_date) BETWEEN '2020-05-16' AND '2020-05-20'
GROUP BY query_49.ga_date, source

ga_source — источник, с которого человек пришел на сайт. Всё, что похоже на vk оператором CASE объединяем в столбец «vk.com». Оператором JOIN добавляем таблицу с данными из ВКонтакте, объединяя по полю даты. Отсеиваем данные — возьмём день последней рекламной кампании и посмотрим на несколько дней после него. На выходе получим таблицу такого вида:

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

Ранее Ctrl + ↓