Datasets:
We will be using datasets from the KDD Cup 1999.
This work is on reference with https://github.com/jadianes/spark-py-notebooks
Dataset Description :
Since 1999, KDD’99 has been the most wildly used data set for the evaluation of anomaly detection methods. This data set is prepared by Stolfo and is built based on the data captured in DARPA’98 IDS evaluation program . DARPA’98 is about 4 gigabytes of compressed raw (binary) tcpdump data of 7 weeks of network traffic, which can be processed into about 5 million connection records, each with about 100 bytes. The two weeks of test data have around 2 million connection records. KDD training dataset consists of approximately 4,900,000 single connection vectors each of which contains 41 features and is labeled as either normal or an attack, with exactly one specific attack type.
The simulated attacks fall in one of the following four categories:
1) Denial of Service Attack (DoS): is an attack in which the attacker makes some computing or memory resource too busy or too full to handle legitimate requests, or denies legitimate users access to a machine.
2) User to Root Attack (U2R): is a class of exploit in which the attacker starts out with access to a normal user account on the system (perhaps gained by sniffing passwords, a dictionary attack, or social engineering) and is able to exploit some vulnerability to gain root access to the system.
3) Remote to Local Attack (R2L): occurs when an attacker who has the ability to send packets to a machine over a network but who does not have an account on that machine exploits some vulnerability to gain local access as a user of that machine.
4) Probing Attack: is an attempt to gather information about a network of computers for the apparent purpose of circumventing its security controls.
This work is divided into 10 tasks which includes RDD creation,RDD sampling,RDD set operations,MLIB logistic regression,etc.
Notebooks
The following notebooks can be examined individually, although there is a more or less linear 'story' when followed in sequence. By using the same dataset they try to solve a related set of tasks with it.
RDD creation
About reading files and parallelize.
RDDs basics
A look at
map, filter, and collect.
Sampling RDDs
RDD sampling methods explained.
RDD set operations
Brief introduction to some of the RDD pseudo-set operations.
Data aggregations on RDDs
RDD actions
reduce, fold, and aggregate.
Working with key/value pair RDDs
How to deal with key/value pairs in order to aggregate and explore data.
MLlib: Basic Statistics and Exploratory Data Analysis
A notebook introducing Local Vector types, basic statistics in MLlib for Exploratory Data Analysis and model selection.
Out[23]:
Out[24]:
Conclusions and possible model selection hints
The previous dataframe showed us which variables are highly correlated. We have kept just those variables with at least one strong correlation. We can use as we please, but a good way could be to do some model selection. That is, if we have a group of variables that are highly correlated, we can keep just one of them to represent the group under the assumption that they convey similar information as predictors. Reducing the number of variables will not improve our model accuracy, but it will make it easier to understand and also more efficient to compute.
For example, from the description of the KDD Cup 99 task we know that the variable
dst_host_same_src_port_rate references the percentage of the last 100 connections to the same port, for the same destination host. In our correlation matrix (and auxiliar dataframes) we find that this one is highly and positively correlated to src_bytes and srv_count. The former is the number of bytes sent form source to destination. The later is the number of connections to the same service as the current connection in the past 2 seconds. We might decide not to include dst_host_same_src_port_rate in our model if we include the other two, as a way to reduce the number of variables and later one better interpret our models.
Later on, in those notebooks dedicated to build predictive models, we will make use of this information to build more interpretable models.
MLlib: Logistic Regression
Labeled points and Logistic Regression classification of network attacks in MLlib. Application of model selection techniques using correlation matrix and Hypothesis Testing.
MLlib: Classification with Logistic Regression
In this notebook we will use Spark's machine learning library MLlib to build a Logistic Regression classifier for network attack detection. We will use the complete KDD Cup 1999 datasets in order to test Spark capabilities with large datasets.
Additionally, we will introduce two ways of performing model selection: by using a correlation matrix and by using hypothesis testing.
Creating the RDD
In [1]:
from pyspark import SparkContext
sc =SparkContext()
In [2]:
data_file = "/home/osboxes/Python with Spark - part 1/pydata/kddcup.data.gz"
raw_data = sc.textFile(data_file)
print "Train data size is {}".format(raw_data.count())
The KDD Cup 1999 also provide test data that we will load in a separate RDD.
In [3]:
ft = urllib.urlretrieve("http://kdd.ics.uci.edu/databases/kddcup99/corrected.gz", "corrected.gz")
In [3]:
test_data_file = "/home/osboxes/Python with Spark - part 1/pydata/corrected.gz"
test_raw_data = sc.textFile(test_data_file)
print "Test data size is {}".format(test_raw_data.count())
Labeled Points
A labeled point is a local vector associated with a label/response. In MLlib, labeled points are used in supervised learning algorithms and they are stored as doubles. For binary classification, a label should be either 0 (negative) or 1 (positive).
Preparing the training data
In our case, we are interested in detecting network attacks in general. We don't need to detect which type of attack we are dealing with. Therefore we will tag each network interaction as non attack (i.e. 'normal' tag) or attack (i.e. anything else but 'normal').
In [4]:
from pyspark.mllib.regression import LabeledPoint
from numpy import array
def parse_interaction(line):
line_split = line.split(",")
# leave_out = [1,2,3,41]
clean_line_split = line_split[0:1]+line_split[4:41]
attack = 1.0
if line_split[41]=='normal.':
attack = 0.0
return LabeledPoint(attack, array([float(x) for x in clean_line_split]))
training_data = raw_data.map(parse_interaction)
Preparing the test data
Similarly, we process our test data file.
In [5]:
test_data = test_raw_data.map(parse_interaction)
Detecting network attacks using Logistic Regression
Logistic regression is widely used to predict a binary response. Spark implements two algorithms to solve logistic regression: mini-batch gradient descent and L-BFGS. L-BFGS is recommended over mini-batch gradient descent for faster convergence.
Training a classifier
In [6]:
from pyspark.mllib.classification import LogisticRegressionWithLBFGS
from time import time
# Build the model
t0 = time()
logit_model = LogisticRegressionWithLBFGS.train(training_data)
tt = time() - t0
print "Classifier trained in {} seconds".format(round(tt,3))
Evaluating the model on new data
In order to measure the classification error on our test data, we use
map on the test_data RDD and the model to predict each test point class.
In [7]:
labels_and_preds = test_data.map(lambda p: (p.label, logit_model.predict(p.features)))
Classification results are returned in pars, with the actual test label and the predicted one. This is used to calculate the classification error by using
filterand count as follows.
In [8]:
t0 = time()
test_accuracy = labels_and_preds.filter(lambda (v, p): v == p).count() / float(test_data.count())
tt = time() - t0
print "Prediction made in {} seconds. Test accuracy is {}".format(round(tt,3), round(test_accuracy,4))
That's a decent accuracy. We know that there is space for improvement with a better variable selection and also by including categorical variables (e.g. we have excluded 'protocol' and 'service').
Model selection
Model or feature selection helps us building more interpretable and efficient models (or a classifier in this case). For illustrative purposes, we will follow two different approaches, correlation matrices and hypothesis testing.
Using a correlation matrix
In a previous notebook we calculated a correlation matrix in order to find predictors that are highly correlated. There are many possible choices there in order to simplify our model. We can pick different combinations of correlated variables and leave just those that represent them. The reader can try different combinations. Here we will choose the following for illustrative purposes:
- From the description of the KDD Cup 99 task we know that the variable
dst_host_same_src_port_ratereferences the percentage of the last 100 connections to the same port, for the same destination host. In our correlation matrix (and auxiliary dataframes) we find that this one is highly and positively correlated tosrc_bytesandsrv_count. The former is the number of bytes sent form source to destination. The later is the number of connections to the same service as the current connection in the past 2 seconds. We decide not to includedst_host_same_src_port_ratein our model since we include the other two.
- Variables
serror_rateandsrv_error_rate(% of connections that have SYN errors for same host and same service respectively) are highly positively correlated. Moreover, the set of variables that they highly correlate with are pretty much the same. They look like contributing very similarly to our model. We will keep justserror_rate.
- A similar situation happens with
rerror_rateandsrv_rerror_rate(% of connections that have REJ errors) so we will keep justrerror_rate.
- Same thing with variables prefixed with
dst_host_for the previous ones (e.g.dst_host_srv_serror_rate).
We will stop here, although the reader can keep experimenting removing correlated variables has before (e.g.
same_srv_rate and diff_srv_rate are good candidates. Our list of variables we will drop includes:dst_host_same_src_port_rate, (column 35).srv_serror_rate(column 25).srv_rerror_rate(column 27).dst_host_srv_serror_rate(column 38).dst_host_srv_rerror_rate(column 40).
Evaluating the new model
Let's proceed with the evaluation of our reduced model. First we need to provide training and testing datasets containing just the selected variables. For that we will define a new function to parse the raw data that keeps just what we need.
In [9]:
def parse_interaction_corr(line):
line_split = line.split(",")
# leave_out = [1,2,3,25,27,35,38,40,41]
clean_line_split = line_split[0:1]+line_split[4:25]+line_split[26:27]+line_split[28:35]+line_split[36:38]+line_split[39:40]
attack = 1.0
if line_split[41]=='normal.':
attack = 0.0
return LabeledPoint(attack, array([float(x) for x in clean_line_split]))
corr_reduced_training_data = raw_data.map(parse_interaction_corr)
corr_reduced_test_data = test_raw_data.map(parse_interaction_corr)
Note: when selecting elements in the split, a list comprehension with a
leave_out list for filtering is more Pythonic than slicing and concatenation indeed, but we have found it less efficient. This is very important when dealing with large datasets. The parse_interaction functions will be called for every element in the RDD, so we need to make them as efficient as possible.
Now we can train the model.
In [10]:
# Build the model
t0 = time()
logit_model_2 = LogisticRegressionWithLBFGS.train(corr_reduced_training_data)
tt = time() - t0
print "Classifier trained in {} seconds".format(round(tt,3))
And evaluate its accuracy on the test data.
In [11]:
labels_and_preds = corr_reduced_test_data.map(lambda p: (p.label, logit_model_2.predict(p.features)))
t0 = time()
test_accuracy = labels_and_preds.filter(lambda (v, p): v == p).count() / float(corr_reduced_test_data.count())
tt = time() - t0
print "Prediction made in {} seconds. Test accuracy is {}".format(round(tt,3), round(test_accuracy,4))
As expected, we have reduced accuracy and also training time. However this doesn't seem a good trade! At least not for logistic regression and considering the predictors we decided to leave out. We have lost quite a lot of accuracy and have not gained a lot of execution time during training. Moreover prediction time didn't improve.
Using hypothesis testing
Hypothesis testing is a powerful tool in statistical inference and learning to determine whether a result is statistically significant. MLlib supports Pearson's chi-squared (χ2) tests for goodness of fit and independence. The goodness of fit test requires an input type of
Vector, whereas the independence test requires a Matrix as input. Moreover, MLlib also supports the input type RDD[LabeledPoint] to enable feature selection via chi-squared independence tests. Again, these methods are part of the Statistics package.
In our case we want to perform some sort of feature selection, so we will provide an RDD of
LabeledPoint. Internally, MLlib will calculate a contingency matrix and perform the Persons's chi-squared (χ2) test. Features need to be categorical. Real-valued features will be treated as categorical in each of its different values. There is a limit of 1000 different values, so we need either to leave out some features or categorise them. In this case, we will consider just features that either take boolean values or just a few different numeric values in our dataset. We could overcome this limitation by defining a more complex parse_interaction function that categorises each feature properly.
In [12]:
feature_names = ["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"]
In [13]:
def parse_interaction_categorical(line):
line_split = line.split(",")
clean_line_split = line_split[6:41]
attack = 1.0
if line_split[41]=='normal.':
attack = 0.0
return LabeledPoint(attack, array([float(x) for x in clean_line_split]))
training_data_categorical = raw_data.map(parse_interaction_categorical)
In [14]:
from pyspark.mllib.stat import Statistics
chi = Statistics.chiSqTest(training_data_categorical)
Now we can check the resulting values after putting them into a Pandas data frame.
In [15]:
import pandas as pd
pd.set_option('display.max_colwidth', 30)
records = [(result.statistic, result.pValue) for result in chi]
chi_df = pd.DataFrame(data=records, index= feature_names, columns=["Statistic","p-value"])
chi_df
Out[15]:
From that we conclude that predictors
land and num_outbound_cmds could be removed from our model without affecting our accuracy dramatically. Let's try this.Evaluating the new model
So the only modification to our first
parse_interaction function will be to remove columns 6 and 19, corresponding to the two predictors that we want not to be part of our model.
In [16]:
def parse_interaction_chi(line):
line_split = line.split(",")
# leave_out = [1,2,3,6,19,41]
clean_line_split = line_split[0:1] + line_split[4:6] + line_split[7:19] + line_split[20:41]
attack = 1.0
if line_split[41]=='normal.':
attack = 0.0
return LabeledPoint(attack, array([float(x) for x in clean_line_split]))
training_data_chi = raw_data.map(parse_interaction_chi)
test_data_chi = test_raw_data.map(parse_interaction_chi)
Now we build the logistic regression classifier again.
In [18]:
# Build the model
t0 = time()
logit_model_chi = LogisticRegressionWithLBFGS.train(training_data_chi)
tt = time() - t0
print "Classifier trained in {} seconds".format(round(tt,3))
And evaluate in test data.
In [19]:
labels_and_preds = test_data_chi.map(lambda p: (p.label, logit_model_chi.predict(p.features)))
t0 = time()
test_accuracy = labels_and_preds.filter(lambda (v, p): v == p).count() / float(test_data_chi.count())
tt = time() - t0
print "Prediction made in {} seconds. Test accuracy is {}".format(round(tt,3), round(test_accuracy,4))
So we can see that, by using hypothesis testing, we have been able to remove two predictors without diminishing testing accuracy at all. Training time improved a bit as well. This might now seem like a big model reduction, but it is something when dealing with big data sources. Moreover, we should be able to categorise those five predictors we have left out for different reasons and, either include them in the model or leave them out if they aren't statistically significant.
Additionally, we could try to remove some of those predictors that are highly correlated, trying not to reduce accuracy too much. In the end, we should end up with a model easier to understand and use.
MLlib: Decision Trees
Use of tree-based methods and how they help explaining models and feature selection.
MLlib: Decision Trees
In this notebook we will use Spark's machine learning library MLlib to build a Decision Tree classifier for network attack detection. We will use the completeKDD Cup 1999 datasets in order to test Spark capabilities with large datasets.
Decision trees are a popular machine learning tool in part because they are easy to interpret, handle categorical features, extend to the multiclass classification setting, do not require feature scaling, and are able to capture non-linearities and feature interactions. In this notebook, we will first train a classification tree including every single predictor. Then we will use our results to perform model selection. Once we find out the most important ones (the main splits in the tree) we will build a minimal tree using just three of them (the first two levels of the tree in order to compare performance and accuracy.
Creating the RDD
As we said, this time we will use the complete dataset provided for the KDD Cup 1999, containing nearly half million network interactions. The file is provided as a Gzip file that we will download locally.
In [2]:
from pyspark import SparkContext
sc =SparkContext()
In [3]:
import urllib
In [2]:
f = urllib.urlretrieve ("http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data.gz", "kddcup.data.gz")
In [4]:
data_file = "/home/osboxes/Python with Spark - part 1/pydata/kddcup.data.gz"
raw_data = sc.textFile(data_file)
print "Train data size is {}".format(raw_data.count())
The KDD Cup 1999 also provide test data that we will load in a separate RDD.
In [4]:
ft = urllib.urlretrieve("http://kdd.ics.uci.edu/databases/kddcup99/corrected.gz", "corrected.gz")
In [5]:
test_data_file = "/home/osboxes/Python with Spark - part 1/pydata/corrected.gz"
test_raw_data = sc.textFile(test_data_file)
print "Test data size is {}".format(test_raw_data.count())
Detecting network attacks using Decision Trees
In this section we will train a classification tree that, as we did with logistic regression, will predict if a network interaction is either
normal or attack.
Training a classification tree using MLlib requires some parameters:
- Training data
- Num classes
- Categorical features info: a map from column to categorical variables arity. This is optional, although it should increase model accuracy. However it requires that we know the levels in our categorical variables in advance. second we need to parse our data to convert labels to integer values within the arity range.
- Impurity metric
- Tree maximum depth
- And tree maximum number of bins
In the next section we will see how to obtain all the labels within a dataset and convert them to numerical factors.
Preparing the data
As we said, in order to benefits from trees ability to seamlessly with categorical variables, we need to convert them to numerical factors. But first we need to obtain all the possible levels. We will use set transformations on a csv parsed RDD.
In [6]:
from pyspark.mllib.regression import LabeledPoint
from numpy import array
csv_data = raw_data.map(lambda x: x.split(","))
test_csv_data = test_raw_data.map(lambda x: x.split(","))
protocols = csv_data.map(lambda x: x[1]).distinct().collect()
services = csv_data.map(lambda x: x[2]).distinct().collect()
flags = csv_data.map(lambda x: x[3]).distinct().collect()
And now we can use this Python lists in our
create_labeled_point function. If a factor level is not in the training data, we assign an especial level. Remember that we cannot use testing data for training our model, not even the factor levels. The testing data represents the unknown to us in a real case.
In [7]:
def create_labeled_point(line_split):
# leave_out = [41]
clean_line_split = line_split[0:41]
# convert protocol to numeric categorical variable
try:
clean_line_split[1] = protocols.index(clean_line_split[1])
except:
clean_line_split[1] = len(protocols)
# convert service to numeric categorical variable
try:
clean_line_split[2] = services.index(clean_line_split[2])
except:
clean_line_split[2] = len(services)
# convert flag to numeric categorical variable
try:
clean_line_split[3] = flags.index(clean_line_split[3])
except:
clean_line_split[3] = len(flags)
# convert label to binary label
attack = 1.0
if line_split[41]=='normal.':
attack = 0.0
return LabeledPoint(attack, array([float(x) for x in clean_line_split]))
training_data = csv_data.map(create_labeled_point)
test_data = test_csv_data.map(create_labeled_point)
Training a classifier
We are now ready to train our classification tree. We will keep the
maxDepth value small. This will lead to smaller accuracy, but we will obtain less splits so later on we can better interpret the tree. In a production system we will try to increase this value in order to find a better accuracy.
In [8]:
from pyspark.mllib.tree import DecisionTree, DecisionTreeModel
from time import time
# Build the model
t0 = time()
tree_model = DecisionTree.trainClassifier(training_data, numClasses=2,
categoricalFeaturesInfo={1: len(protocols), 2: len(services), 3: len(flags)},
impurity='gini', maxDepth=4, maxBins=100)
tt = time() - t0
print "Classifier trained in {} seconds".format(round(tt,3))
Evaluating the model
In order to measure the classification error on our test data, we use
map on the test_data RDD and the model to predict each test point class.
In [9]:
predictions = tree_model.predict(test_data.map(lambda p: p.features))
labels_and_preds = test_data.map(lambda p: p.label).zip(predictions)
Classification results are returned in pars, with the actual test label and the predicted one. This is used to calculate the classification error by using
filterand count as follows.
In [10]:
t0 = time()
test_accuracy = labels_and_preds.filter(lambda (v, p): v == p).count() / float(test_data.count())
tt = time() - t0
print "Prediction made in {} seconds. Test accuracy is {}".format(round(tt,3), round(test_accuracy,4))
NOTE: the zip transformation doesn't work properly with pySpark 1.2.1. It does in 1.3
Interpreting the model
Understanding our tree splits is a great exercise in order to explain our classification labels in terms of predictors and the values they take. Using the
toDebugString method in our three model we can obtain a lot of information regarding splits, nodes, etc.
In [11]:
print "Learned classification tree model:"
print tree_model.toDebugString()
For example, a network interaction with the following features (see description here) will be classified as an attack by our model:
count, the number of connections to the same host as the current connection in the past two seconds, being greater than 32.dst_bytes, the number of data bytes from destination to source, is 0.serviceis neither level 0 nor 52.logged_inis false.
From our services list we know that:
In [12]:
print "Service 0 is {}".format(services[0])
print "Service 52 is {}".format(services[52])
So we can characterise network interactions with more than 32 connections to the same server in the last 2 seconds, transferring zero bytes from destination to source, where service is neither urp_i nor tftp_u, and not logged in, as network attacks. A similar approach can be used for each tree terminal node.
We can see that
count is the first node split in the tree. Remember that each partition is chosen greedily by selecting the best split from a set of possible splits, in order to maximize the information gain at a tree node (see more here). At a second level we find variables flag (normal or error status of the connection) and dst_bytes (the number of data bytes from destination to source) and so on.
This explaining capability of a classification (or regression) tree is one of its main benefits. Understaining data is a key factor to build better models.
Building a minimal model using the three main splits
So now that we know the main features predicting a network attack, thanks to our classification tree splits, let's use them to build a minimal classification tree with just the main three variables:
count, dst_bytes, and flag.
We need to define the appropriate function to create labeled points.
In [13]:
def create_labeled_point_minimal(line_split):
# leave_out = [41]
clean_line_split = line_split[3:4] + line_split[5:6] + line_split[22:23]
# convert flag to numeric categorical variable
try:
clean_line_split[0] = flags.index(clean_line_split[0])
except:
clean_line_split[0] = len(flags)
# convert label to binary label
attack = 1.0
if line_split[41]=='normal.':
attack = 0.0
return LabeledPoint(attack, array([float(x) for x in clean_line_split]))
training_data_minimal = csv_data.map(create_labeled_point_minimal)
test_data_minimal = test_csv_data.map(create_labeled_point_minimal)
That we use to train the model.
In [14]:
# Build the model
t0 = time()
tree_model_minimal = DecisionTree.trainClassifier(training_data_minimal, numClasses=2,
categoricalFeaturesInfo={0: len(flags)},
impurity='gini', maxDepth=3, maxBins=32)
tt = time() - t0
print "Classifier trained in {} seconds".format(round(tt,3))
Now we can predict on the testing data and calculate accuracy.
In [15]:
predictions_minimal = tree_model_minimal.predict(test_data_minimal.map(lambda p: p.features))
labels_and_preds_minimal = test_data_minimal.map(lambda p: p.label).zip(predictions_minimal)
In [16]:
t0 = time()
test_accuracy = labels_and_preds_minimal.filter(lambda (v, p): v == p).count() / float(test_data_minimal.count())
tt = time() - t0
print "Prediction made in {} seconds. Test accuracy is {}".format(round(tt,3), round(test_accuracy,4))
So we have trained a classification tree with just the three most important predictors, in half of the time, and with a not so bad accuracy. In fact, a classification tree is a very good model selection tool!
Spark SQL: structured processing for Data Analysis
In this notebook a schema is inferred for our network interactions dataset. Based on that, we use Spark's SQL
DataFrameabstraction to perform a more structured exploratory data analysis.Spark SQL and Data Frames
This notebook will introduce Spark capabilities to deal with data in a structured way. Basically, everything turns around the concept of Data Frame and usingSQL language to query them. We will see how the data frame abstraction, very popular in other data analytics ecosystems (e.g. R and Python/Pandas), it is very powerful when performing exploratory data analysis. In fact, it is very easy to express data queries when used together with the SQL language. Moreover, Spark distributes this column-based data structure transparently, in order to make the querying process as efficient as possible.
Creating the RDD
As we did in previous notebooks, we will use the reduced dataset (10 percent) provided for the KDD Cup 1999, containing nearly half million nework interactions. The file is provided as a Gzip file that we will download locally.
In [1]:
from pyspark import SparkContext
sc =SparkContext()
In [2]:
data_file = "/home/osboxes/Python with Spark - part 1/pydata/kddcup.data_10_percent.gz"
raw_data = sc.textFile(data_file).cache()
Getting a Data Frame
A Spark
DataFrame is a distributed collection of data organized into named columns. It is conceptually equivalent to a table in a relational database or a data frame in R or Pandas. They can be constructed from a wide array of sources such as a existing RDD in our case.
The entry point into all SQL functionality in Spark is the
SQLContext class. To create a basic instance, all we need is a SparkContext reference. Since we are running Spark in shell mode (using pySpark) we can use the global context object sc for this purpose.
In [3]:
from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)
Inferring the schema
With a
SQLContext, we are ready to create a DataFrame from our existing RDD. But first we need to tell Spark SQL the schema in our data.
Spark SQL can convert an RDD of
Row objects to a DataFrame. Rows are constructed by passing a list of key/value pairs as kwargs to the Row class. The keys define the column names, and the types are inferred by looking at the first row. Therefore, it is important that there is no missing data in the first row of the RDD in order to properly infer the schema.
In our case, we first need to split the comma separated data, and then use the information in KDD's 1999 task description to obtain the column names.
In [4]:
from pyspark.sql import Row
csv_data = raw_data.map(lambda l: l.split(","))
row_data = csv_data.map(lambda p: Row(
duration=int(p[0]),
protocol_type=p[1],
service=p[2],
flag=p[3],
src_bytes=int(p[4]),
dst_bytes=int(p[5])
)
)
Once we have our RDD of
Row we can infer and register the schema.
In [5]:
interactions_df = sqlContext.createDataFrame(row_data)
interactions_df.registerTempTable("interactions")
Now we can run SQL queries over our data frame that has been registered as a table.
In [6]:
# Select tcp network interactions with more than 1 second duration and no transfer from destination
tcp_interactions = sqlContext.sql("""
SELECT duration, dst_bytes FROM interactions WHERE protocol_type = 'tcp' AND duration > 1000 AND dst_bytes = 0
""")
tcp_interactions.show()
The results of SQL queries are RDDs and support all the normal RDD operations.
In [7]:
# Output duration together with dst_bytes
tcp_interactions_out = tcp_interactions.map(lambda p: "Duration: {}, Dest. bytes: {}".format(p.duration, p.dst_bytes))
for ti_out in tcp_interactions_out.collect():
print ti_out
We can easily have a look at our data frame schema using
printSchema.
In [8]:
interactions_df.printSchema()
Queries as DataFrame operations
Spark
DataFrame provides a domain-specific language for structured data manipulation. This language includes methods we can concatenate in order to do selection, filtering, grouping, etc. For example, let's say we want to count how many interactions are there for each protocol type. We can proceed as follows.
In [9]:
from time import time
t0 = time()
interactions_df.select("protocol_type", "duration", "dst_bytes").groupBy("protocol_type").count().show()
tt = time() - t0
print "Query performed in {} seconds".format(round(tt,3))
Now imagine that we want to count how many interactions last more than 1 second, with no data transfer from destination, grouped by protocol type. We can just add to filter calls to the previous.
In [10]:
t0 = time()
interactions_df.select("protocol_type", "duration", "dst_bytes").filter(interactions_df.duration>1000).filter(interactions_df.dst_bytes==0).groupBy("protocol_type").count().show()
tt = time() - t0
print "Query performed in {} seconds".format(round(tt,3))
We can use this to perform some exploratory data analysis. Let's count how many attack and normal interactions we have. First we need to add the label column to our data.
In [11]:
def get_label_type(label):
if label!="normal.":
return "attack"
else:
return "normal"
row_labeled_data = csv_data.map(lambda p: Row(
duration=int(p[0]),
protocol_type=p[1],
service=p[2],
flag=p[3],
src_bytes=int(p[4]),
dst_bytes=int(p[5]),
label=get_label_type(p[41])
)
)
interactions_labeled_df = sqlContext.createDataFrame(row_labeled_data)
This time we don't need to register the schema since we are going to use the OO query interface.
Let's check the previous actually works by counting attack and normal data in our data frame.
In [12]:
t0 = time()
interactions_labeled_df.select("label").groupBy("label").count().show()
tt = time() - t0
print "Query performed in {} seconds".format(round(tt,3))
Now we want to count them by label and protocol type, in order to see how important the protocol type is to detect when an interaction is or not an attack.
In [13]:
t0 = time()
interactions_labeled_df.select("label", "protocol_type").groupBy("label", "protocol_type").count().show()
tt = time() - t0
print "Query performed in {} seconds".format(round(tt,3))
At first sight it seems that udp interactions are in lower proportion between network attacks versus other protocol types.
And we can do much more sophisticated groupings. For example, add to the previous a "split" based on data transfer from target.
In [14]:
t0 = time()
interactions_labeled_df.select("label", "protocol_type", "dst_bytes").groupBy("label", "protocol_type", interactions_labeled_df.dst_bytes==0).count().show()
tt = time() - t0
print "Query performed in {} seconds".format(round(tt,3))
We see how relevant is this new split to determine if a network interaction is an attack.
We will stop here, but we can see how powerful this type of queries are in order to explore our data. Actually we can replicate all the splits we saw in previous notebooks, when introducing classification trees, just by selecting, groping, and filtering our dataframe. For a more detailed (but less real-world) list of Spark's
DataFrame operations and data sources, have a look at the official documentation here.
No comments:
Post a Comment