<jupyter_start><jupyter_text>Lecture Notebook: Big Data and Graph DataApache Spark is a big data engine that runs on compute clusters, including on the cloud.  This notebook is set up assuming that (1) Spark is running on a cloud server that is public and (2) we need to run the actual Python commands on that server, requiring us to put `%%spark` "magic" commands at the start of each cell.You may wish to look at this notebook first, without directly running it, until you understand how to launch your own Spark cluster.<jupyter_code>!apt install libkrb5-dev
!pip install sparkmagic
%load_ext sparkmagic.magics<jupyter_output><empty_output><jupyter_text>The following line connects to Spark running remotely.  You will need to change the URL after the `-u` to connect to an active server. Check the instruction on how to set up EMR cluster on AWS if you are not aware of how to do it.<jupyter_code>%spark add -s my_session -l python -u http://ec2-18-213-115-72.compute-1.amazonaws.com:8998 
# The above can connect to an EMR node running Spark + Livy, assuming the firewall is set to let anyone in
%spark delete -s my_session_2<jupyter_output>An error was encountered:
Could not find 'my_session_2' session in list of saved sessions. Possible sessions are ['my_session']<jupyter_text>Example of Loading Sharded Data (Lecture Slides A) Some preliminaries:  **Every** cell in this notebook will need `%%spark` at the start so it runs on the remote machine with Spark instead of on the machine with Jupyter.<jupyter_code>%%spark
import json
# import requests
import urllib.request as urllib
%%spark
import sys
print(sys.version)
%%spark
# 10K records from linkedin
# linked_in = requests.get('X')
# If you have problems with the above, use urllib 
linked_in = urllib.urlopen('https://XXX/test_data_10000.json')

# my_list = [json.loads(line) for line in linked_in.iter_lines()]
# If you have problems with the the above, use urllib 
my_list = [json.loads(line) for line in linked_in.readlines()]
len(my_list)<jupyter_output><empty_output><jupyter_text>Load the list into SparkSpark needs to know the structure of the data in its dataframes, i.e., their schemas.  Since our JSON structure for LinkedIn is complex, we need to define the schema.There are some basic types:  * The table is a `StructType` with a list of fields (each row)  * Most fields, in our case, are `StringType`.  * We also have nested dictionary for the name, which is a `MapType` from `StringType` keys to `StringType` values.  * `skills` is an `ArrayType` since it's a list, and it contains `StringType`s.  * `also_view` is an array of structs.See Pyspark documentation on `StructType` and examples such as https://www.programcreek.com/python/example/104715/pyspark.sql.types.StructType.<jupyter_code>%%spark

# Spark requires that we define a schema for the LinkedIn data...
from pyspark.sql.types import StringType, StructField, StructType, ArrayType, MapType
schema = StructType([
        StructField("_id", StringType(), True),
        StructField("name", MapType(StringType(), StringType()), True),
        StructField("locality", StringType(), True),
        StructField("skills", ArrayType(StringType()), True),
        StructField("industry", StringType(), True),
        StructField("summary", StringType(), True),
        StructField("url", StringType(), True),
        StructField("also_view", ArrayType(\
                    StructType([\
                      StructField("url", StringType(), True),\
                      StructField("id", StringType(), True)])\
                    ), True)\
         ])
%%spark
# Load the remote data as a list of dictionaries
linked_df = sqlContext.createDataFrame(my_list, schema).\
  repartition('_id')

linked_df.show(5)
%%spark
linked_df.rdd.getNumPartitions()
%%spark
linked_df.filter(linked_df.locality == 'United States')[['_id', 'name', 'locality']].show(5)
%%spark
linked_df.select("_id", "locality").show(5)
%%spark
### Clean out the list from memory
my_list = []
%%spark
linked_df.createOrReplaceTempView('linked_in')
sqlContext.sql('select * from linked_in').show(5)
%%spark
sqlContext.sql('select _id, name.given_name, name.family_name from linked_in').show(5)
%%spark
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

acro = udf(lambda x: ''.join([n[0] for n in x.split()]), StringType())

linked_df.select("_id", acro("locality").alias("acronym")).show(5)
%%spark
# Which industries are most popular?
sqlContext.sql('select count(_id), industry '+\
               'from linked_in '+\
               'group by industry '+\
               'order by count(_id) desc').\
    show(5)<jupyter_output><empty_output><jupyter_text>Graphs For the next set of examples, we will look at graph-structured data.  It turns out our LinkedIn dataset has a list of nodes (by int ID, but associated with the user ID we used in the linked_in table) and a list of edges.<jupyter_code>%%spark

import urllib.request as urllib
import zipfile
import os

#url = 'E'
url = 'https://XXX/linkedin.edges.zip'
filehandle, _ = urllib.urlretrieve(url)

zip_file_object = zipfile.ZipFile(filehandle, 'r')
fname = zip_file_object.open('linkedin.edges')

edges = []
MAX = 100000

for link in fname:
  edge = link.decode('utf-8').split(' ')
  edges.append([int(edge[0]), int(edge[1])])
  if len(edges) >= MAX:
    break

