Привет! Меня зовут Григорий Митраков, я 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:

Модель данных:

Итоговый дашборд включает объединение предыдущих дашбордов в один: «Карта здоровья».

Получился такой:

Данные в дашборд загружаются через Центр данных Одного озера подключением к семантическим моделям предыдущих дашбордов. Показатели на «Карте здоровья» подсвечиваются цветами, в зависимости от значений.

Общая схема источников данных:

Заключение

Все эти дашборды объединил в приложение, которым удобно пользоваться, так как основные показатели выведены в одном окне. При необходимости легко можно просмотреть более детально по каждому показателю.