Compression Codec and File Format

PySpark - Spark SQL

Script:

import findspark
findspark.init('/Users/donghua/spark-2.4.0-bin-hadoop2.7')

from pyspark import SparkContext
sc = SparkContext('local[2]', 'Handson PySpark Chapter 6')

from pyspark.sql import Row, SQLContext, SparkSession
# SQLContext replaced by SparkSession since 2.0, SQLContext can be created through
# sql_context = SQLConect(sc)
spark = SparkSession(sc).builder.getOrCreate()

raw_data = sc.textFile('file:///tmp/kddcup.data_10_percent.gz')
csv = raw_data.map(lambda x: x.split(','))

print(csv.take(1))

rows = csv.map(lambda p: Row(duration=int(p[0]), protocol=p[1],service=p[2]))

rows.take(1)

df=spark.createDataFrame(rows)

df.printSchema()

df.show(5)

df.registerTempTable('rdd')

spark.sql("""SELECT duration from rdd WHERE protocol='tcp' and duration > 2000""")

df.select("duration").filter("protocol='tcp'").filter("duration>2000").show(5)

Output (Jupyter):

import findspark
findspark.init('/Users/donghua/spark-2.4.0-bin-hadoop2.7')

from pyspark import SparkContext
sc = SparkContext('local[2]', 'Handson PySpark Chapter 6')

from pyspark.sql import Row, SQLContext, SparkSession
# SQLContext replaced by SparkSession since 2.0, SQLContext can be created through
# sql_context = SQLConect(sc)
spark = SparkSession(sc).builder.getOrCreate()

raw_data = sc.textFile('file:///tmp/kddcup.data_10_percent.gz')
csv = raw_data.map(lambda x: x.split(','))

print(csv.take(1))

[['0', 'tcp', 'http', 'SF', '181', '5450', '0', '0', '0', '0', '0', '1', '0', '0', '0', '0', '0', '0', '0', '0', '0', '0', '8', '8', '0.00', '0.00', '0.00', '0.00', '1.00', '0.00', '0.00', '9', '9', '1.00', '0.00', '0.11', '0.00', '0.00', '0.00', '0.00', '0.00', 'normal.']]

rows = csv.map(lambda p: Row(duration=int(p[0]), protocol=p[1],service=p[2]))

rows.take(1)

[Row(duration=0, protocol='tcp', service='http')]

df=spark.createDataFrame(rows)

df.printSchema()

root
 |-- duration: long (nullable = true)
 |-- protocol: string (nullable = true)
 |-- service: string (nullable = true)

df.show(5)

+--------+--------+-------+
|duration|protocol|service|
+--------+--------+-------+
|       0|     tcp|   http|
|       0|     tcp|   http|
|       0|     tcp|   http|
|       0|     tcp|   http|
|       0|     tcp|   http|
+--------+--------+-------+
only showing top 5 rows

df.registerTempTable('rdd')

spark.sql("""SELECT duration from rdd WHERE protocol='tcp' and duration > 2000""")

df.select("duration").filter("protocol='tcp'").filter("duration>2000").show(5)

+--------+
|duration|
+--------+
|   12454|
|   10774|
|   13368|
|   10350|
|   10409|
+--------+
only showing top 5 rows

Some of the code referenced from hands-pyspark-big-data-analysis-video

PySpark - MLLib

Script:

import findspark
findspark.init('/Users/donghua/spark-2.4.0-bin-hadoop2.7')

from pyspark import SparkContext
from pyspark.sql import SparkSession

spark=SparkSession(SparkContext()).builder.master('local[2]').appName('Handson PySpark Chapter 5').getOrCreate()

sc = spark.sparkContext
sc.setLogLevel('debug')
sc.getConf().getAll()


import urllib.request
url = 'http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data_10_percent.gz'
localfile = '/tmp/kddcup.data_10_percent.gz'
f = urllib.request.urlretrieve(url,localfile)

raw_data = sc.textFile('file:///tmp/kddcup.data_10_percent.gz')
csv = raw_data.map(lambda x: x.split(','))
duration = raw_data.map(lambda x: [int(x[0])])

