Create ConnectionS park & Python: MLlib Basic Statistics & Exploratory Data Analysis

Автор оригинала: Jose A Dianes.

Инструкции

Моя серия учебников Spark & Python может быть рассмотрена индивидуально, хотя существует более или менее линейная “история”, если следовать ей последовательно. Используя один и тот же набор данных, они пытаются решить с ним связанный набор задач.

Это не единственный способ, но хороший способ следовать этим учебникам Spark-сначала клонировать репо GitHub , а затем запускать свой собственный IPython notebook в режиме PySpark . Например, если у нас есть автономная Spark установка, работающая в нашем localhost с максимальным объемом 6 Гб на узел, назначенный IPython:

MASTER="spark://127.0.0.1:7077" SPARK_EXECUTOR_MEMORY="6G" IPYTHON_OPTS="notebook --pylab inline" ~/spark-1.3.1-bin-hadoop2.6/bin/pyspark

Обратите внимание, что путь к команде pyspark будет зависеть от вашей конкретной установки. Таким образом, в качестве требования вам необходимо установить Spark на той же машине, на которой вы собираетесь запустить сервер IPython notebook .

Дополнительные параметры Spark см. в разделе здесь . В общем случае работает правило передачи опций, описанное в виде spark.executor.memory as SPARK_EXECUTOR_MEMORY при вызове IPython/PySpark.

Наборы данных

Мы будем использовать наборы данных из KDD Cup 1999 .

Рекомендации

Справочником по этим и другим темам, связанным с Spark, является Learning Spark Holden Karau, Andy Konwinski, Patrick Wendell и Matei Zaharia.

Набор данных соревнований KDD Cup 1999 подробно описан здесь .

Вступление

До сих пор мы использовали различные картографические и агрегационные функции на простых и парных Rdd ключ/значение, чтобы получить простую статистику, которая поможет нам понять наши наборы данных. В этом уроке мы познакомимся с библиотекой машинного обучения Spark MLlib через ее базовую статистическую функциональность, чтобы лучше понять наш набор данных. Мы будем использовать сокращенные 10-процентные наборы данных KDD Cup 1999 .

Получение данных и создание RDD

Как и в нашем первом блокноте, мы будем использовать сокращенный набор данных (10 процентов), предоставленный для KDD Cup 1999 , содержащий почти полмиллиона сетевых взаимодействий. Файл предоставляется в виде файла Gzip, который мы будем загружать локально.

import urllib
f = urllib.urlretrieve ("http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data_10_percent.gz", "kddcup.data_10_percent.gz")


data_file = "./kddcup.data_10_percent.gz"
raw_data = sc.textFile(data_file)

Локальные Векторы

A local vector часто используется в качестве базового типа для RDDS в Spark MLlib. Локальный вектор имеет целочисленные и 0-типизированные индексы и значения двойного типа, хранящиеся на одной машине. MLlib поддерживает два типа локальных векторов: плотные и разреженные. Плотный вектор поддерживается двойным массивом, представляющим его входные значения, в то время как разреженный вектор поддерживается двумя параллельными массивами: индексами и значениями.

Для плотных векторов MLlib использует либо Python lists , либо тип NumPy /array . Рекомендуется использовать более поздний вариант, так что вы можете просто передавать массивы NumPy.

Для разреженных векторов пользователи могут построить объект Sparse Vector из MLlib или передать SciPy | scipy.sparse column vectors, если SciPy доступен в их среде. Самый простой способ создания разреженных векторов-использовать заводские методы, реализованные в Vectors .

RDD плотных векторов

Давайте представим каждое сетевое взаимодействие в нашем наборе данных в виде плотного вектора. Для этого мы будем использовать тип NumPy | array .

import numpy as np

def parse_interaction(line):
    line_split = line.split(",")
    # keep just numeric and logical values
    symbolic_indexes = [1,2,3,41]
    clean_line_split = [item for i,item in enumerate(line_split) if i not in symbolic_indexes]
    return np.array([float(x) for x in clean_line_split])

vector_data = raw_data.map(parse_interaction)

Сводная статистика

MLlib Spark предоставляет сводную статистику столбцов для RDD[Vector] через функцию colStats , доступную в Statistics . Метод возвращает экземпляр Многомерной статистической сводки , которая содержит по столбцам max , min , mean , variance и number of nonzeros , а также total count .

from pyspark.mllib.stat import Statistics 
from math import sqrt 

# Compute column summary statistics.
summary = Statistics.colStats(vector_data)

print "Duration Statistics:"
print " Mean: {}".format(round(summary.mean()[0],3))
print " St. deviation: {}".format(round(sqrt(summary.variance()[0]),3))
print " Max value: {}".format(round(summary.max()[0],3))
print " Min value: {}".format(round(summary.min()[0],3))
print " Total value count: {}".format(summary.count())
print " Number of non-zero values: {}".format(summary.numNonzeros()[0])

Duration Statistics:  
Mean: 47.979  
St. deviation: 707.746  
Max value: 58329.0  
Min value: 0.0  
Total value count: 494021  
Number of non-zero values: 12350.0  

