Анализ рынка вакансий аналитики и BI: дашборд в Tableau

Как мы собирали данные?

Схема данных основана на коротком представлении вакансии, которую возвращает метод GET /vacancies. Из представления собираются следующие поля: тип вакансии, идентификатор, премиальность вакансии, необходимость прохождения тестирования, адрес компании, информация о зарплате, график работы и другие. Соответствующий CREATE-запрос для таблицы:

CREATE TABLE headhunter.vacancies_short  (      `added_at` DateTime,      `query_string` String,      `type` String,      `level` String,      `direction` String,      `vacancy_id` UInt64,      `premium` UInt8,      `has_test` UInt8,      `response_url` String,      `address_city` String,      `address_street` String,      `address_building` String,      `address_description` String,      `address_lat` String,      `address_lng` String,      `address_raw` String,      `address_metro_stations` String,      `alternate_url` String,      `apply_alternate_url` String,      `department_id` String,      `department_name` String,      `salary_from` Nullable(Float64),      `salary_to` Nullable(Float64),      `salary_currency` String,      `salary_gross` Nullable(UInt8),      `name` String,      `insider_interview_id` Nullable(UInt64),      `insider_interview_url` String,      `area_url` String,      `area_id` UInt64,      `area_name` String,      `url` String,      `published_at` DateTime,      `employer_url` String,      `employer_alternate_url` String,      `employer_logo_urls_90` String,      `employer_logo_urls_240` String,      `employer_logo_urls_original` String,      `employer_name` String,      `employer_id` UInt64,      `response_letter_required` UInt8,      `type_id` String,      `type_name` String,      `archived` UInt8,      `schedule_id` Nullable(String)  )  ENGINE = ReplacingMergeTree  ORDER BY vacancy_id

Первый скрипт собирает данные с HeadHunter по API и отправляет их в Clickhouse. Он использует следующие библиотеки:

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

Далее загружаем таблицу с запросами и подключаемся к CH:

queries = pd.read_csv('hh_data.csv')  client = Client(host='1.234.567.890', user='default', password='', port='9000', database='headhunter')

Таблица queries хранит список поисковых запросов. Она содержит следующие колонки: тип запроса, уровень вакансии для поиска, направление вакансии и саму поисковую фразу. В строку с запросом можно помещать логические операторы: например, чтобы найти вакансии, в которых должны присутствовать ключевые слова «Python», «data» и «анализ» между ними можно указать логическое «И».

Не всегда вакансии в выдаче соответствуют ожиданиям: случайно в базу могут попасть повара, маркетологи и администраторы магазина. Чтобы этого не произошло, опишем функцию check_name(name) — она будет принимать наименование вакансии и возвращать True в случае, если вакансия не подошла по названию.

def check_name(name):      bad_names = [r'курьер', r'грузчик', r'врач', r'менеджер по закупу',             r'менеджер по продажам', r'оператор', r'повар', r'продавец',            r'директор магазина', r'директор по продажам', r'директор по маркетингу',            r'кабельщик', r'начальник отдела продаж', r'заместитель', r'администратор магазина',             r'категорийный', r'аудитор', r'юрист', r'контент', r'супервайзер', r'стажер-ученик',             r'су-шеф', r'маркетолог$', r'региональный', r'ревизор', r'экономист', r'ветеринар',             r'торговый', r'клиентский', r'начальник цеха', r'территориальный', r'переводчик',             r'маркетолог /', r'маркетолог по']      for item in bad_names:          if re.match(item, name):              return True

Затем объявляем бесконечный цикл — мы собираем данные без перерыва. Идём по DataFrame queries и сразу забираем оттуда тип вакансии, уровень, направление и поисковый запрос в отдельные переменные. Сначала по ключевому слову отправляем один запрос к методу /GET vacancies и получаем количество страниц. После идём от нулевой до последней страницы, отправляем те же запросы и заполняем список vacancies_from_response с полученными в выдаче короткими представлениями всех вакансий. В параметрах указываем 10 вакансий на страницу — больше ограничения HH API получить не позволяют. Так как мы не указали параметр area, API возвращает вакансии по всему миру.