from pyspark.mllib.stat import Statistics
summary = Statistics.colStats(duration)
summary.mean()[0]
summary.count()

metrics = csv.map(lambda x: [x[0],x[4],x[5]])
metrics.take(2)

Statistics.corr(metrics, method="spearman")

Statistics.corr(metrics, method="pearson")

from pyspark.mllib.linalg import Vectors

visitors_freq = Vectors.dense(0.13, 0.61, 0.8, 0.5, 0.3)
print(Statistics.chiSqTest(visitors_freq))

visitors_freq = Vectors.dense(0.13, 0.61, 0.8, 0.5, 8)
print(Statistics.chiSqTest(visitors_freq))

print(Statistics.chiSqTest(duration.collect()))

spark.stop()

Output (Jupyter):

import findspark
findspark.init('/Users/donghua/spark-2.4.0-bin-hadoop2.7')

from pyspark import SparkContext
from pyspark.sql import SparkSession

spark=SparkSession(SparkContext()).builder.master('local[2]').appName('Handson PySpark Chapter 5').getOrCreate()

sc = spark.sparkContext
sc.setLogLevel('debug')

sc.getConf().getAll()

[('spark.sql.warehouse.dir', '/user/hive/warehouse'),
 ('spark.rdd.compress', 'True'),
 ('spark.app.id', 'local-1553755489097'),
 ('spark.driver.port', '51208'),
 ('spark.serializer.objectStreamReset', '100'),
 ('spark.executor.id', 'driver'),
 ('spark.submit.deployMode', 'client'),
 ('spark.app.name', 'Handson PySpark Chapter 5'),
 ('spark.driver.host', '192.168.31.177'),
 ('spark.ui.showConsoleProgress', 'true'),
 ('spark.master', 'local[2]')]

import urllib.request
url = 'http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data_10_percent.gz'
localfile = '/tmp/kddcup.data_10_percent.gz'
f = urllib.request.urlretrieve(url,localfile)

raw_data = sc.textFile('file:///tmp/kddcup.data_10_percent.gz')
csv = raw_data.map(lambda x: x.split(','))
duration = raw_data.map(lambda x: [int(x[0])])

from pyspark.mllib.stat import Statistics
summary = Statistics.colStats(duration)

summary.mean()[0]

0.06611054995637812

summary.count()

494021

metrics = csv.map(lambda x: [x[0],x[4],x[5]])
metrics.take(2)

[['0', '181', '5450'], ['0', '239', '486']]

Statistics.corr(metrics, method="spearman")

array([[ 1.        ,  0.01419628,  0.29918926],
       [ 0.01419628,  1.        , -0.16793059],
       [ 0.29918926, -0.16793059,  1.        ]])

Statistics.corr(metrics, method="pearson")

array([[ 1.00000000e+00,  4.25823027e-03,  5.43953448e-03],
       [ 4.25823027e-03,  1.00000000e+00, -1.59677215e-06],
       [ 5.43953448e-03, -1.59677215e-06,  1.00000000e+00]])

from pyspark.mllib.linalg import Vectors

visitors_freq = Vectors.dense(0.13, 0.61, 0.8, 0.5, 0.3)
print(Statistics.chiSqTest(visitors_freq))

Chi squared test summary:
method: pearson
degrees of freedom = 4 
statistic = 0.5852136752136753 
pValue = 0.9646925263439344 
No presumption against null hypothesis: observed follows the same distribution as expected..

visitors_freq = Vectors.dense(0.13, 0.61, 0.8, 0.5, 8)
print(Statistics.chiSqTest(visitors_freq))

Chi squared test summary:
method: pearson
degrees of freedom = 4 
statistic = 22.469462151394424 
pValue = 1.6158934330234853E-4 
Very strong presumption against null hypothesis: observed follows the same distribution as expected..

print(Statistics.chiSqTest(duration.collect()))

Chi squared test summary:
method: pearson
degrees of freedom = 494020 
statistic = 2041502.1434188513 
pValue = 0.0 
Very strong presumption against null hypothesis: observed follows the same distribution as expected..

