<jupyter_start><jupyter_text>Basic Spark Configuration<jupyter_code># # PySpark Setup
# export PYSPARK_PYTHON=python3
# export PYSPARK_DRIVER_PYTHON="jupyter"
# export PYSPARK_DRIVER_PYTHON_OPTS="notebook"
# export PATH="$SPARK_HOME/bin:$PATH"

# pyspark 
type(sc)
type(spark)
# uniform entry point
# version
sc.version
# check the UI
# change configurations
# memory

# master, tab,<jupyter_output><empty_output><jupyter_text>Core Classes in Spark<jupyter_code>SparkContext # major entry to Spark
RDD # basic abstraction in Spark (low level)

pyspark.sql.SQLContext # main entry to DataFrame (depreciateing)
pyspark.sql.DataFrame # basic abstraction in DataFrame (depreciateing)
pyspark.sql.SparkSession # main entry to Data Frame 

#streaming.StreamingContext # major entry to streaming
#streaming.DStream # basic abstraction in streaming<jupyter_output><empty_output><jupyter_text>SparkContext<jupyter_code># create RDD directly
a = [i for i in range(10000)]
rdd_a = sc.parallelize(a)
type(a), type(rdd_a)
# storage: 
# list: in local memory (for sure)
# rdd_a: if the data will be materialized, in remote cluster memory 
a = [i for i in range(1000000)]
rdd_a = sc.parallelize(a) # spark a command, it has not be executed
%%time
rdd_a.count() # execution happen
%%time
rdd_a.cache()
rdd_a.count()
%%time
rdd_a.count()
# my_new_function in your local python file
# want to call my_new_function in spark 
sc.addPyFile('my_new_function.py')
sc.accumulator # only modifiable by Spark, not read
sc.broadcast # only readable by Spark, not modifiable
b = [1, 2, 3 ]
b_broad = sc.broadcast(b)

b_broad
# add extra files for dependency: addFile, addPyFile

# RDD creation method: parallelize, range

# Spreadout variables: accumulator, broadcast
# how to use accumulator & broadcast variables
# array (local) vs. RDD (remote)
# transform
a = [i for i in range(1000000)]
rdd_a = sc.parallelize(a) # spark a command, it has not be executed
rdd_a_1 = rdd_a.map(lambda x: x + 1)
# action
rdd_a_1.count()
# Tranform : do not run, until ... 
# Action :<jupyter_output><empty_output><jupyter_text>Resilient Distributed Dataset (RDD)<jupyter_code># creation: local memory, local file, external file

# Properties: partitions, persistence, getStorageLevel, partitions

# key-value pair RDD

# lazy evaluation (transformation & action), chain together 

# basic operations: filter, map, reduce, reduce by key, sample, 
#                   union, intersect, join, cartesian, coalesce
#                   reduce, collect, count, take
# Word Count example: textFile, flatMap, map, reducedByKey, saveAsTextFile

# load text
text_file = sc.textFile("job_description.txt")
text_file.take(4) 
t1 = text_file.flatMap(lambda x: x.split(' '))
t1.take(5)
t2 = t1.map(lambda x: (x, 1))
t2.reduceByKey?
t3 = t2.reduceByKey(lambda x, y: x + y)
t3.take(5)
t3.saveAsTextFile('all-test-txt')
sc.textFile("job_description.txt").flatMap(lambda x: x.split(' ')).map(lambda x: (x, 1)).reduceByKey(lambda x, y: x + y).saveAsTextFile('all-test-txt')
# process (separate out, try persist or not)
from pyspark import Accumulator
# Broadcast and Accumulator
a = sc.broadcast(range(10))<jupyter_output><empty_output><jupyter_text>Spark SQL/DataFrame  Spark Datasource API<jupyter_code>from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)
spark.createDataFrame?
# In a big company ... 
df = spark.sql('''
    select date, time, name
    from one_table
    where date = '2016-12-10'
''')
# backend: call Hive, use Spark kernel to run Hive-on-Spark, the table will be considered as df 
# Transform, NOT action 
# In lcoal environment ... (preferable)
spark.createDataFrame()
?spark.createDataFrame
import pandas as pd
df_pd = pd.read_csv('train.csv')
df_pd1 = df_pd[1:2]
df_pd1
# df_pd.fillna(value=0);
df_sp = spark.createDataFrame(df_pd1)
df_sp
type(df_pd), type(df_sp)
# If you don't like pandas for some reason ... 
df_sp1 = spark.read.csv('train.csv')
# parquet format is native for spark, it is faster than csv files
df_sp.write.parquet('save_1_txt')
df_sp.write.csv('save_2_txt')
# How to read data, how to write data? (character-deaths.csv)
# read/write (text, csv, parquet) 
# In pandas; series & data frame , have concept of both column & index
# In spark, only data frame , have concept of only column<jupyter_output><empty_output><jupyter_text>Spark Dataframe API  Type<jupyter_code># Explore all spark types
from pyspark.sql.types import StructField, StructType, ByteType, \
    DoubleType, DecimalType, StringType, ArrayType, IntegerType