while True:     for query_type, level, direction, query_string in zip(queries['Тип'], queries['Уровень'], queries['Направление'], queries['Ключевое слово']):             print(f'ключевое слово: {query_string}')             url = 'https://api.hh.ru/vacancies'             par = {'text': query_string, 'per_page':'10', 'page':0}             r = requests.get(url, params=par).json()             added_at = datetime.now().strftime('%Y-%m-%d %H:%M:%S')             pages = r['pages']             found = r['found']             vacancies_from_response = []               for i in range(0, pages + 1):                 par = {'text': query_string, 'per_page':'10', 'page':i}                 r = requests.get(url, params=par).json()                 try:                     vacancies_from_response.append(r['items'])                 except Exception as E:                     continue

Теперь проходим по каждой вакансии на каждой странице двойным итератором. Сперва отправим запрос к Clickhouse и проверим, нет ли уже в базе вакансии с таким идентификатором и таким поисковым запросом. Если проверка пройдена — проверяем название вакансии. В случае неудачи переходим к следующей.

for item in vacancies_from_response:                 for vacancy in item:                     if client.execute(f"SELECT count(1) FROM vacancies_short WHERE vacancy_id={vacancy['id']} AND query_string='{query_string}'")[0][0] == 0:                         name = vacancy['name'].replace("'","").replace('"','')                         if check_name(name):                             continue

Теперь проходим по вакансии и собираем все нужные поля. В случае отсутствия некоторых данных будем отправлять пустые строки:

vacancy_id = vacancy['id']                         is_premium = int(vacancy['premium'])                         has_test = int(vacancy['has_test'])                         response_url = vacancy['response_url']                         try:                             address_city = vacancy['address']['city']                             address_street = vacancy['address']['street']                             address_building = vacancy['address']['building']                             address_description = vacancy['address']['description']                             address_lat = vacancy['address']['lat']                             address_lng = vacancy['address']['lng']                             address_raw = vacancy['address']['raw']                             address_metro_stations = str(vacancy['address']['metro_stations']).replace("'",'"')                         except TypeError:                             address_city = ""                             address_street = ""                             address_building = ""                             address_description = ""                             address_lat = ""                             address_lng = ""                             address_raw = ""                             address_metro_stations = ""                         alternate_url = vacancy['alternate_url']                         apply_alternate_url = vacancy['apply_alternate_url']                         try:                             department_id = vacancy['department']['id']                         except TypeError as E:                             department_id = ""                         try:                             department_name = vacancy['department']['name']                         except TypeError as E:                             department_name = ""                         try:                             salary_from = vacancy['salary']['from']                         except TypeError as E:                             salary_from = "cast(Null as Nullable(UInt64))"                         try:                             salary_to = vacancy['salary']['to']                         except TypeError as E:                             salary_to = "cast(Null as Nullable(UInt64))"                         try:                             salary_currency = vacancy['salary']['currency']                         except TypeError as E:                             salary_currency = ""                         try:                             salary_gross = int(vacancy['salary']['gross'])                         except TypeError as E:                             salary_gross = "cast(Null as Nullable(UInt8))"                         try:                             insider_interview_id = vacancy['insider_interview']['id']                         except TypeError:                             insider_interview_id = "cast(Null as Nullable(UInt64))"                         try:                             insider_interview_url = vacancy['insider_interview']['url']                         except TypeError:                             insider_interview_url = ""                         area_url = vacancy['area']['url']                         area_id = vacancy['area']['id']                         area_name = vacancy['area']['name']                         url = vacancy['url']                         published_at = vacancy['published_at']                         published_at = datetime.strptime(published_at,'%Y-%m-%dT%H:%M:%S%z').strftime('%Y-%m-%d %H:%M:%S')                         try:                             employer_url = vacancy['employer']['url']                         except Exception as E:                             print(E)                             employer_url = ""                         try:                             employer_alternate_url = vacancy['employer']['alternate_url']                         except Exception as E:                             print(E)                             employer_alternate_url = ""                         try:                             employer_logo_urls_90 = vacancy['employer']['logo_urls']['90']                             employer_logo_urls_240 = vacancy['employer']['logo_urls']['240']                             employer_logo_urls_original = vacancy['employer']['logo_urls']['original']                         except Exception as E:                             print(E)                             employer_logo_urls_90 = ""                             employer_logo_urls_240 = ""                             employer_logo_urls_original = ""                         employer_name = vacancy['employer']['name'].replace("'","").replace('"','')                         try:                             employer_id = vacancy['employer']['id']                         except Exception as E:                             print(E)                         response_letter_required = int(vacancy['response_letter_required'])                         type_id = vacancy['type']['id']                         type_name = vacancy['type']['name']                         is_archived = int(vacancy['archived'])

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