spark.stop()

Some of the code referenced from hands-pyspark-big-data-analysis-video

PySpark - Aggregating and Summarising Data

Script:

import findspark
findspark.init('/Users/donghua/spark-2.4.0-bin-hadoop2.7')

from pyspark import SparkContext
from pyspark.sql import SparkSession

sc = SparkContext('local[2]','Handson Spark Chapter 4')
spark = SparkSession(sc).builder.getOrCreate()

sc.setLogLevel('debug')

import urllib.request
url = 'http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data_10_percent.gz'
localfile = '/tmp/kddcup.data_10_percent.gz'
f = urllib.request.urlretrieve(url,localfile)

raw_data = sc.textFile('file:///tmp/kddcup.data_10_percent.gz')

csv = raw_data.map(lambda x: x.split(','))
normal_data = csv.filter(lambda x: x[41]=='normal.')
duration = normal_data.map(lambda x: int(x[0]))
total_duration = duration.reduce(lambda a,b: a+b)
print("total duration: {}, average duration: {}".format(total_duration, total_duration/normal_data.count()))

# db is a tube of (total, count) in this context
duration_count = duration.aggregate(
 (0,0),
 lambda db, new_value: (db[0]+new_value, db[1]+1),
 lambda db1, db2: (db1[0]+db2[0], db1[1]+db2[1])
)

print("average duration: {}".format(duration_count[0]/duration_count[1]))

# below is same as kv = csv.keyBy(lambda x: x[41])
kv = csv.map(lambda x: (x[41],x))
print(kv.take(1))

kv_duration = csv.map(lambda x: (x[41],float(x[0]))).reduceByKey(lambda x,y: x+y)
kv_duration.collect()

kv.countByKey()

sc.stop()

Output (Jupyter):

import findspark
findspark.init('/Users/donghua/spark-2.4.0-bin-hadoop2.7')

from pyspark import SparkContext
from pyspark.sql import SparkSession

sc = SparkContext('local[2]','Handson Spark Chapter 4')
spark = SparkSession(sc).builder.getOrCreate()

sc.setLogLevel('debug')

import urllib.request
url = 'http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data_10_percent.gz'
localfile = '/tmp/kddcup.data_10_percent.gz'
f = urllib.request.urlretrieve(url,localfile)

raw_data = sc.textFile('file:///tmp/kddcup.data_10_percent.gz')

csv = raw_data.map(lambda x: x.split(','))
normal_data = csv.filter(lambda x: x[41]=='normal.')
duration = normal_data.map(lambda x: int(x[0]))
total_duration = duration.reduce(lambda a,b: a+b)
print("total duration: {}, average duration: {}".format(total_duration, total_duration/normal_data.count()))

total duration: 21075991, average duration: 216.65732231336992

# db is a tube of (total, count) in this context
duration_count = duration.aggregate(
 (0,0),
 lambda db, new_value: (db[0]+new_value, db[1]+1),
 lambda db1, db2: (db1[0]+db2[0], db1[1]+db2[1])
)

print("average duration: {}".format(duration_count[0]/duration_count[1]))

average duration: 216.65732231336992

# below is same as kv = csv.keyBy(lambda x: x[41])
kv = csv.map(lambda x: (x[41],x))
print(kv.take(1))

[('normal.', ['0', 'tcp', 'http', 'SF', '181', '5450', '0', '0', '0', '0', '0', '1', '0', '0', '0', '0', '0', '0', '0', '0', '0', '0', '8', '8', '0.00', '0.00', '0.00', '0.00', '1.00', '0.00', '0.00', '9', '9', '1.00', '0.00', '0.11', '0.00', '0.00', '0.00', '0.00', '0.00', 'normal.'])]

kv_duration = csv.map(lambda x: (x[41],float(x[0]))).reduceByKey(lambda x,y: x+y)
kv_duration.collect()