# singlton vs initiated type
type(DoubleType())
StringType()
# construct arrayType
# ArrayType() would have error
ArrayType(StringType())<jupyter_output><empty_output><jupyter_text>Schema<jupyter_code>df_sp.schema
df_sp.printSchema()
df_sp1.printSchema()
df_tmp = spark.read.parquet('save_1_txt/')
df_tmp.printSchema()
# infer schema directly
# specify schema from StructType
my_schema = StructType([
     StructField('PassengerId', IntegerType(), True),
     StructField('Name', StringType(), True)
    ])
my_schema
df_sp.schema
data = [
    [1, 'David'],
    [2, 'Jason']
]
df_tmp = spark.createDataFrame(data, schema=my_schema)
df_tmp.printSchema()
# pay attention to the space after \ ......
pd.read_csv('train.csv') \
    .head(2)
df_tmp = spark.read.parquet('save_1_txt/')
%%time 
df_tmp.count()
%%time
df_tmp.cache()
df_tmp.count()
%%time
df_tmp.count()
# faster after cache
# compare different datatype, Byte vs. Integer<jupyter_output><empty_output><jupyter_text>Row<jupyter_code>from pyspark.sql import Row
a = df_sp.take(10)
a[0]
a[0]['PassengerId']
a[0][4]
b = a[0]
b
b.index(1)
b.count(1) # the number of values == 1
b.asDict()
Row(PassengerId=1, Embarked='S', NewRow=Row(C1=1, C2=3))
adict = {'Age': 22.0,
 'Embarked': 'S',
 'Fare': 7.25,
 'Name': 'Braund, Mr. <NAME>',
 'Parch': 0,
 'PassengerId': 1,
 'Pclass': 3,
 'Sex': 'male',
 'SibSp': 1,
 'Survived': 0,
 'Ticket': 'A/5 21171'}
Row(**adict)
# dict like object 
# nested Row<jupyter_output><empty_output><jupyter_text>Column<jupyter_code># df_sp['PassengerId'].take(2), take operation only valid for data frame, not for column
# understand column type, and what functions could be used over it

# sadly no directly displayd column by itself (no action)<jupyter_output><empty_output><jupyter_text>DataFrame<jupyter_code># Display operations
# dtypes, schema, take, head, tail, collect, columns, count
# freqItems, cross_table
type(df_pd), type(df_sp)
df_pd.shape
df_sp.count(), len(df_sp.columns)
df_pd.tail(2)
df_sp.head(2)
# df_sp.collect() 
# dangerous operation, not suggested, only if you know your data is small enough 
df_pd.dtypes
df_sp.dtypes
df_sp.printSchema()
df_pd.describe()
df_sp.describe().collect()
df_sp.describe().toPandas()
df_sp.freqItems(['Sex']).take(10)
df_pd.Sex.value_counts()
# SQL Key operations
# Union, intersection, difference, Cartesian product

# Selection, projection, join (inner, left, right, outer)
# SQL vs Dataframe API comparison<jupyter_output><empty_output><jupyter_text>Read data<jupyter_code># Pandas vs. Spark (train.csv)
# header, dtypes, N/A
# separator, Parquet
# import pandas as pd
# df_pd = pd.read_csv('train_mod.csv')
# df_sp = spark.createDataFrame(df_pd)<jupyter_output><empty_output><jupyter_text>filter<jupyter_code>df_pd.shape
df_pd.query(''' Sex == 'male' and Age > 60 ''').head(2)
df_sp.filter(''' Sex = 'male' ''').head(2)
df_sp.filter(''' Sex = 'male' and Age > 50 ''').head(2)
# data preprocessing: spark dataframe，pandas or pig？
# small, large; mostly due to legacy reason 
df_pd.query('''PassengerId < 5 and Age < 50 ''')
df_sp.filter('''Sex = 'female' ''').head(1)
# loc, query
# filter ('', df)<jupyter_output><empty_output><jupyter_text>selection<jupyter_code>df_pd[['PassengerId', 'Sex']].head(2)
df_sp.select('PassengerId', 'Sex').head(2)
# [[]], select
# select<jupyter_output><empty_output><jupyter_text>join<jupyter_code>df_pd_1 = df_pd.query(''' Sex == 'male' ''')
df_pd_2 = df_pd.query(''' Age >= 30 ''')
df_sp_1 = df_sp.filter(''' Sex = 'male' ''')
df_sp_2 = df_sp.filter(''' Age >= 30 ''')
?df_pd_1.merge
df_test_1 = df_pd_1.merge(df_pd_2, how='inner', on='PassengerId')
df_test_1.head(2)
df_test_1.shape
df_pd_1.shape
df_test_2 = df_sp_1.join(df_sp_2, how='inner', on='PassengerId')
df_test_2.count(), len(df_test_2.columns)
df_test_2.head(2)
# pay attention to column name 
df_sp_1.join(
    df_sp_2.select('PassengerId', 'Sex', 'Age') \
        .withColumnRenamed('Sex', 'Sex_new') \
        .withColumnRenamed('Age', 'Age_new'),
    how='inner', on='PassengerId')
