How-to: Прогнозируем отток клиентов с помощью Apache Spark MLlib

Spark MLlib набирает все большую популярность благодаря своей элегантности и практичности. Сегодня мы продемонстрируем практическое применение этого инструмента.

Spark MLlib – это библиотека для реализации машинного обучения и решения сопутствующих задач, связанных с большими данными. MLlib позволяет с помощью нескольких строк кода обучить модель на массиве данных, содержащем миллиард записей, используя при этом сотни машин. Данная библиотека существенно упрощает процесс разработки моделей.

В этой статье мы применим MLlib для прогнозирования оттока клиентов телефонной компании (churn prediction) – это одно из наиболее распространенных приложений машинного обучения в телекоммуникационной отрасли, а также в других отраслях, где применяется бизнес-модель на основе подписки.

Мы выполним анализ и моделирование с помощью Python. Для манипуляций с данными используем объект Spark DataFrame. Для оптимизации рабочего процесса автоматизируем извлечение признаков, обучение и тестирование модели с помощью конвейера Spark Pipeline. (Ядро MLlib входит в состав дистрибутива CDH 5.5 (Cloudera Enterprise 5.5), но поддержка конвейеров появится только в следующих релизах.)

Данная статья основана на материалах курса «Data science для телекоммуникационного бизнеса» («Data Science for Telecom»), представленного на конференции Strata + Hadoop World Singapore 2015. Весь исходный код с результатами доступен в формате IPython notebook. В репозитории также присутствует скрипт, с помощью которого можно запустить IPython notebook в CDH-кластере.

Загружаем данные в Spark DataFrame

Для обучения модели мы используем набор данных по оттоку клиентов из репозитория Калифорнийского университета в Ирвайне (UCI). В этом наборе данных каждая запись содержит информацию об одном абоненте телефонной компании. Последнее поле каждой записи сообщает нам, отказался ли данный абонент от услуг компании.

Набор данных содержит всего 5 000 записей (т.е. абонентов), что на много порядков меньше объема, с которым способен работать Spark. Небольшой размер данных позволит нам с легкостью протестировать интересующие нас инструменты на ноутбуке.

Перечень полей:

• state – штат;
• account length – абонентский стаж;
• area code – код региона;
• phone number – номер телефона;
• international plan – тарифный план для международных звонков;
• voice mail plan – тарифный план для голосовой почты;
• number vmail messages – количество сообщений голосовой почты;
• total day minutes – общая длительность звонков в дневное время (мин);
• total day calls – общее количество звонков в дневное время;
• total day charge – общая стоимость звонков в дневное время;
• total eve minutes – общая длительность звонков в вечернее время (мин);
• total eve calls – общее количество звонков в вечернее время;
• total eve charge – общая стоимость звонков в вечернее время;
• total night minutes – общая длительность звонков в ночное время (мин);
• total night calls – общее количество звонков в ночное время;
• total night charge – общая стоимость звонков в ночное время;
• total intl minutes – общая длительность международных звонков (мин);
• total intl calls – общее количество международных звонков;
• total intl charge – общая стоимость международных звонков;
• number customer service calls – количество звонков в службу поддержки.
• churned – покинул ли клиент компанию.

Последнее поле «churned» представляет собой категорийную переменную, принимающую значение «True» («да») или «False» («нет»). Эту переменную мы и хотим предсказать. Остальные поля послужат в качестве признаков.

Чтобы загрузить эти данные в Spark DataFrame, необходимо указать тип каждого поля. Для работы с CSV-форматом нам понадобится пакет spark-csv, который разрабатывается отдельного от основного проекта Spark:

from pyspark.sql import SQLContext

from pyspark.sql.types import *

sqlContext = SQLContext(sc)

schema = StructType([ \

    StructField("state", StringType(), True), \

    StructField("account_length", DoubleType(), True), \

    StructField("area_code", StringType(), True), \

    StructField("phone_number", StringType(), True), \

    StructField("intl_plan", StringType(), True), \

    StructField("voice_mail_plan", StringType(), True), \

    StructField("number_vmail_messages", DoubleType(), True), \

    StructField("total_day_minutes", DoubleType(), True), \

    StructField("total_day_calls", DoubleType(), True), \

    StructField("total_day_charge", DoubleType(), True), \

    StructField("total_eve_minutes", DoubleType(), True), \

    StructField("total_eve_calls", DoubleType(), True), \

    StructField("total_eve_charge", DoubleType(), True), \

    StructField("total_night_minutes", DoubleType(), True), \

    StructField("total_night_calls", DoubleType(), True), \

    StructField("total_night_charge", DoubleType(), True), \

    StructField("total_intl_minutes", DoubleType(), True), \

    StructField("total_intl_calls", DoubleType(), True), \

    StructField("total_intl_charge", DoubleType(), True), \

    StructField("number_customer_service_calls", DoubleType(), True), \

    StructField("churned", StringType(), True)])


churn_data = sqlContext.read \

    .format('com.databricks.spark.csv') \

    .load('churn.all', schema = schema)

Обучаем модель

Библиотека MLlib предоставляет набор алгоритмов для машинного обучения и обработки больших наборов данных. В нашем примере мы создадим конвейер (Pipeline), который позволит нам автоматизировать извлечение признаков и обучение модели. В качестве алгоритма для модели мы используем случайный лес (random forest).