[('normal.', 21075991.0),
 ('buffer_overflow.', 2751.0),
 ('loadmodule.', 326.0),
 ('perl.', 124.0),
 ('neptune.', 0.0),
 ('smurf.', 0.0),
 ('guess_passwd.', 144.0),
 ('pod.', 0.0),
 ('teardrop.', 0.0),
 ('portsweep.', 1991911.0),
 ('ipsweep.', 43.0),
 ('land.', 0.0),
 ('ftp_write.', 259.0),
 ('back.', 284.0),
 ('imap.', 72.0),
 ('satan.', 64.0),
 ('phf.', 18.0),
 ('nmap.', 0.0),
 ('multihop.', 1288.0),
 ('warezmaster.', 301.0),
 ('warezclient.', 627563.0),
 ('spy.', 636.0),
 ('rootkit.', 1008.0)]

kv.countByKey()

defaultdict(int,
            {'normal.': 97278,
             'buffer_overflow.': 30,
             'loadmodule.': 9,
             'perl.': 3,
             'neptune.': 107201,
             'smurf.': 280790,
             'guess_passwd.': 53,
             'pod.': 264,
             'teardrop.': 979,
             'portsweep.': 1040,
             'ipsweep.': 1247,
             'land.': 21,
             'ftp_write.': 8,
             'back.': 2203,
             'imap.': 12,
             'satan.': 1589,
             'phf.': 4,
             'nmap.': 231,
             'multihop.': 7,
             'warezmaster.': 20,
             'warezclient.': 1020,
             'spy.': 2,
             'rootkit.': 10})

sc.stop()

Some of the code referenced from hands-pyspark-big-data-analysis-video

PySpark - Data Cleansing and Wrangling

Script:

import findspark
findspark.init('/Users/donghua/spark-2.4.0-bin-hadoop2.7')

from pyspark import SparkContext
from pyspark.sql import SparkSession

sc = SparkContext('local[2]','Pyspark Handson - Chapter 3')
spark = SparkSession(sc).builder.getOrCreate()

sc.setLogLevel('debug')

import urllib.request
url = 'http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data_10_percent.gz'
localfile = '/tmp/kddcup.data_10_percent.gz'
f = urllib.request.urlretrieve(url,localfile)

from time import time
from IPython.core.magics.execution import _format_time as format_delta

start_time = time()

raw_data = sc.textFile('file:///tmp/kddcup.data_10_percent.gz')

stop_time = time()
print('time: {}'.format(format_delta(stop_time-start_time)))


# withReplacement – can elements be sampled multiple times (replaced when sampled out)
# fraction – probability that each element is chosen; fraction must be [0, 1]
# seed – seed for the random number generator

sampled = raw_data.sample(False,0.1,42)
contains_normal_sample = sampled.map(lambda x: x.split(",")).    filter(lambda x: "normal" in x)

contains_normal = raw_data.map(lambda x: x.split(',')).    filter(lambda x: "normal" in x)
# Measure timing for the count operaiton
start_time = time()
contains_normal.count()
stop_time = time()
print('time: {}'.format(format_delta(stop_time-start_time)))


start_time = time()

contains_normal_sample.count()

stop_time = time()
print('time: {}'.format(format_delta(stop_time-start_time)))


start_time = time()

contains_normal.cache()

contains_normal.count()

stop_time = time()
print('time: {}'.format(format_delta(stop_time-start_time)))

raw_data.count()

data_in_memory = raw_data.takeSample(False, 10000, 42)
type(data_in_memory)

contains_normal_py =[line.split(',') for line in data_in_memory if "normal" in line]


len(contains_normal_py)
print(data_in_memory[0])
print(contains_normal_py[0])


normal_sample = sampled.filter(lambda line: "normal" in line)

print (sampled.count(), normal_sample.count())

non_normal_sample = sampled.subtract(normal_sample)

non_normal_sample.count()

feature_1 = sampled.map(lambda line: line.split(',')).  map(lambda features: features[1]).distinct()
feature_1.collect()

feature_2 = sampled.map(lambda line: line.split(',')).  map(lambda features: features[2]).distinct()
print(feature_2.collect())

Output (Jupyter):

import findspark
findspark.init('/Users/donghua/spark-2.4.0-bin-hadoop2.7')

from pyspark import SparkContext
from pyspark.sql import SparkSession