# Age, Sex
# merge
# join<jupyter_output><empty_output><jupyter_text>column transformation<jupyter_code>df_pd.assign(NewAge=lambda x: x.Age + 10).head(2)
df_pd['NewAge'] = df_pd['Age'] + 10
import pyspark.sql.functions as F
df_sp.withColumn('NewAge', F.col('Age') + 10).head(2)
# %timeit
# df, assign
# withColumn
# udf, datatype<jupyter_output><empty_output><jupyter_text>groupby operation<jupyter_code># groupby, agg/apply
# udf, columns.droplevel()
# groupby, agg, alias, function
df_pd.groupby('Sex').agg({
        'PassengerId': 'count',
        'Age': 'mean'
    })
df_sp.groupby('Sex').agg({
        'PassengerId': 'count',
        'Age': 'mean'
    }).head(10)<jupyter_output><empty_output><jupyter_text>replace/missing data<jupyter_code>df_pd.fillna({
        'Sex': 'male',
        'Age': 0
    }).head(2)
df_sp.fillna({
        'Sex': 'male',
        'Age': 0
    }).head(2)
# replace (dict, None, regex)
# replace 
# value_count
# freqItems, crosstab<jupyter_output><empty_output><jupyter_text>Set operation (union, intersection, difference)<jupyter_code>df_pd_1 = df_pd.loc[lambda x: x.Survived == 1][['PassengerId', 'Name']]
df_pd_2 = df_pd.loc[lambda x: x.Sex == 'male'][['PassengerId', 'Name']]
print(df_pd_1.shape, df_pd_2.shape)
# concat, append, isin, ~ 
df_pd_1.append(df_pd_2).shape
df_pd_1.isin(df_pd_2).shape
df_sp_1 = df_sp.filter('Survived=1').select('PassengerId', 'Name')
df_sp_2 = df_sp.filter("Sex='male'").select('PassengerId', 'Name')
print(df_sp_1.count(), df_sp_2.count())
# union, intersect, subtract
df_sp_1.union(df_sp_2).count()<jupyter_output><empty_output><jupyter_text>SQL's approach<jupyter_code>df_sp.createOrReplaceTempView('df_table')
spark.sql('''
    select Age, count(*) as cnt
    from df_table
    where Sex = 'male'
    group by Age
    order by 2 desc 
''').take(10)
# select from where group by order by with<jupyter_output><empty_output><jupyter_text>window function<jupyter_code># IMPOTANT ESPECIALLY FOR INTERVIEW
spark.sql('''
    select PassengerId, Age, Pclass, 
        avg(Age) over (partition by Pclass) as age_in_sex_group, 
        (Age < avg(Age) over (partition by Pclass)) as smaller_than_group_avg
    from df_table
    order by 1 
''').take(10)
from pyspark.sql import Window
w1 = Window().partitionBy('Pclass')
df_sp.withColumn('age_in_sex_group', F.avg('Age').over(w1)).head(2)
# over partition by order by 
# groupby, apply
# Window partitionBy, orderBy<jupyter_output><empty_output><jupyter_text>GraphX & GraphFrame<jupyter_code># Spark GraphX Example
from graphframes import GraphFrame
v = spark.createDataFrame([
  ("a", "Alice", 34),
  ("b", "Bob", 36),
  ("c", "Charlie", 30),
  ("d", "David", 29),
  ("e", "Esther", 32),
  ("f", "Fanny", 36),
  ("g", "Gabby", 60)
], ["id", "name", "age"])
# Edge DataFrame
e = spark.createDataFrame([
  ("a", "b", "friend"),
  ("b", "c", "follow"),
  ("c", "b", "follow"),
  ("f", "c", "follow"),
  ("e", "f", "follow"),
  ("e", "d", "friend"),
  ("d", "a", "friend"),
  ("a", "e", "friend")
], ["src", "dst", "relationship"])
# Create a GraphFrame
g = GraphFrame(v, e)
# motif finding (a)-[e]->(c) ;
# subgraph
# connected component
# pageRank<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1525394793000427, 'fraction_numerical': 0.016474605206999574, 'mean_word_length': 5.462217319360176, 'pattern_counts': {'":': 0, '<': 89, '<?xml version=': 0, '>': 91, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '21', 'dir': 'jupyter-structured-clean-dedup', 'id': '480372', 'license': 'Apache-2.0', 'n_tokens_mistral': 4592, 'n_tokens_neox': 4384, 'n_words': 1042, 'path': 'Spark_Practice_20161130.ipynb', 'repo_name': 'dinrker/Spark_Practice', 'size': '11720'}