Распараллеливайте свои массивные вычисления SHAP с помощью MLlib и PySpark.

Пошаговое руководство по эффективному объяснению ваших моделей с помощью SHAP.

Введение в MLlib

Библиотека машинного обучения Apache Spark (MLlib) предназначена в первую очередь для обеспечения масштабируемости и скорости за счет использования среды выполнения Spark для распространенных распределенных случаев использования в обучении с учителем, таком как классификация и регрессия, обучении без учителя, таком как кластеризация и совместная фильтрация, а также в других областях. случаи, подобные уменьшению размерности. В этой статье я расскажу, как мы можем использовать SHAP для объяснения модели Gradient Boosted Trees (GBT), которая соответствует нашим данным в масштабе.

Что такое градиентные деревья?

Прежде чем мы поймем, что такое деревья с градиентным усилением, нам нужно понять, что такое повышение. Повышение – это метод ансамбля, который последовательно объединяет несколько слабых учеников, чтобы получить в целом сильного ученика. В случае деревьев с градиентным усилением каждый слабый ученик представляет собой дерево решений, которое последовательно минимизирует ошибки (MSE в случае регрессии и потери журнала в случае классификации), сгенерированные предыдущим деревом решений в этой последовательности. Чтобы прочитать о GBT более подробно, обратитесь к этому сообщению в блоге.

Понимание нашего импорта

from pyspark.sql import SparkSession
from pyspark import SparkContext, SparkConf
from pyspark.ml.classification import GBTClassificationModel
import pyspark.sql.functions as F
from pyspark.sql.types import *

Первые два импорта предназначены для инициализации сеанса Spark. Он будет использоваться для преобразования нашего фрейма данных pandas в искровой. Третий импорт используется для загрузки нашей модели GBT в память, которая будет передана нашему объяснению SHAP для создания объяснений. Предпоследний и последний импорт предназначен для выполнения функций SQL и использования типов SQL. Они будут использоваться в нашей определяемой пользователем функции (UDF), которую я опишу позже.

Преобразование нашего вектора функций MLlib GBT в кадр данных Pandas

Объяснитель SHAP принимает в качестве входных данных кадр данных. Однако для обучения модели MLlib GBT требуется предварительная обработка данных. В частности, категориальные переменные в наших данных необходимо преобразовать в числовые переменные, используя либо индексирование категорий, либо горячее кодирование. Чтобы узнать больше о том, как обучать модель GBT, обратитесь к этой статье). Результирующий столбец функцииявляется SparseVector (чтобы узнать больше о нем, см. раздел Предварительная обработка данных в этом примере). Это выглядит как-то ниже:

Показанный выше столбец функции относится к одному учебному экземпляру. Нам нужно преобразовать этот SparseVector для всех наших учебных экземпляров. Один из способов сделать это — итеративно обрабатывать каждую строку и добавлять ее в наш фрейм данных pandas, который мы будем передавать нашему объяснению SHAP (ой!). Существует гораздо более быстрый способ, который использует тот факт, что все наши данные загружены в память (если нет, мы можем загружать их партиями и выполнять предварительную обработку для каждой партии в памяти). По словам Шихар Дуа:

1. Создайте список словарей, в котором каждый словарь соответствует строке входных данных.

2. Создайте фрейм данных из этого списка.

Итак, на основе вышеизложенного метода мы получаем что-то вроде этого:

rows_list = []
for row in spark_df.rdd.collect(): 
    dict1 = {} 
    dict1.update({k:v for k,v in zip(spark_df.cols,row.features)})
    rows_list.append(dict1) 
pandas_df = pd.DataFrame(rows_list)

Если rdd.collect() выглядит пугающе, на самом деле это довольно просто объяснить. Отказоустойчивые распределенные наборы данных (RDD) – это фундаментальные структуры данных Spark, которые представляют собой неизменяемоераспределение объектов. Каждый набор данных в RDD далее подразделяется на логические разделы, которые можно вычислять на разных рабочих узлах нашего кластера Spark. Таким образом, все, что делает PySpark RDD collect(), — это извлекает данные со всех рабочих узлов на узел драйвера. Как вы можете догадаться, это узкое место в памяти, и если мы обрабатываем данные, превышающие объем памяти нашего узла драйвера, нам нужно увеличить количество наших разделов RDD и отфильтровать их по индексу раздела. Как это сделать читайте здесь.

Не верьте мне на слово по поводу казни. Проверьте статистику.

Вот метрики одного из запланированных заданий моего блокнота Databricks:

Размер входных данных: 11,9 ГБ (~ 12,78 ГБ), общее время выполнения всех задач: 20 мин, количество записей: 165,16 КБ

Работа с библиотекой SHAP

Теперь мы готовы передать наш предварительно обработанный набор данных в SHAP TreeExplainer. Помните, что SHAP — это метод атрибуции локальных признаков, который объясняет индивидуальныепрогнозы как алгебраическую сумму значений Шепли признаков нашей модели.

Мы используем TreeExplainer по следующим причинам:

