Развертывание сквозного машинного обучения с помощью PythonAnywhere, Flask и Power BI

Машинное обучение можно использовать для прогнозирования и кластеризации данных. Как мы можем интегрировать некоторые базовые возможности машинного обучения с Power BI и иметь серверную часть, которая обрабатывает логику и зависимости? В этой статье представлены базовые инструкции, показывающие один из способов развертывания API машинного обучения Python и подключения через Power BI.

Обзор проблемы

Для демонстрации я создал фиктивный набор данных, содержащий данные о жилье. Имея всего несколько функций, можно сделать приблизительное предположение о том, как они будут сгруппированы. Вы можете увидеть данные ниже - пара домов меньше, пара посередине и один большой выброс.

Учитывая перерывы в этих данных, мы можем вручную выполнить кластеризацию, просто просмотрев записи. Однако для большого набора данных это может быть невозможно, а написание правил вручную может занять много времени. Как мы можем использовать машинное обучение для автоматической кластеризации наших данных в группы похожих? В этой демонстрации мы будем использовать следующий подход:

  1. Создайте учетную запись на PythonAnywhere для размещения нашего Python API.
  2. Отправьте данные из Power BI (точнее, PowerQuery) в наш веб-API и добавьте группу кластеров в качестве нового столбца в наших данных.

Настройка PythonAnywhere

Создание учетной записи

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

Нажмите кнопку создания учетной записи, и вам будет предложено ввести имя пользователя, адрес электронной почты и пароль. После входа в систему вы увидите заголовок в правом верхнем углу. Мы будем использовать вкладки «Панель управления», «Консоли», «Файлы» и «Интернет».

Настройка виртуальной среды Python

Первое, что нам нужно сделать после входа в систему, - это настроить виртуальную среду с соответствующими пакетами python. Нажмите кнопку Bash на панели PythonAnywhere Dashboard, чтобы открыть новую консоль, и выполните следующие команды:

  1. Создайте новую виртуальную среду, указав версию и имя Python. Я использовал версию 3.7 и pbi_demo для названия. Окружение должно активироваться, как только вы его создадите. Вы также можете использовать команду workon для переключения между средами.
mkvirtualenv --python=/usr/bin/python3.7 pbi_demo

2. Установите необходимые пакеты Python с помощью pip:

pip install flask flask-restful marshmallow numpy pandas scikit-learn

Как только это будет завершено, вернитесь на панель управления. Следующим шагом будет создание веб-приложения.

Создание веб-приложения

Нажмите на заголовок Интернет вверху, а затем выберите параметр Добавить новое веб-приложение. Примите домен по умолчанию и выберите ручную настройку с Python версии 3.7 (или соответствующую тому, что вы выбрали на предыдущем шаге). Есть несколько дополнительных параметров, которые мы хотим изменить, описанные в следующих шагах.

Файл конфигурации WSGI

Первое изменение - это файл конфигурации WSGI под заголовком Код. Щелкните ссылку (заканчивается на .py), чтобы открыть файл.

Этот код вызывается первым при загрузке сервера. Код по умолчанию в этом файле настраивается PythonAnywhere и содержит комментарии, которые дают вам обзор некоторых функций. Для этой демонстрации удалите все и замените следующим (измените строку username в соответствии с вашей сборкой):

import sys

# add your project directory to the sys.path
project_home = u'/home/username/pbi_demo'
if project_home not in sys.path:
    sys.path = [project_home] + sys.path

# import flask app but need to call it "application" for WSGI to work
from SimpleRecommendation import app as application  # noqa

Этот код добавляет наш проект в системный путь, а затем импортирует наше приложение. Сохраните этот файл и вернитесь на вкладку Интернет.

Виртуальная среда

Следующий шаг - указать нашему веб-приложению виртуальную среду, которую мы создали. Нажмите ссылку Введите путь к виртуальному серверу и введите свой путь (пример ниже).

/home/username/.virtualenvs/pbi_demo

HTTPS