%%spark

from pyspark.sql.types import IntegerType
schema = StructType([
        StructField("from", IntegerType(), True),
        StructField("to", IntegerType(), True)
         ])
# Load the remote data as a list of dictionaries
edges_df = sqlContext.createDataFrame(edges, schema)

edges_df.show(5)
%%spark
edges_df.createOrReplaceTempView('edges')
sqlContext.sql('select from as id, count(to) as degree from edges group by from').show(5)<jupyter_output><empty_output><jupyter_text>Traversing the Graph<jupyter_code>%%spark

from pyspark.sql.functions import col

# Start with a subset of nodes
start_nodes_df = edges_df[['from']].filter(edges_df['from'] < 1000).\
  select(col('from').alias('id')).drop_duplicates()

start_nodes_df.show(5)

# The neighbors require us to join
# and we'll use Spark DataFrames syntax here
neighbor_nodes_df = start_nodes_df.\
  join(edges_df, start_nodes_df.id == edges_df['from']).\
  select(col('to').alias('id'))

neighbor_nodes_df.show(5)
%%spark
edges_df[['from']].orderBy('from').drop_duplicates().show()

edges_df.filter(edges_df['from'] == 1).show()
%%spark
neighbor_neighbor_nodes_df = neighbor_nodes_df.\
  join(edges_df, neighbor_nodes_df.id == edges_df['from']).\
  select(col('to').alias('id'))

neighbor_neighbor_nodes_df.show(5)
%%spark
def iterate(df, depth):
  df.createOrReplaceTempView('iter')

  # Base case: direct connection
  result = sqlContext.sql('select from, to, 1 as depth from iter')

  for i in range(1, depth):
    result.createOrReplaceTempView('result')
    result = sqlContext.sql('select r1.from as from, r2.to as to, r1.depth+1 as depth  '\
                            'from result r1 join iter r2 '\
                            'on r1.to=r2.from')
  return result
%%spark
iterate(edges_df.filter(edges_df['from'] < 1000), 1).orderBy('from','to').show()
%%spark
iterate(edges_df.filter(edges_df['from'] < 1000), 2).orderBy('from','to').show()
%%spark
iterate(edges_df.filter(edges_df['from'] < 1000), 3).orderBy('from','to').show()
%%spark
# Clear list of edges from Python memory
# to free up space
edges = []


### Now let's get the list of node IDs
#url = 'N'
url = 'https://XXX/linkedin.nodes.new.zip'
nodehandle, _ = urllib.urlretrieve(url)

zip_file_object = zipfile.ZipFile(nodehandle, 'r')
fname = zip_file_object.open('linkedin.nodes.new')

nodes = []
MAX = 100000

for node in fname:
  node_tuple = node.split()
  nodes.append([int(node_tuple[0]), str(node_tuple[1])])
  if len(nodes) >= MAX:
    break<jupyter_output><empty_output><jupyter_text>Joins in Spark, Beyond Graph TraversalsWhat if we want to connect our edges to the people from our previous crawl?  Sadly the edges use int node IDs that don't correspond to the people dataframe.  But in fact the node data includes this information, so let's load and exploit that. Let's load the information about nodes, and their correspondence to the user ID.<jupyter_code>%%spark
schema = StructType([
        StructField("nid", IntegerType(), True),
        StructField("user", StringType(), True)
         ])
# Load the remote data as a list of dictionaries
nodes_df = sqlContext.createDataFrame(nodes, schema)

nodes_df.show(5)<jupyter_output><empty_output><jupyter_text>Finding Friends, by ID<jupyter_code>%%spark

nodes_df.createOrReplaceTempView('nodes')
edges_df.createOrReplaceTempView('edges')

friends_df = \
sqlContext.sql('select n1.user, n2.user as friend ' +\
               'from (nodes n1 join edges e on n1.nid = e.from) join nodes n2 on e.to = n2.nid')

friends_df.show(5)<jupyter_output><empty_output><jupyter_text>Connecting Friends to Names<jupyter_code>%%spark
friends_df.createOrReplaceTempView('friends')

sqlContext.sql('select u1.name.given_name as user, u2.name.given_name as friend '+\
               'from (linked_in u1 join friends on u1._id = user) join linked_in u2 on u2._id = friend').show(5)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12099567099567099, 'fraction_numerical': 0.011796536796536796, 'mean_word_length': 4.546818727490996, 'pattern_counts': {'":': 0, '<': 40, '<?xml version=': 0, '>': 38, 'https://': 4, 'lorem ipsum': 0, 'www.': 1, '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': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '167056', 'license': 'Apache-2.0', 'n_tokens_mistral': 3050, 'n_tokens_neox': 2903, 'n_words': 919, 'path': 'opends4all-resources/opends4all-scalable-data-processing/CLUSTER-DATA-PROCESSING-parallelism-partitioning-advanced-Cloud.ipynb', 'repo_name': 'ophdesdi/OpenDS4All', 'size': '9240'}