Сводная статистика по меткам

Интересная часть сводной статистики, в нашем случае, исходит из возможности получить ее по типу сетевой атаки или “метки” в нашем наборе данных. Таким образом, мы сможем лучше охарактеризовать нашу зависимую переменную набора данных в терминах диапазона значений независимых переменных.

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

def parse_interaction_with_key(line):
    line_split = line.split(",")
    # keep just numeric and logical values
    symbolic_indexes = [1,2,3,41]
    clean_line_split = [item for i,item in enumerate(line_split) if i not in symbolic_indexes]
    return (line_split[41], np.array([float(x) for x in clean_line_split]))

label_vector_data = raw_data.map(parse_interaction_with_key)

Следующий шаг не очень сложный. Мы используем filter на RDD, чтобы исключить другие метки, кроме той, из которой мы хотим собрать статистику.

normal_label_data = label_vector_data.filter(lambda x: x[0]=="normal.")

Теперь мы можем использовать новый RDD для вызова colStats по значениям.

normal_summary = Statistics.colStats(normal_label_data.values())

И собирать результаты, как мы делали это раньше.

print "Duration Statistics for label: {}".format("normal")
print " Mean: {}".format(normal_summary.mean()[0],3)
print " St. deviation: {}".format(round(sqrt(normal_summary.variance()[0]),3))
print " Max value: {}".format(round(normal_summary.max()[0],3))
print " Min value: {}".format(round(normal_summary.min()[0],3))
print " Total value count: {}".format(normal_summary.count())
print " Number of non-zero values: {}".format(normal_summary.numNonzeros()[0])

Duration Statistics for label: normal  
Mean: 216.657322313  
St. deviation: 1359.213  
Max value: 58329.0  
Min value: 0.0  
Total value count: 97278  
Number of non-zero values: 11690.0  

Вместо того чтобы работать с парой ключ/значение, мы могли бы просто отфильтровать необработанные данные, разделенные с помощью метки в столбце 41. Затем мы можем проанализировать результаты, как мы делали это раньше. Это тоже сработает. Однако организация наших данных в виде пар ключ/значение откроет дверь для лучших манипуляций. Поскольку values() является преобразованием в RDD, а не действием, мы не выполняем никаких вычислений, пока не вызовем colStats в любом случае.

Но давайте обернем это в функцию, чтобы мы могли повторно использовать ее с любой меткой.

def summary_by_label(raw_data, label):
    label_vector_data = raw_data.map(parse_interaction_with_key).filter(lambda x: x[0]==label)
    return Statistics.colStats(label_vector_data.values())

Давайте снова попробуем использовать ярлык “нормальный”.

normal_sum = summary_by_label(raw_data, "normal.")

print "Duration Statistics for label: {}".format("normal")
print " Mean: {}".format(normal_sum.mean()[0],3)
print " St. deviation: {}".format(round(sqrt(normal_sum.variance()[0]),3))
print " Max value: {}".format(round(normal_sum.max()[0],3))
print " Min value: {}".format(round(normal_sum.min()[0],3))
print " Total value count: {}".format(normal_sum.count())
print " Number of non-zero values: {}".format(normal_sum.numNonzeros()[0])

Duration Statistics for label: normal  
Mean: 216.657322313  
St. deviation: 1359.213  
Max value: 58329.0  
Min value: 0.0  
Total value count: 97278  
Number of non-zero values: 11690.0  

Давайте попробуем теперь с какой-нибудь сетевой атакой. У нас все они перечислены здесь .

guess_passwd_summary = summary_by_label(raw_data, "guess_passwd.")

print "Duration Statistics for label: {}".format("guess_password")
print " Mean: {}".format(guess_passwd_summary.mean()[0],3)
print " St. deviation: {}".format(round(sqrt(guess_passwd_summary.variance()[0]),3))
print " Max value: {}".format(round(guess_passwd_summary.max()[0],3))
print " Min value: {}".format(round(guess_passwd_summary.min()[0],3))
print " Total value count: {}".format(guess_passwd_summary.count())
print " Number of non-zero values: {}".format(guess_passwd_summary.numNonzeros()[0])

Duration Statistics for label: guess_password  
Mean: 2.71698113208  
St. deviation: 11.88  
Max value: 60.0  
Min value: 0.0  
Total value count: 53  
Number of non-zero values: 4.0  

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

label_list = ["back.","buffer_overflow.","ftp_write.",
              "guess_passwd.","imap.","ipsweep.",
              "land.","loadmodule.","multihop.",
              "neptune.","nmap.","normal.","perl.",
              "phf.","pod.","portsweep.",
              "rootkit.","satan.","smurf.","spy.",
              "teardrop.","warezclient.",
              "warezmaster."]

Затем мы получаем список статистики для каждой метки.

stats_by_label = [(label, summary_by_label(raw_data, label)) for label in label_list]

Теперь мы получаем столбец duration , первый в нашем наборе данных (т. Е. индекс 0).