Наконец, в целях безопасности мы принудительно настроим HTTPS:

Вот и все на веб-вкладке, пора написать немного Python!

Код Python

Настройка проекта

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

Начальный код приложения

Flask и Flask-RESTful используются как веб и API-фреймворк. Отдельный модуль Python будет предназначен для того, чтобы просто развернуть приложение и api для использования на более поздних этапах.

app.py

from flask import Flask
from flask_restful import Api

app = Flask(__name__)
api = Api(app)

Схемы

Marshmallow используется для проверки и структурирования данных как для входящих запросов (сделанных Power BI), так и исходящих ответов обратно в Power BI после запуска логики кластеризации.

Вы увидите, что в приведенном ниже файле request_schemas.py есть класс RequestInput (наследует схему Marshmallow), который соответствует столбцам в нашем CSV.

Класс ClusterResultSchema определяет выходные данные, которые будут возвращены в Power BI после выполнения кода кластеризации - row, чтобы разрешить обратное соединение с исходными данными и group для обозначения кластера, в котором находится каждый дом. Параметр many = True используется, поскольку входные и выходные данные имеют несколько записей.

request_schemas.py

from marshmallow import Schema, fields

class RequestInput(Schema):
    square_footage = fields.Integer(required=True, strict=True)
    number_of_rooms = fields.Integer(required=True, strict=True)
    price = fields.Integer(required=True, strict=True)

# Input Schema
request_input_schema_multiple = RequestInput(many=True)

response_schemas.py

from marshmallow import Schema, fields

class ClusterResultSchema(Schema):
    row = fields.Integer(required=True, strict=True)
    group = fields.Integer(required=True, strict=True)

# Output Schema
response_schema_multiple = ClusterResultSchema(many=True)

Машинное обучение

Scikit-learn, а точнее KMeans Clustering, используется для кластеризации данных. Функция cluster_houses ниже отвечает за кластеризацию данных с учетом входного словаря, содержащего различные записи (дома).

жилищное_кластеринг.py

from sklearn.cluster import KMeans
import numpy as np
import pandas as pd

def cluster_houses(input_data):
    """Cluster housing data using KMeans

    Args:
        input_data (dict): api request data containing houses to be clustered

    Returns:
        list: list of dict objects containing row and cluster group
    """
    data = pd.DataFrame.from_dict(input_data).to_numpy()
    kmeans = KMeans(n_clusters=3, random_state=42).fit(data)
    return [{'row': (i+1), 'group': label} for i, label in enumerate(kmeans.labels_)]

Операции API

Основная часть кода обрабатывает и анализирует данные, а также отправляет ответ. Приведенный ниже файл api_operations.py отвечает за связывание всех частей вместе и раскрытие конечной точки API. Используя метод HTTP Post, Power BI отправляет данные, которые будут проанализированы, выполнены классификации, а затем результаты будут сброшены в строку json в ответе.

api_operations.py

# Flask Library Imports
from flask import request, json
from flask_restful import Resource

# Local object imports
from Starter.app import api
from Schemas.request_schemas import request_input_schema_multiple
from Schemas.response_schemas import response_schema_multiple

# ML Logic
from API.housing_clustering import cluster_houses


def parse_request(request):
    """Parse json from request

    Args:
        request: incoming request from User api call

    Returns:
        dict: json dict from request object
    """
    parsed_result = request.get_json(force=True)
    return parsed_result 


class HousingClustering(Resource):      
    
    def post(self):

        try:
            # Parse request into json and validate schema
            json_data = parse_request(request)
            errors = request_input_schema_multiple.validate(json_data)
            
            # If errors exist abort and send to api caller
            if errors:
                return str(errors), 400

            # Run clustering
            results = cluster_houses(json_data)
            
            # Dump results and ok status back to caller
            return response_schema_multiple.dumps(results), 200

        # Unknown error return 404
        except:
            return 'Unknown Error', 404

# Expose API endpoint
api.add_resource(HousingClustering, '/HousingClustering')