Подходит: TreeExplainer — это класс, который вычисляет значения SHAP для древовидных моделей (Random Forest, XGBoost, LightGBM, GBT и т. д.).
Точный: вместо того, чтобы имитировать отсутствующие функции путем случайной выборки, используется древовидная структура, просто игнорируя пути принятия решений, основанные на отсутствующих функциях. Таким образом, выходные данные TreeExplainer являются детерминированными и не зависят от фонового набора данных.
Эффективность: вместо перебора каждой возможной комбинации функций (или их подмножества) все комбинации одновременно проталкиваются по дереву с использованием более сложного алгоритма для отслеживания результата каждой комбинации — снижение сложности с O (TL2ᵐ) для всех возможных коалиций к полиному O(TLD²) (где m — количество признаков, T количество деревьев, Lмаксимальное количество листьев и Dмаксимальная глубина дерева).

Флаг check_additivity = False запускает проверку правильности, чтобы убедиться, что сумма значений SHAP равна выходным данным модели. Однако этот флаг требует выполнения прогнозов, которые не поддерживаются Spark, поэтому для него необходимо установить значение False, поскольку он все равно игнорируется. Как только мы получим значения SHAP, мы преобразуем их в кадр данных pandas из массива Numpy, чтобы его было легко интерпретировать.

Следует отметить, что порядок набора данных сохраняется при преобразовании кадра данных Spark в pandas, но обратное неверно.

Пункты выше приводят нас к фрагменту кода ниже:

gbt = GBTClassificationModel.load('your-model-path') 
explainer = shap.TreeExplainer(gbt)
shap_values = explainer(pandas_df, check_additivity = False)
shap_pandas_df = pd.DataFrame(shap_values.values, cols = pandas_df.columns)

Введение в UDF Pyspark и когда их использовать

Пользовательские функции — это сложные пользовательские функции, которые работают с определенной строкой нашего набора данных. Эти функции обычно используются, когда собственных функций Spark недостаточно для решения проблемы. Функции Spark по своей природе быстрее, чем определяемые пользователем функции, поскольку изначально представляют собой структуру JVM, методы которой реализуются локальными вызовами API Java. Однако пользовательские функции PySpark — это реализации Python, которые требуют перемещения данных между интерпретатором Python и JVM (см. стрелку 4 на рисунке выше). Это неизбежно приводит к некоторой задержке обработки.

Если задержки обработки недопустимы, лучше всего создать оболочку Python для вызова UDF Scala из самого PySpark. Отличный пример показан в этом блоге. Однако для моего случая использования PySpark UDF было достаточно, поскольку его легко понять и написать код.

В приведенном ниже коде объясняется функция Python, которая должна выполняться на каждом рабочем/исполнительном узле. Мы просто выбираем самые высокие значения SHAP (абсолютные значения, так как мы также хотим найти наиболее важные негативные функции) и добавляем их в соответствующие списки pos_features и neg_features и в добавьте оба этих списка к списку функций, который возвращается вызывающей стороне.

def shap_udf(row):
    dict = {} 
    pos_features = [] 
    neg_features = [] 
    for feature in row.columns: 
        dict[feature] = row[feature] 
    dict_importance = {key: value for key, value in
    sorted(dict.items(), key=lambda item: __builtin__.abs(item[1]),   
    reverse = True)} 
    for k,v in dict_importance.items(): 
        if __builtin__.abs(v) >= <your-threshold-shap-value>: 
             if v > 0: 
                 pos_features.append((k,v)) 
             else: 
                 neg_features.append((k,v)) 
   features = [] 
   features.append(pos_features[:5]) 
   features.append(neg_features[:5]) 
   return features

Затем мы регистрируем нашу UDF PySpark с именем нашей функции Python (в моем случае это shap_udf) и указываем тип возвращаемого значения (обязательный в Python и Java) функции в параметрах F .udf(). Во внешнем ArrayType() есть два списка: один для положительных характеристик, а другой для отрицательных. Поскольку каждый отдельный список состоит не более чем из 5 пар StructType() (функция-имя, форма-значение), он представляет собой внутренний ArrayType(). Ниже приведен код:

udf_obj = F.udf(shap_udf, ArrayType(ArrayType(StructType([ StructField(‘Feature’, StringType()), 
StructField(‘Shap_Value’, FloatType()),
]))))

Теперь мы просто создаем новый фрейм данных Spark со столбцом под названием «Shap_Importance», который вызывает нашу пользовательскую функцию для каждой строки spark_shapdf фрейма данных. Чтобы разделить положительные и отрицательные функции, мы создаем два столбца в новом фрейме данных Spark с именем final_sparkdf. Наш окончательный фрагмент кода выглядит следующим образом:

new_sparkdf = spark_df.withColumn(‘Shap_Importance’, udf_obj(F.struct([spark_shapdf[x] for x in spark_shapdf.columns])))
final_sparkdf = new_sparkdf.withColumn(‘Positive_Shap’, final_sparkdf.Shap_Importance[0]).withColumn(‘Negative_Shap’, new_sparkdf.Shap_Importance[1])

И, наконец, мы извлекли все важные функции нашей модели GBT для каждого тестового экземпляра без использования каких-либо явных циклов for! Консолидированный код можно найти в приведенном ниже списке GitHub.

P.S. Это моя первая попытка написать статью, и если есть какие-либо фактические или статистические несоответствия, пожалуйста, свяжитесь со мной, и я буду более чем счастлив учиться вместе с вами! :)