Общая схема процесса машинного обучения с учителем и оценки модели имеет следующий вид:

Мы уже загрузили данные в Spark DataFrame с названием churn_data. Далее необходимо выполнить извлечение признаков (feature extraction), т.е. преобразовать данные, чтобы получить векторы признаков (feature vector) и метки классов (label). Вектор признаков – это массив вещественных чисел, представляющих независимые переменные, которые модель может использовать для обучения и прогнозирования. Метка класса – это одно вещественное число, являющееся зависимой переменной, которую модель пытается предсказать. В задачах бинарной классификации используются значения 0,0 и 1,0 для представления двух возможны вариантов прогноза. В нашем случае 0,0 означает, что «клиент не покинет компанию», а 1,0 – «клиент покинет компанию».

Извлечение признаков подразумевает широкий спектр возможных преобразований, позволяющих получить векторы признаков и метки классов. В нашем случае мы проиндексируем категорийные переменные, представленные строками (такие как intl_plan), чтобы преобразовать их в числа.

Кроме того, нас интересуют не все поля данных. Например, номер телефона (phone_number) вряд ли является полезным признаком, поэтому мы не будем использовать его в нашей модели. С другой стороны, общее количество звонков в дневное время (total_day_calls) напротив, вероятно, является значимым признаком, следовательно, будет присутствовать в нашей модели. Чтобы включить эти преобразования в конвейер, мы выделим их в отдельные этапы: StringIndexer и VectorAssembler.

from pyspark.ml.feature import StringIndexer

from pyspark.ml.feature import VectorAssembler

label_indexer = StringIndexer(inputCol = 'churned', outputCol = 'label')

plan_indexer = StringIndexer(inputCol = 'intl_plan', outputCol = 'intl_plan_indexed')

reduced_numeric_cols = ["account_length", "number_vmail_messages", "total_day_calls",

                        "total_day_charge", "total_eve_calls", "total_eve_charge",

                        "total_night_calls", "total_intl_calls", "total_intl_charge"]

assembler = VectorAssembler(

    inputCols = ['intl_plan_indexed'] + reduced_numeric_cols,

    outputCol = 'features')

Теперь необходимо разбить данные на обучающий (train) и тестовый (test) наборы. С помощью обучающего набора мы обучим модель, а тестовый набор будет использован для ее оценки.

(train, test) = churn_data.randomSplit([0.7, 0.3])

Теперь мы можем собрать наш конвейер и обучить модель. MLlib позволяет реализовать это с помощью всего лишь нескольких строк кода!

from pyspark.ml import Pipeline

from pyspark.ml.classification import RandomForestClassifier

classifier = RandomForestClassifier(labelCol = 'label', featuresCol = 'features')

pipeline = Pipeline(stages=[plan_indexer, label_indexer, assembler, classifier])

model = pipeline.fit(train)

Оцениваем модель

Как же нам узнать, хорошую ли модель мы создали? Можем ли мы показать, что ее прогнозы точнее, чем случайное угадывание? Для бинарных классификаторов хорошей метрикой является площадь под ROC-кривой. Чтобы построить ROC-кривую, модель должна выполнить классификацию при различных значениях порога. Изменяя порог, мы перебираем различные сочетания доли истинно-положительных прогнозов (true positive rate, TPR) и доли ложно-положительных прогнозов (false positive rate, FPR). Таким образом, мы проходим промежуточные варианты между двумя крайностями: (1) оба показателя, TPR и FPR, равны 0, потому что во всех случаях было предсказано, что «клиент не покинет компанию»; (2) оба показателя, TPR и FPR, равны 1, потому что во всех случаях было предсказано, что «клиент покинет компанию».

Для классификатора, который случайным образом равновероятно предсказывает, уйдет клиент или нет, ROC-кривая будет иметь вид прямой, являющейся диагональю единичного квадрата. Диагональ делит единичный квадрат на два равных треугольника, поэтому площадь под данной ROC-кривой равна 0,5. Следовательно, если, оценивая некоторый классификатор, мы получаем площадь под ROC-кривой, равную 0,5, это говорит о том, что данный классификатор позволяет отличить два класса не лучше, чем случайное угадывание. Чем ближе значение площади к единице, тем лучше классификатор. Если площадь под ROC-кривой меньше 0,5, мы можем улучшить результаты модели, инвертировав ее прогнозы.

Благодаря MLlib, вычисление площади под ROC-кривой является очень простой задачей.

Если бы мы использовали для тестирования модели обучающий набор данных, мы получали бы слишком оптимистичные оценки. Поэтому, чтобы избежать завышенных оценок и переобучения (overfitting), мы оцениваем нашу модель только на тестовом наборе данных.

from pyspark.ml.evaluation import BinaryClassificationEvaluator

predictions = model.transform(test)

evaluator = BinaryClassificationEvaluator()

auroc = evaluator.evaluate(predictions, {evaluator.metricName: "areaUnderROC"})

В случае нашей модели площадь под ROC-кривой составила более 0,8. Это означает, что данная модель способна делать достаточно точные прогнозы.

Заключение

Сегодня мы рассмотрели одну из множества задач, которые могут быть эффективно решены с помощью MLlib. Другие примеры и руководства по машинному обучению и Spark в целом вы можете найти здесь.

По материалам: Cloudera

Перевод Станислава Петренко