sc = SparkContext('local[2]','Pyspark Handson - Chapter 3')
spark = SparkSession(sc).builder.getOrCreate()

sc.setLogLevel('debug')

import urllib.request
url = 'http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data_10_percent.gz'
localfile = '/tmp/kddcup.data_10_percent.gz'
f = urllib.request.urlretrieve(url,localfile)

from time import time
from IPython.core.magics.execution import _format_time as format_delta

start_time = time()

raw_data = sc.textFile('file:///tmp/kddcup.data_10_percent.gz')

stop_time = time()
print('time: {}'.format(format_delta(stop_time-start_time)))

time: 88.8 ms

# withReplacement – can elements be sampled multiple times (replaced when sampled out)
# fraction – probability that each element is chosen; fraction must be [0, 1]
# seed – seed for the random number generator

sampled = raw_data.sample(False,0.1,42)
contains_normal_sample = sampled.map(lambda x: x.split(",")).\
    filter(lambda x: "normal" in x)

contains_normal = raw_data.map(lambda x: x.split(',')).\
    filter(lambda x: "normal" in x)

start_time = time()

contains_normal.count()

stop_time = time()
print('time: {}'.format(format_delta(stop_time-start_time)))

time: 3.71 s

start_time = time()

contains_normal_sample.count()

stop_time = time()
print('time: {}'.format(format_delta(stop_time-start_time)))

time: 1.21 s

start_time = time()

contains_normal.cache()

contains_normal.count()

stop_time = time()
print('time: {}'.format(format_delta(stop_time-start_time)))

time: 86.1 ms

raw_data.count()

494021

data_in_memory = raw_data.takeSample(False, 10000, 42)
type(data_in_memory)

list

contains_normal_py =[line.split(',') for line in data_in_memory if "normal" in line]

len(contains_normal_py)

1998

print(data_in_memory[0])

0,icmp,ecr_i,SF,520,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,511,511,0.00,0.00,0.00,0.00,1.00,0.00,0.00,255,255,1.00,0.00,1.00,0.00,0.00,0.00,0.00,0.00,smurf.

print(contains_normal_py[0])

['0', 'tcp', 'http', 'SF', '223', '9121', '0', '0', '0', '0', '0', '1', '0', '0', '0', '0', '0', '0', '0', '0', '0', '0', '7', '8', '0.00', '0.00', '0.00', '0.00', '1.00', '0.00', '0.25', '72', '255', '1.00', '0.00', '0.01', '0.04', '0.00', '0.00', '0.00', '0.00', 'normal.']

normal_sample = sampled.filter(lambda line: "normal" in line)

print (sampled.count(), normal_sample.count())

49387 9733

non_normal_sample = sampled.subtract(normal_sample)

non_normal_sample.count()

39654

feature_1 = sampled.map(lambda line: line.split(',')).\
  map(lambda features: features[1]).distinct()

feature_1.collect()

['tcp', 'icmp', 'udp']

feature_2 = sampled.map(lambda line: line.split(',')).\
  map(lambda features: features[2]).distinct()

print(feature_2.collect())

['http', 'smtp', 'auth', 'ecr_i', 'finger', 'ftp', 'domain_u', 'ntp_u', 'eco_i', 'private', 'ftp_data', 'telnet', 'pop_3', 'mtp', 'link', 'gopher', 'other', 'IRC', 'klogin', 'echo', 'time', 'remote_job', 'hostnames', 'uucp_path', 'nntp', 'http_443', 'efs', 'uucp', 'sql_net', 'daytime', 'rje', 'csnet_ns', 'sunrpc', 'bgp', 'vmnet', 'nnsp', 'whois', 'domain', 'printer', 'kshell', 'iso_tsap', 'name', 'supdup', 'pop_2', 'ldap', 'login', 'netbios_ns', 'imap4', 'Z39_50', 'discard', 'systat', 'exec', 'netstat', 'netbios_dgm', 'urh_i', 'urp_i', 'courier', 'ctf', 'shell', 'netbios_ssn', 'ssh', 'X11']

Some of the code referenced from hands-pyspark-big-data-analysis-video