duration_by_label = [ 
    (stat[0], 
     np.array([
         float(stat[1].mean()[0]), 
         float(sqrt(stat[1].variance()[0])), 
         float(stat[1].min()[0]), 
         float(stat[1].max()[0]), 
         int(stat[1].count())])) 
    for stat in stats_by_label]

Это мы можем поместить в фрейм данных Панды .

import pandas as pd
pd.set_option('display.max_columns', 50)

stats_by_label_df = pd.DataFrame.from_items(duration_by_label, columns=["Mean", "Std Dev", "Min", "Max", "Count"], orient='index')

И напечатать.

print "Duration statistics, by label"

stats_by_label_df

Статистика продолжительности, по меткам

Чтобы повторно использовать этот код и получить фрейм данных из любой переменной в нашем наборе данных, мы определим функцию.

def get_variable_stats_df(stats_by_label, column_i):
    column_stats_by_label = [
        (stat[0], 
         np.array([
             float(stat[1].mean()[column_i]), 
             float(sqrt(stat[1].variance()[column_i])), 
             float(stat[1].min()[column_i]), 
             float(stat[1].max()[column_i]), 
             int(stat[1].count())])) 
        for stat in stats_by_label
    ]
    return pd.DataFrame.from_items(
        column_stats_by_label, 
        columns=["Mean", "Std Dev", "Min", "Max", "Count"], 
        orient='index')

Давайте попробуем еще раз duration .

get_variable_stats_df(stats_by_label,0)

Теперь перейдем к следующему числовому столбцу в наборе данных: src_bytes .

print "src_bytes statistics, by label"
get_variable_stats_df(stats_by_label,1)

src_bytes statistics, by label

И так далее. Повторно используя функции summary_by_label и get_variable_stats_df , мы можем выполнить некоторый исследовательский анализ данных в больших наборах данных с помощью Spark.

Корреляции

MLlib Spark поддерживает Pearson и Spearman для вычисления методов попарной корреляции между многими рядами. Оба они предоставляются методом corr в пакете Statistics .

У нас есть два варианта в качестве входных данных. Либо два RDD[Double] s, либо RDD[Vector] . В первом случае результатом будет значение Double , а во втором – целая корреляционная матрица. Из-за характера наших данных мы получим второе.

from pyspark.mllib.stat import Statistics 
correlation_matrix = Statistics.corr(vector_data, method="spearman")

Как только у нас будут готовы корреляции, мы сможем начать проверять их значения.

import pandas as pd
pd.set_option('display.max_columns', 50)

col_names = ["duration","src_bytes","dst_bytes",
             "land","wrong_fragment",
             "urgent","hot","num_failed_logins",
             "logged_in","num_compromised",
             "root_shell","su_attempted",
             "num_root","num_file_creations",
             "num_shells","num_access_files",
             "num_outbound_cmds",
             "is_hot_login","is_guest_login","count",
             "srv_count","serror_rate",
             "srv_serror_rate","rerror_rate",
             "srv_rerror_rate","same_srv_rate",
             "diff_srv_rate","srv_diff_host_rate",
             "dst_host_count","dst_host_srv_count",
             "dst_host_same_srv_rate","dst_host_diff_srv_rate",
             "dst_host_same_src_port_rate",
             "dst_host_srv_diff_host_rate","dst_host_serror_rate",
             "dst_host_srv_serror_rate",
             "dst_host_rerror_rate","dst_host_srv_rerror_rate"]

corr_df = pd.DataFrame(
                    correlation_matrix, 
                    index=col_names, 
                    columns=col_names)

corr_df

Мы использовали здесь Pandas | DataFrame для более полного отображения корреляционной матрицы. Теперь нам нужны те переменные, которые сильно коррелируют. Для этого мы немного манипулируем фреймами данных.

# get a boolean dataframe where true means that 
# a pair of variables is highly correlated
highly_correlated_df = (abs(corr_df) > .8) & (corr_df < 1.0)

# get the names of the variables so we can use 
# them to slice the dataframe
correlated_vars_index = (highly_correlated_df==True).any()
correlated_var_names = correlated_vars_index[correlated_vars_index==True].index

# slice it
highly_correlated_df.loc[correlated_var_names,correlated_var_names]

Выводы

Возможные Подсказки По Выбору Модели

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

Например, из описания задачи KDD Cup 99 мы знаем, что переменная dst_host_same_src_port_rate ссылается на процент последних 100 подключений к одному и тому же порту для одного и того же хоста назначения. В нашей корреляционной матрице (и вспомогательных фреймах данных) мы находим, что эта матрица сильно и положительно коррелирует с src_bytes и srv_count . Первый-это количество байтов, отправленных от источника к месту назначения. Позднее-это количество подключений к той же службе, что и текущее соединение за последние 2 секунды. Мы могли бы решить не включать dst_host_same_src_port_rate в нашу модель, если мы включим два других, как способ уменьшить количество переменных и позже лучше интерпретировать наши модели.

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