Точка входа

Наконец, простой модуль высокого уровня используется в качестве точки входа для PythonAnywhere. Он просто импортирует объекты Flask и API, чтобы файл оставался легким и четко структурированным. Это позволяет упростить усовершенствования, дополнительные функции или конечные точки и т. Д., Которые можно легко включить.

Если вы помните, при настройке файла WSGI ранее объект приложения из SimpleRecommendation был импортирован в этот код - это файл, на который он ссылается.

SimpleRecommendation.py
from Starter.app import *
from API.api_operations import *

И это все, что касается кода Python. Следующим шагом будет загрузка этих файлов и тестирование приложения.

Загрузки веб-приложений

Пришло время загрузить наши файлы .py в PythonAnywhere. Если вы нажмете на заголовок Файлы, вы увидите существующие системные файлы и файлы по умолчанию. Создайте новый каталог с именем pbi_demo (обратите внимание, что он использовался в нашем более раннем конфигурационном файле), а затем загрузите файлы с той же структурой / вложенностью папок. SimpleRecommendation.py будет находиться на верхнем уровне этого нового каталога, а другие файлы - в новых подкаталогах.

После того, как все файлы будут загружены, нажмите вкладку Интернет, а затем кнопку Обновить, чтобы перезагрузить сервер веб-приложений.

PowerQuery

Код

Обработка данных, отправка результатов и получение кластеров могут выполняться в PowerQuery. На высоком уровне приведенный ниже код выполняет следующие действия:

  1. Считайте данные в формате CSV, содержащие атрибуты жилья, с помощью функции Csv.Document.
  2. Преобразуйте данные в строку json, которая может быть принята нашим API с помощью функции Json.FromValue.
  3. Анализируйте результаты, возвращаемые нашим API (включая строку для присоединения и группу кластеров), с помощью функции Json.Document.
  4. Преобразуйте данные обратно в таблицу и присоединитесь к нашему исходному набору данных, по сути добавив группу кластеров в качестве нового столбца.
let
    // Read in input csv
    Source = Csv.Document(
        File.Contents("D:\test_data.csv"),
        [
            Delimiter=",", 
            Columns=3, 
            Encoding=1252, 
            QuoteStyle=QuoteStyle.None
        ]
    ),

    // Promote headers and update types
    headers_promoted = Table.PromoteHeaders(
        Source, 
        [PromoteAllScalars=true]
    ),
    type_changes = Table.TransformColumnTypes(
        headers_promoted,
        {
            {"square_footage", Int64.Type}, 
            {"number_of_rooms", Int64.Type}, 
            {"price", Int64.Type}
        }
    ),

    // Build API Parms
    api_url = "https://user.pythonanywhere.com/HousingClustering",
    content_input = Json.FromValue(type_changes),
    header = [
        #"Content-Type"="application/json", 
        #"accept"="application/json"
    ],

    // Make API call
    web_results = Json.Document(
        Web.Contents(
            api_url, 
            [Content=content_input]
        )
    ),

    // Convert api results to a table
    json_doc = Json.Document(web_results),
    table_from_list = Table.FromRecords(json_doc),
    update_types = Table.TransformColumnTypes(
        table_from_list, 
        {
            {"row", Int64.Type}, 
            {"group", Int64.Type}
        }
    ),

    // Join source and results
    idx = Table.AddIndexColumn(
        type_changes, 
        "Index",  // Col Name
        1,        // Initial
        1,        // Increment
        Int64.Type
    ),
    joined_tables = Table.Join(
        idx,      // Table 1
        "Index",  // Join Col 1
        update_types, // Table 2
        "row"         // Join Col 2
    ),
    final_cols = Table.RemoveColumns(
        joined_tables, 
        {"Index", "row"}
    )

in
   final_cols

Результаты

Вы можете увидеть результаты ниже после завершения последнего шага кода PowerQuery.

