<jupyter_start><jupyter_text>If the notebook is run locally, then sc (SparkContext) would be pre-configured. If running using binder, we need to create SparkContext.<jupyter_code>#This notebook comes with a pre-configured sparkContext called sc
try:
    sc
except NameError:
    sc = SparkContext(master='spark://master:7077')
    with open("data/sequence.txt") as f:
         sequence = [x.strip('\n') for x in f.readlines()]
    file_rdd = sc.parallelize(sequence)
    with open("data/people.json") as f:
        json_data = [x.strip('\n') for x in f.readlines()]
    json_rdd = sc.parallelize(json_data)
else:
    file_rdd = sc.textFile("data/sequence.txt")
    json_rdd = sc.textFile("data/people.json")
sc
#RDDs
dummy_list = range(1000)
#this is a list
print type(dummy_list)
rddlist = sc.parallelize(dummy_list)
#this is a RDD
print type(rddlist)
#More RDDs
print json_rdd.collect()
print type(json_rdd.collect())
#Spark supports text files, SequenceFiles, and any other Hadoop InputFormat.
start_time = time()
filtered_rdd = file_rdd.filter(lambda x: len(x) < 2)
end_time = time()
print "Time taken (in seconds) = " + str(end_time - start_time)
start_time = time()
filtered_data = file_rdd.filter(lambda x: len(x) < 2).collect()
end_time = time()
print "Time taken (in seconds) = " + str(end_time - start_time)
start_time = time()
mapped_rdd = file_rdd.map(lambda x: 2*x)
end_time = time()
print "Time taken (in seconds) = " + str(end_time - start_time)
start_time = time()
mapped_data = file_rdd.map(lambda x: 2*x).collect()
end_time = time()
print "Time taken (in seconds) = " + str(end_time - start_time)
print type(filtered_rdd)
print type(filtered_data)
print type(mapped_rdd)
print type(mapped_data)
print "Conclusion: RDDs are lazy. They do nothing unless there is an action is called."
start_time = time()
print len(file_rdd.map(lambda x: 2*x).filter(lambda x: len(x)>1).take(10))
end_time = time()
print "Time taken (in seconds) = " + str(end_time - start_time)
start_time = time()
print len(file_rdd.map(lambda x: 2*x).filter(lambda x: len(x)>1).take(100000))
end_time = time()
print "Time taken (in seconds) = " + str(end_time - start_time)
start_time = time()
print len(file_rdd.map(lambda x: 2*x).filter(lambda x: len(x)>1).collect())
end_time = time()
print "Time taken (in seconds) = " + str(end_time - start_time)
print "Lazy is better."
start_time = time()
print "We want to count the number of 1, 2, ... digit numbers."
# file_path = "data/sequence.txt"
# file_rdd = sc.textFile(file_path) 
mapped_rdd = file_rdd.map(lambda a: (len(a), 1))
count_rdd = mapped_rdd.reduceByKey(lambda a, b: a+b).sortByKey()
print count_rdd.collect()
end_time = time()
print "Time taken (in seconds) = " + str(end_time - start_time)
start_time = time()
print "We want to count the number of 1, 2, ... digit numbers."
mapped_data = np.asarray(map(lambda a: (len(a), 1), file_rdd.collect()))
# count_map = mapped_rdd.reduceByKey(lambda a, b: a+b).sortByKey().collect()
# print count_map
end_time = time()
print "Time taken (in seconds) = " + str(end_time - start_time)
print "Lets add all the numbers. We have two ways of doing that."
print "Approach 1: update a counter variable"
counter = 0
def increment_counter(x):
    global counter
    counter+=x
mapped_rdd = file_rdd.map(lambda a: int(a))
mapped_rdd.foreach(increment_counter)
print "Sum using first approach: ", counter
print "Approach 2: use reduce operation"
print "Sum using second appraoch: ", mapped_rdd.reduce(operator.add)
print "Which one is correct?"
accum = sc.accumulator(0)
mapped_rdd.foreach(lambda x: accum.add(x))
print("Actual sum: ", accum.value)
print "So far we talked about big data. What about structured big data?"
sqlContext = SQLContext(sc)
sqlContext
#dataframes
df = sqlContext.read.json("data/people.json")
df.show()
print "A DataFrame is the structured version of RDD. This is the familiar relational view of the data."
df.printSchema()
print "DF can infer schema on its own. We can always override the inferred schema."
rdd_df = file_rdd.map(lambda a: Row(a)).toDF()
rdd_df.show()
rdd_df.printSchema()
# rdd_df = file_rdd.map(lambda a: Row(int(a))).toDF()
# rdd_df.show()
# rdd_df.printSchema()
df.show()
print "DF can be queried in two ways: API and sql queries"
print "First method: API"
df.select(df['first_name'], df['gender']).show()
df.groupBy(df["gender"]).count().show()
df.groupBy(df["email"]).agg(func.count(df["email"]).alias("count")).orderBy("count", ascending=False).show()
print "Second method: sql"
df.registerTempTable("people")
sqlContext.sql("SELECT * FROM people WHERE gender = 'Female'").show()
print "We can define our own functions as well."
domain = func.udf(lambda s: s.split('@')[1], StringType())
df.select(domain(df.email).alias('domain'))\
.groupBy('domain').agg(func.count('domain').alias("count")).orderBy("count", ascending=False).show()<jupyter_output>We can define our own functions as well.
+--------------------+-----+
|              domain|count|
+--------------------+-----+
|         alibaba.com|    8|
|             163.com|    7|
|        examiner.com|    7|
|      friendfeed.com|    6|
|            lulu.com|    6|
|             free.fr|    6|
|             fda.gov|    6|
|           apple.com|    6|
|       woothemes.com|    6|
|         cornell.edu|    6|
|     sourceforge.net|    6|
|             mlb.com|    6|
|           wikia.com|    5|
|        engadget.com|    5|
|pagesperso-orange.fr|    5|
|             usa.gov|    5|
|       wordpress.org|    5|
|        cbslocal.com|    5|
|            ucla.edu|    5|
|             pbs.org|    5|
+--------------------+-----+
only showing top 20 rows
{'fraction_non_alphanumeric': 0.15611140677665425, 'fraction_numerical': 0.010998758204718823, 'mean_word_length': 3.710108604845447, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 7, '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': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '183542', 'license': 'MIT', 'n_tokens_mistral': 2015, 'n_tokens_neox': 1962, 'n_words': 545, 'path': 'spark/PyDelhi/PySpark.ipynb', 'repo_name': 'shagunsodhani/talks', 'size': '5637'}