Tematyka zajęć:
Pliki do realizacji zadań ( dostępne na serwerze w katalogu /home/spark/files ):
import findspark findspark.init()
from pyspark import SparkConf
from pyspark import SparkContext
sc = SparkContext.getOrCreate(SparkConf().setMaster("local[4]"))
import random flips = 1000000 coins = range(flips) rdd_coins = sc.parallelize(coins) rdd_flips = rdd_coins.map(lambda i: random.random()) rdd_heads = rdd_flips.filter(lambda r: r < 0.51 ) rdd_heads.count()
import random
num_samples = 1000000
def inside(p):
x,y = random.random(), random.random()
return x*x + y*y < 1
count = sc.parallelize(range(0, num_samples)).filter(inside).count()
pi = 4* count/num_samples
print(pi)
import org.apache.spark.sql._
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.{IntegerType, StringType, ArrayType}
val arrayStructureData = Seq(
Row(Row("James","","Smith"),List("Java","Scala","C++"),"OH","M"),
Row(Row("Anna","Rose",""),List("Spark","Java","C++"),"NY","F"),
Row(Row("Julia","","Williams"),List("CSharp","VB"),"OH","F"),
Row(Row("Maria","Anne","Jones"),List("CSharp","VB"),"NY","M"),
Row(Row("Jen","Mary","Brown"),List("CSharp","VB"),"NY","M"),
Row(Row("Mike","Mary","Williams"),List("Python","VB"),"OH","M")
)
val arrayStructureSchema = new StructType().
add("name",new StructType().
add("firstname",StringType).
add("middlename",StringType).
add("lastname",StringType)).
add("languages", ArrayType(StringType)).
add("state", StringType).
add("gender", StringType)
val df = spark.createDataFrame(spark.sparkContext.parallelize(arrayStructureData),arrayStructureSchema) df.printSchema() df.show()
df.filter(df("state") === "OH").show(false)
df.filter('state === "OH").show(false)
df.filter($"state" === "OH").show(false)
df.filter(df.col("state") === "OH").show(false)
df.where(df("state") === "OH").show(false)
df.where('state === "OH").show(false)
df.where($"state" === "OH").show(false)
df.where(df.col("state") === "OH").show(false)
df.filter("gender == 'M'").show(false)
df.where("gender == 'M'").show(false)
df.filter(df("state") === "OH" && df("gender") === "M").show(false)
df.where(df("state") === "OH" && df("gender") === "M").show(false)
import org.apache.spark.sql.functions.array_contains
df.filter(array_contains(df("languages"),"Java")).show(false)
df.filter(df("name.lastname") === "Williams").show(false)
df.filter(df("name.lastname") === "Williams").explain()
df.filter(array_contains(df("languages"),"Java")).explain()
df.filter(df("state") === "OH" && df("gender") === "M").explain()
import org.apache.spark.sql._
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.{IntegerType, StringType, ArrayType}
import org.apache.spark.sql.SparkSession
val session = SparkSession.builder.config("spark.master","local[1]").getOrCreate
val schema = new StructType().
add("fname",StringType).
add("lname",StringType).
add("courses",StringType).
add("grade",IntegerType).
add("year",IntegerType)
val data = session.read.option("header","true").option("delimiter","|").schema(schema).csv("/home/spark/files/dane1.csv")
data.printSchema()
data.show()
data.where("fname = 'Weronika'").show(false)
data.where('fname === "Weronika").show(false)
data.where(data.col("fname") === "Weronika").show(false)
data.select($"lname").show(false)
data.select($"lname",$"fname").show(false)
data.groupBy($"grade",$"year").count().show(false)
data.groupBy($"grade",$"year").count().orderBy($"grade",$"year").show(false)
data.createGlobalTempView("lista")
spark.sql("SELECT * from global_temp.lista").show()
spark.sql("SELECT grade, year, count(*) from global_temp.lista group by grade, year order by grade, year").show()
from pyspark.sql.dataframe import DataFrame from pyspark.sql import SparkSession spark = SparkSession(sc)
df = spark.read.format('com.databricks.spark.csv').\
options(header='true', \
inferschema='true').load("/home/spark/files/gcredit.csv",header=True);
df.columns
df[['Account Balance','No of dependents']].toPandas().describe()
def describe_pd(df_in, columns, style):
'''
Function to union the basic stats results and deciles
:param df_in: the input dataframe
:param columns: the cloumn name list of the numerical variable
:param style: the display style
:return : the numerical describe info. of the input dataframe
:author: Wenqiang Feng
:email: von198@gmail.com
'''
if style == 1:
percentiles = [25, 50, 75]
else:
percentiles = np.array(range(0, 110, 10))
percs = np.transpose([np.percentile(df_in.select(x).collect(), percentiles) for x in columns])
percs = pd.DataFrame(percs, columns=columns)
percs['summary'] = [str(p) + '%' for p in percentiles]
spark_describe = df_in.describe().toPandas()
new_df = pd.concat([spark_describe, percs],ignore_index=True,sort=True)
new_df = new_df.round(2)
return new_df[['summary'] + columns]
num_cols = ['Account Balance','No of dependents']
df.select(num_cols).describe().show()
import numpy as np import pandas as pd
output = describe_pd(df,num_cols,1)
output['summary']= output['summary'].astype(str) # convert just columns output[num_cols] = output[num_cols].apply(pd.to_numeric)
output.dtypes
spark.createDataFrame(output).show()
output = describe_pd(df,num_cols,2)
output['summary']= output['summary'].astype(str) # convert just columns output[num_cols] = output[num_cols].apply(pd.to_numeric) spark.createDataFrame(output).show()
var = 'Age (years)' # pyspark.sql.function #df.select(skewness(var),kurtosis(var)).show() # pandas skew(), kurtosis() df[['Age (years)']].toPandas().skew(),df[['Age (years)']].toPandas().kurtosis()
df.select('Credit Amount').show(5)
data1 = df.select('Age (years)').toPandas()
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from scipy import stats
%matplotlib inline
plt.rcParams['figure.figsize'] =(16,9)
plt.style.use('ggplot')
sns.set()
var = 'Age (years)'
x = data1[var]
bins = np.arange(0, 100, 5.0)
plt.figure(figsize=(10,8))
# the histogram of the data
plt.hist(x, bins, alpha=0.8, histtype='bar', color='gold',
ec='black',weights=np.zeros_like(x) + 100. / x.size)
plt.xlabel(var)
plt.ylabel('percentage')
plt.xticks(bins)
plt.show()
#fig.savefig(var+".pdf", bbox_inches='tight')
var = 'Age (years)'
x = data1[var]
bins = np.arange(0, 100, 5.0)
########################################################################
hist, bin_edges = np.histogram(x,bins,
weights=np.zeros_like(x) + 100. / x.size)
# make the histogram
fig = plt.figure(figsize=(20, 8))
ax = fig.add_subplot(1, 2, 1)
# Plot the histogram heights against integers on the x axis
ax.bar(range(len(hist)),hist,width=1,alpha=0.8,ec ='black', color='gold')
# # Set the ticks to the middle of the bars
ax.set_xticks([0.5+i for i,j in enumerate(hist)])
# Set the xticklabels to a string that tells us what the bin edges were
labels =['{}'.format(int(bins[i+1])) for i,j in enumerate(hist)]
#labels.insert(0,'0')
ax.set_xticklabels(labels)
plt.xlabel(var)
plt.ylabel('percentage')
########################################################################
hist, bin_edges = np.histogram(x,bins) # make the histogram
ax = fig.add_subplot(1, 2, 2)
# Plot the histogram heights against integers on the x axis
ax.bar(range(len(hist)),hist,width=1,alpha=0.8,ec ='black', color='gold')
# # Set the ticks to the middle of the bars
ax.set_xticks([0.5+i for i,j in enumerate(hist)])
# Set the xticklabels to a string that tells us what the bin edges were
labels =['{}'.format(int(bins[i+1])) for i,j in enumerate(hist)]
#labels.insert(0,'0')
ax.set_xticklabels(labels)
plt.xlabel(var)
plt.ylabel('count')
plt.suptitle('Histogram of {}: Left with percentage output;Right with count output'
.format(var), size=16)
plt.show()
#fig.savefig(var+".pdf", bbox_inches='tight')
def age_condition(x):
if pd.isnull(x):
return "missing"
elif x < 25:
return "<25"
elif 25 <= x <= 34:
return "25-34"
elif 35 <= x <= 44:
return "35-44"
elif 45 <= x <= 54:
return "45-54"
elif 55 <= x <= 64:
return "55-64"
else:
return "65+"
from pyspark.sql.functions import udf from pyspark.sql.types import StringType, DoubleType age_udf = udf(lambda x: age_condition(x), StringType())
df = df.withColumn("age_class", age_udf("Age (years)"))
df.select(['age_class','Age (years)']).show(3)
df.select(['age_class','Credit Amount']).\
groupBy('age_class').count().show()
df.stat.crosstab("age_class", "Occupation").show()
from pyspark.sql import functions as F
from pyspark.sql.functions import rank,sum,col
from pyspark.sql import Window
window = Window.rowsBetween(Window.unboundedPreceding,Window.unboundedFollowing)
# withColumn('Percent %',F.format_string("%5.0f%%\n",col('Credit_num')*100/col('total'))).\
tab = df.select(['age_class','Credit Amount']).\
groupBy('age_class').\
agg(F.count('Credit Amount').alias('Credit_num'),
F.mean('Credit Amount').alias('Credit_avg'),
F.min('Credit Amount').alias('Credit_min'),
F.max('Credit Amount').alias('Credit_max')).\
withColumn('total',sum(col('Credit_num')).over(window)).\
withColumn('Percent',col('Credit_num')*100/col('total')).\
drop(col('total'))
tab.show()
plot_data = tab.toPandas()
plot_data.sort_values('age_class')
custom_dict = {'<25': 0, '25-34': 1, '35-44': 2, '45-54': 3, '55-64': 4, '65+': 5}
plot_data['index']= plot_data['age_class'].replace(custom_dict)
plot_data.sort_values('index')
plot_data = plot_data.sort_values('index')
# Data to plot
labels = plot_data.age_class
sizes = plot_data.Percent
colors = ['gold', 'yellowgreen', 'lightcoral','blue', 'lightskyblue','green','red']
explode = (0, 0.1, 0, 0,0,0) # explode 1st slice
# Plot
plt.figure(figsize=(10,8))
plt.pie(sizes, explode=explode, labels=labels, colors=colors,
autopct='%1.1f%%', shadow=True, startangle=140)
plt.axis('equal')
plt.show()
labels = plot_data.age_class
missing = plot_data.Percent
ind = [x for x, _ in enumerate(labels)]
plt.figure(figsize=(10,8))
plt.bar(ind, missing, width=0.8, label='missing', color='gold')
plt.xticks(ind, labels)
plt.ylabel("percentage")
plt.show()
labels = ['missing', '<25', '25-34', '35-44', '45-54','55-64','65+']
missing = np.array([0.000095, 0.024830, 0.028665, 0.029477, 0.031918,0.037073,0.026699])
man = np.array([0.000147, 0.036311, 0.038684, 0.044761, 0.051269, 0.059542, 0.054259])
women = np.array([0.004035, 0.032935, 0.035351, 0.041778, 0.048437, 0.056236,0.048091])
ind = [x for x, _ in enumerate(labels)]
plt.figure(figsize=(10,8))
plt.bar(ind, women, width=0.8, label='women', color='gold', bottom=man+missing)
plt.bar(ind, man, width=0.8, label='man', color='silver', bottom=missing)
plt.bar(ind, missing, width=0.8, label='missing', color='#CD853F')
plt.xticks(ind, labels)
plt.ylabel("percentage")
plt.legend(loc="upper left")
plt.title("demo")
plt.show()
# prepare for the plot data
var = 'Credit Amount'
plot_data = df.select(var).toPandas()
x= plot_data[var]
bins =[0,200,400,600,700,800,900,1000,2000,3000,4000,5000,6000,10000,25000]
hist, bin_edges = np.histogram(x,bins,weights=np.zeros_like(x) + 100. / x.size) # make the histogram
fig = plt.figure(figsize=(10, 8))
ax = fig.add_subplot(1, 1, 1)
# Plot the histogram heights against integers on the x axis
ax.bar(range(len(hist)),hist,width=1,alpha=0.8,ec ='black',color = 'gold')
# # Set the ticks to the middle of the bars
ax.set_xticks([0.5+i for i,j in enumerate(hist)])
# Set the xticklabels to a string that tells us what the bin edges were
#labels =['{}k'.format(int(bins[i+1]/1000)) for i,j in enumerate(hist)]
labels =['{}'.format(bins[i+1]) for i,j in enumerate(hist)]
#labels.insert(0,'0')
ax.set_xticklabels(labels)
#plt.text(-0.6, -1.4,'0')
plt.xlabel(var)
plt.ylabel('percentage')
plt.show()
import findspark
findspark.init()
# configure spark variables
from pyspark.context import SparkContext
from pyspark import SparkConf
from pyspark.sql.context import SQLContext
from pyspark.sql.session import SparkSession
sc = SparkContext().getOrCreate(SparkConf().setMaster("local[4]"))
sqlContext = SQLContext(sc)
spark = SparkSession(sc)
# load up other dependencies
import re
import pandas as pd
m = re.finditer(r'.*?(spark).*?', "I'm searching for a spark in PySpark", re.I)
for match in m:
print(match, match.start(), match.end())
raw_data_files = ['/home/spark/files/access_log1', '/home/spark/files/access_log2'] base_df = spark.read.text(raw_data_files) base_df.printSchema()
type(base_df)
base_df_rdd = base_df.rdd type(base_df_rdd)
base_df.show(10, truncate=False)
base_df_rdd.take(10)
print((base_df.count(), len(base_df.columns)))
sample_logs = [item['value'] for item in base_df.take(15)] sample_logs
host_pattern = r'(^\S+)\s'
hosts = [re.search(host_pattern, item).group(1)
if re.search(host_pattern, item)
else 'no match'
for item in sample_logs]
hosts
ts_pattern = r'\[(\d{2}/\w{3}/\d{4}:\d{2}:\d{2}:\d{2} \S{1}\d{4})]'
timestamps = [re.search(ts_pattern, item).group(1)
if re.search(ts_pattern, item)
else 'no match'
for item in sample_logs]
timestamps
method_uri_protocol_pattern = r'\"(\S+)\s(\S+)\s*(\S*)\"'
method_uri_protocol = [re.search(method_uri_protocol_pattern, item).groups()
if re.search(method_uri_protocol_pattern, item)
else 'no match'
for item in sample_logs]
method_uri_protocol
from pyspark.sql.functions import regexp_extract
logs_df = base_df.select(regexp_extract('value', r'(^[\S+]+) -', 1).alias('host'),
regexp_extract('value', r'(^[\S+]+) - - \[(.*)\]', 2).alias('timestamp'),
regexp_extract('value', r'(^[\S+]+) - - \[(.*)\] "(\w+)', 3).alias('method'),
regexp_extract('value', r'(^[\S+]+) - - \[(.*)\] "(\w+) (.*?) (.*?)', 4).alias('path'),
regexp_extract('value', r'(^[\S+]+) - - \[(.*) \+(.*)\] "(\w+) (.*?) (.*?)" ', 6).alias('protocol'),
regexp_extract('value', r'(^[\S+]+) - - \[(.*) \+(.*)\] "(\w+) (.*?) (.*?)" (\d+) ', 7).cast('integer').alias('code'),
regexp_extract('value', r'(^[\S+]+) - - \[(.*) \+(.*)\] "(\w+) (.*?) (.*?)" (\d+) (\d+)', 8).cast('integer').alias('size'))
logs_df.show(10, truncate=True)
print((logs_df.count(), len(logs_df.columns)))
(base_df
.filter(base_df['value']
.isNull())
.count())
bad_rows_df = logs_df.filter(logs_df['host'].isNull()|
# logs_df['timestamp'].isNull() |
# logs_df['method'].isNull() |
# logs_df['path'].isNull() |
logs_df['code'].isNull() |
logs_df['size'].isNull()|
logs_df['protocol'].isNull())
bad_rows_df.count()
logs_df = logs_df.na.fill({'size': 0})
bad_rows_df.count()
from pyspark.sql.functions import udf
month_map = {
'Jan': 1, 'Feb': 2, 'Mar':3, 'Apr':4, 'May':5, 'Jun':6, 'Jul':7,
'Aug':8, 'Sep': 9, 'Oct':10, 'Nov': 11, 'Dec': 12
}
def parse_clf_time(text):
""" Convert Common Log time format into a Python datetime object
Args:
text (str): date and time in Apache time format [dd/mmm/yyyy:hh:mm:ss (+/-)zzzz]
Returns:
a string suitable for passing to CAST('timestamp')
"""
# NOTE: We're ignoring the time zones here, might need to be handled depending on the problem you are solving
return "{0:04d}-{1:02d}-{2:02d} {3:02d}:{4:02d}:{5:02d}".format(
int(text[7:11]),
month_map[text[3:6]],
int(text[0:2]),
int(text[12:14]),
int(text[15:17]),
int(text[18:20])
)
udf_parse_time = udf(parse_clf_time)
logs_df = (logs_df.select('*', udf_parse_time(logs_df['timestamp'])
.cast('timestamp')
.alias('time'))
.drop('timestamp'))
logs_df.printSchema()
import numpy as np import pandas as pd
content_size_summary_df = logs_df.describe(['size'])
content_size_summary_df.toPandas()
from pyspark.sql import functions as F
(logs_df.agg(F.min(logs_df['size']).alias('min_content_size'),
F.max(logs_df['size']).alias('max_content_size'),
F.mean(logs_df['size']).alias('mean_content_size'),
F.stddev(logs_df['size']).alias('std_content_size'),
F.count(logs_df['size']).alias('count_content_size'))
.toPandas())
status_freq_df = (logs_df
.groupBy('code')
.count()
.sort('code')
.cache())
print('Total distinct HTTP Status Codes:', status_freq_df.count())
status_freq_pd_df = (status_freq_df
.toPandas()
.sort_values(by=['count'],
ascending=False))
status_freq_pd_df
import matplotlib.pyplot as plt
import seaborn as sns
import numpy as np
%matplotlib inline
sns.catplot(x='code', y='count', data=status_freq_pd_df,
kind='bar', order=status_freq_pd_df['code'])
log_freq_df = status_freq_df.withColumn('log(count)',
F.log(status_freq_df['count']))
log_freq_df.show()
log_freq_pd_df = (log_freq_df
.toPandas()
.sort_values(by=['log(count)'],
ascending=False))
sns.catplot(x='code', y='log(count)', data=log_freq_pd_df,
kind='bar', order=status_freq_pd_df['code'])
W ramach zadania należy opracować punkt drugi (filtrowanie danych) z wykorzystaniem środowiska PySpark (obydwa przykłady).