try:      schedule = vacancy['schedule']['id']  except Exception as E:      print(E)      schedule = ''"  if schedule == 'flyInFlyOut':      continue

Теперь формируем список из полученных переменных, заменяем в нём None-значения на пустые строки во избежании конфликтов с Clickhouse и вставляем строку в таблицу.

vacancies_short_list = [added_at, query_string, query_type, level, direction, vacancy_id, is_premium, has_test, response_url, address_city, address_street, address_building, address_description, address_lat, address_lng, address_raw, address_metro_stations, alternate_url, apply_alternate_url, department_id, department_name,  salary_from, salary_to, salary_currency, salary_gross, insider_interview_id, insider_interview_url, area_url, area_name, url, published_at, employer_url, employer_logo_urls_90, employer_logo_urls_240,  employer_name, employer_id, response_letter_required, type_id, type_name, is_archived, schedule]  for index, item in enumerate(vacancies_short_list):      if item is None:          vacancies_short_list[index] = ""  tuple_to_insert = tuple(vacancies_short_list)  print(tuple_to_insert)  client.execute(f'INSERT INTO vacancies_short VALUES {tuple_to_insert}')

Как подключили Tableau к данным?

Tableau Public не умеет работать с базами данных, поэтому мы написали коннектор Clickhouse к Google Sheets. Он использует библиотеки gspread и oauth2client для авторизации в Google Spreadsheets API и библиотеку schedule для ежедневной работы по графику.

Работа с Google Spreadseets API подробно разобрана в материале «Собираем данные по рекламным кампаниям ВКонтакте»

import schedule  from clickhouse_driver import Client  import gspread  import pandas as pd  from oauth2client.service_account import ServiceAccountCredentials  from datetime import datetime    scope = ['https://spreadsheets.google.com/feeds', 'https://www.googleapis.com/auth/drive']  client = Client(host='54.227.137.142', user='default', password='', port='9000', database='headhunter')  creds = ServiceAccountCredentials.from_json_keyfile_name('credentials.json', scope)  gc = gspread.authorize(creds)

Опишем функцию update_sheet() — она будет брать все данные из Clickhouse и вставлять их в таблицу Google Docs.

def update_sheet():     print('Updating cell at', datetime.now())     columns = []     for item in client.execute('describe table headhunter.vacancies_short'):         columns.append(item[0])     vacancies = client.execute('SELECT * FROM headhunter.vacancies_short')     df_vacancies = pd.DataFrame(vacancies, columns=columns)     df_vacancies.to_csv('vacancies_short.csv', index=False)     content = open('vacancies_short.csv', 'r').read()     gc.import_csv('1ZWS2kqraPa4i72hzp0noU02SrYVo0teD7KZ0c3hl-UI', content.encode('utf-8'))

Чтобы скрипт запускался в 16:00 по МСК каждый день используем библиотеку schedule:

schedule.every().day.at("13:00").do(update_sheet)  while True:     schedule.run_pending()

А что в результате?

Рома построил на полученных данных дашборд.

И в youtube-ролике рассказывает о том, как эффективно использовать дашборд

Инсайты, которые можно извлечь из дашборда

1. Аналитики с навыком бизнес-аналитики востребованы на рынке больше всего: по такому запросу нашлось больше всего вакансий. Тем не менее, средняя зарплата выше у продуктовых аналитиков и аналитиков BI.
2. В Москве средние зарплаты выше на 10-30 тысяч рублей, чем в Санкт-Петербурге и на 30-40 тысячи рублей, чем в регионах. Там же работы нашлось больше всего в России.
3. Самые высокооплачиваемые должности: руководитель отдела аналитики (в среднем, 110 тыс. руб. в месяц), инженер баз данных (138 тыс. руб. в месяц) и директор по машинному обучению (250 тыс. руб. в месяц).
4. Самые полезные навыки на рынке — владение Python c библиотеками pandas и numpy, Tableau, Power BI, Etl и Spark. Вакансий с такими требованиями больше и зарплаты в них указаны выше прочих. Для Python-программистов знание matplotlib ценится на рынке выше, чем владение plotly.

Полный код проекта доступен на GitHub