Привет! Меня зовут Григорий Митраков, я BI-аналитик в рекламном агентстве.
В статье расскажу вам как создать приложение в Power BI по пожеланиям заказчиков.
Предыстория
После того, как я устроился работать в агентство ко мне обратилась руководитель HR отдела для разработки небольшого автоматизированного отчета в BI. До этого основные показатели (текучесть, вовлеченность и др.) рассчитывались в табличках Excel полу ручным способом.
Так выглядели отчеты в Excel файлах:
Причину автоматизации мне объяснили тем, что около 4-х часов в неделю сотрудник HR отдела тратил на расчет показателей в таблицах Excel. (Что хорошо – видно, что какими-то графиками и диаграммами уже пользовались).
Сбор требований к отчету
После первой встречи с HR отделом в лице руководителя, специалиста, занимавшегося сбором данных и оформлением отчетов в таблицах Excel, выяснилось, что источники с необходимыми данными находятся в регулярных выгрузках из 1С Предприятие, Google Sheets, Битрикс24, PostgreSQL, iSpring. По визуальному оформлению как таковых «жестких» требований не было, была лишь просьба, учесть наглядность представляемых данных и «желательно как в таблицах Excel».
Работа над дашбордом
Задача показалась мне интересной ввиду достаточно большого количества разнообразных источников данных и свободы в выборе визуализации данных.
Для прозрачности работы, контроля текущего статуса над задачей и внесение своевременных правок в оформление и расчет метрик решили собираться раз в неделю по четвергам (этот день назвали «День HR»).
Для всех источников данных решил использовать БД PostgreSQL (была развернута в компании). Для разработки ETL-процессов — Apache Airflow.
Ниже представлена схема ETL процесса:
Данные раз в сутки (по расписанию) загружаю:
с помощью API REST с опросников (google sheets),
с помощью API REST с платформы для обучения сотрудников iSpring,
с Битрикс24 веб-скрейпингом (у Битрикс24 есть за дополнительную плату расширение BI-Аналитика, но в компании бесплатная версия, поэтому получение данных организовано таким образом),
с сетевых папок, в которых хранятся регулярные выгрузки в формате .xlsx из 1С.
Пример DAG (таблица с курсами iSpring):
import json
import time
import datetime
import requests
import pandas as pd
from airflow.decorators import dag, task
from airflow.models import Variable
from sqlalchemy import create_engine
from airflow.providers.telegram.hooks.telegram import TelegramHook
# Параметры для аутентификации и входа
user = 'grigoriy'
host = '101.100.9.43'
db = 'internal_data'
pwd = Variable.get('planning_datas_password')
postgresql_url = f'postgresql+psycopg2://{user}:{pwd}@{host}/{db}'
engine = create_engine(postgresql_url)
# запрос к БД для получения id различных курсов
query_db = '''SELECT t.module_id, t.content_item_id, t.course_id
FROM ispring_course_modules_table AS t
GROUP BY t.module_id, t.content_item_id, t.course_id'''
URL_TOKEN = 'https://api-learn.ispringlearn.ru/api/v3/token'
URL_CONTENT = 'https://api-learn.ispringlearn.ru/courses/modules'
HEADER = {
'Content-Type': 'application/x-www-form-urlencoded',
'Accept': 'application/json'
}
HEADERS_R = {
'X-Target-Locale': 'en-US',
'Accept': 'application/json'
}
CLIENT_SECRET = Variable.get('ispring_api_client_secret')
DATA_URLENCODE = {
'client_id': 'xxbbaaxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx',
'client_secret': CLIENT_SECRET,
'grant_type': 'client_credentials'
}
default_args = {'owner': 'xxxxxxx' }
TELEGRAM_TOKEN = Variable.get('tg_bot_token')
CHAT_ID = '0000000000'
# Необходимые функции:
def error_to_telegram(token: str, chat_id: str, e, name_dag: str):
'''
Функция отправляет сообщения об ошибках в чат телеграмм
:param: token: "xxx:xxx" API токен для telegram telegram_token
:param: chat_id куда отправлять сообщение
:param: e - ошибка
:param: name_dag - название DAG
'''
telegram_conn_id = 'telegram_default'
telegram_hook = TelegramHook(telegram_conn_id,
token,
chat_id)
t = datetime.datetime.now()
date_and_time = str(t.date()) + ' ' + str(t.time())
message = f'🛑 В *DAG {name_dag}* произошла ошибка {e}'
message_text = f'{message}. Время ошибки: {date_and_time}'
telegram_hook.send_message({
'text': message_text,
'disable_notification': True
})
@dag(default_args=default_args,
schedule_interval='30 12 * * *',
start_date=datetime.datetime(2023, 1, 1),
catchup=False
)
def get_data_course_modules_from_ispring():
@task
def get_id_courses_db(query: str, engine) -> pd.DataFrame:
"""
Функция получает список различных id курсов из БД
:param query: запрос к БД
:return: список
"""
dis_id_db = pd.read_sql(query, engine)
return dis_id_db
@task
def get_token_ispring(url_token: str,
header: dict,
data_urlcode: dict,
telegram_token: str,
id_chat: str) -> str:
'''
Функция получает токен
:param: url_token - url
:param: header - headers
:param: data_urlcode - params
:return: token
'''
res = requests.post(url=url_token,
headers=header,
data=data_urlcode)
if res.status_code == 200:
# Результаты получения токена:
res_token = json.loads(res.text)
# Токен для доступа:
authorization_token = res_token['access_token']
return authorization_token
else:
error_to_telegram(token=telegram_token,
chat_id=id_chat,
e=str(res.status_code) + ' ' + str(res.text),
name_dag='get_data_course_modules_from_ispring')
@task
def get_data_courses(url_content: str,
header_r: dict,
autho_token: str,
id_courses_t: pd.DataFrame,
telegram_token: str,
id_chat: str) -> pd.DataFrame:
# Присвоим токен к заголовку:
header_r['Authorization'] = autho_token
result = requests.get(url=url_content, headers=header_r)
if result.status_code == 200:
# Таблица со списком курсов:
course_content = pd.DataFrame(json.loads(result.text)["modules"])
# Переименуем столбцы:
course_content = course_content.rename(columns={'moduleId': 'module_id',
'contentItemId': 'content_item_id',
'courseId': 'course_id',
'authorId': 'author_id',
'addedDate': 'added_date',
'viewUrl': 'view_url'})
# Приведем в необходимый тип данные:
course_content['added_date'] = pd.to_datetime(course_content['added_date'], utc=True).dt.tz_localize(None).astype('datetime64[ns]')
course_content = course_content.astype({'module_id': str,
'content_item_id': str,
'course_id': str,
'title': str,
'description': str,
'author_id': str,
'added_date': 'datetime64[ns]',
'view_url': str})
# Выберем необходимые столбцы:
cols = ['module_id', 'content_item_id', 'course_id', 'title', 'description',
'author_id', 'added_date', 'view_url']
course_content = course_content[cols]
return course_content[~((course_content['module_id'].isin(id_courses_t['module_id']))
& (course_content['content_item_id'].isin(id_courses_t['content_item_id']))
& (course_content['course_id'].isin(id_courses_t['course_id'])))].reset_index(drop=True)
else:
error_to_telegram(token=telegram_token,
chat_id=id_chat,
e=str(result.status_code) + ' ' + str(result.text),
name_dag='get_data_course_modules_from_ispring')
@task
def append_data_to_db(data: pd.DataFrame, table: str, engine):
"""
Функция добавляет данные из датафрейма в БД
:param data: датафрейм
:param table: имя таблицы
:return: None
"""
for i in range(len(data) // 10000 + 1):
data.iloc[i * 10000: (i + 1) * 10000].to_sql(con=engine,
name=table,
if_exists='append',
index=False)
time.sleep(3)
# Загрузка таблицы со списком курсов в БД
dis_id_db = get_id_courses_db(query=query_db, engine=engine)
token_ispring = get_token_ispring(url_token=URL_TOKEN,
header=HEADER,
data_urlcode=DATA_URLENCODE,
telegram_token=TELEGRAM_TOKEN,
id_chat=CHAT_ID)
df = get_data_courses(url_content=URL_CONTENT,
header_r=HEADERS_R,
autho_token=token_ispring,
id_courses_t=dis_id_db,
telegram_token=TELEGRAM_TOKEN,
id_chat=CHAT_ID)
append_data_to_db(data=df, table='ispring_course_modules_table', engine=engine)
get_data_course_modules_from_ispring = get_data_course_modules_from_ispring()
С каждого Python модуля получаю алерты в Телеграм-чат, в случае, если загрузка прошла неудачно.
Пример такого алерта:
Из БД PostgreSQL получаю данные в дашборды, некоторые таблички получаю с вьюшек, вьюшки с материализованных представлений. Схема в БД:
Финальная обработка данных осуществляется в Power Query.
Разработку начали с важных метрик, двигаясь постепенно к второстепенным по мере выполнения отчетов.
Первый дашборд представлял метрику «Текучесть в компании» (ниже представлены первые 2 страницы):
Модель данных получилась такая:
Второй дашборд представлял метрику «Вовлеченность в компании» (ниже представлены первые 2 страницы):
Модель данных получилась такая:
Следующий дашборд по оценке «Внутреннего NPS» - считается по формуле: (кол-во положительных оценок – кол-во отрицательных оценок) / общее кол-во оценок. Этот дашборд показывает количество и оценку внутренних задач в компании, выполняемых отделами или сотрудниками и оцениваемых постановщиками задач.
Ниже представлены первые две страницы:
Модель данных:
Следующий дашборд по метрике «Изменения грейдов сотрудников».
Ниже представлены первые две страницы:
Модель данных:
Следующий дашборд по метрике «Адаптация сотрудников в компании».
Ниже представлены первые две страницы дашборда:
Модель данных:
Следующий дашборд показывает результаты прохождения сотрудниками компании курсов на iSpring:
Модель данных:
Итоговый дашборд включает объединение предыдущих дашбордов в один: «Карта здоровья».
Получился такой:
Данные в дашборд загружаются через Центр данных Одного озера подключением к семантическим моделям предыдущих дашбордов. Показатели на «Карте здоровья» подсвечиваются цветами, в зависимости от значений.
Общая схема источников данных:
Заключение
Все эти дашборды объединил в приложение, которым удобно пользоваться, так как основные показатели выведены в одном окне. При необходимости легко можно просмотреть более детально по каждому показателю.