Наш API возвращал номер группы (начиная с 0) для каждой записи. Фактическое число ничего не означает, но дома с одинаковыми номерами сгруппированы вместе, показывая аналогичные характеристики, определенные KMeans. С этим небольшим набором данных все прошло хорошо. Два самых маленьких и самых дешевых дома были сгруппированы вместе (Кластер №2), два следующих по величине сгруппированы вместе (Кластер №0), а самый большой выброс, который мы заметили вручную, сгруппирован сам по себе (Кластер №1)!

Преимущества

Итак, каковы преимущества подхода, основанного на API? Мы могли бы запустить эту программу на Python локально и просто добавить результаты в наш CSV-файл; однако выделение нашей логики машинного обучения в отдельный API дает некоторые дополнительные преимущества:

  1. Мы обрабатываем зависимости Python в одном месте (в PythonAnywhere в этом примере). Это избавляет от необходимости напрямую интегрировать Python с Power BI или беспокоиться о настройке нескольких пользовательских компьютеров с правильными библиотеками и совместимостью.
  2. Вместо того, чтобы запускать классификацию один раз локально, теперь мы можем обновить наш набор данных Power BI в Службе новыми данными - извлечение кластеров без необходимости что-либо изменять. Просто введите новые данные и получите обновленные результаты - PowerQuery и наш API Python справятся со всей тяжелой работой.
  3. Мы можем оптимизировать алгоритм машинного обучения на сервере, не внося изменений в наш файл Power BI или код PowerQuery. Power BI просто знает, что нужно отправить три столбца, и ожидает, что группа (и столбец строки присоединятся) взамен. Пока наш API выполняет эту задачу, мы можем оптимизировать фактическую кластеризацию, тестировать различные алгоритмы и т. Д. - и все это без необходимости в Power BI знать о каких-либо изменениях.

Предостережения

В этой демонстрации показан пример сквозного машинного обучения на основе исходного набора данных путем извлечения результатов в инструмент бизнес-аналитики. Однако это не означает, что вам следует использовать именно этот подход. Стоит упомянуть несколько моментов, которые могут немного изменить то, как вы реализуете что-то подобное на практике:

  1. Скорее всего, у вашей компании есть предпочтительный способ развертывания виртуальных машин и API. Следуйте подходу вашей компании. При настройке личной учетной записи PythonAnywhere с помощью Flask можно быстро раскрыть конечную точку API, ваша компания, вероятно, использует более строгий подход по уважительным причинам (облачный провайдер по выбору с точки зрения затрат и преимуществ интеграции, управление учетными записями служб, технические ресурсы, размещенные внутри межсетевого экрана компании и т. д.).
  2. Существуют более эффективные способы для очень больших наборов данных. Отправка большого количества записей с помощью метода POST API может быть медленной в зависимости от скорости сети. Альтернативным вариантом может быть массовая загрузка данных, обработка, а затем сохранение результатов, например, в таблице SQL. Эту предварительную обработку можно выполнять в нерабочее время, и это ускорит время ожидания данных в Power BI. В этот момент он просто превращается в SQL-запрос для получения окончательных результатов. Однако это требует дополнительных инженерных работ и логики ETL.
  3. Существуют облачные решения, которые предоставляют большую часть этой функциональности из коробки. Например, Машинное обучение Azure является одним из примеров, который хорошо интегрируется в среду Azure. Потоки данных также обладают возможностями искусственного интеллекта, которые обеспечивают более тесную интеграцию с Power BI. Если у вас есть Power BI Premium, это будет намного проще настроить, хотя и с меньшим количеством параметров настройки.

Резюме

Я надеюсь, что это пошаговое руководство было полезным и заставит вас задуматься о способах интеграции машинного обучения и искусственного интеллекта с Power BI таким образом, чтобы это можно было повторить и автоматизировать.

Заинтересованы в ваших мыслях, если вы нашли этот подход полезным или использовали разные подходы в прошлом для решения аналогичных проблем - прокомментируйте ниже!

Все примеры и файлы доступны на Github.

Первоначально опубликовано на https://datastud.dev.