<gh_stars>0
# -*- coding: utf-8 -*-
"""Friend Recommendation.ipynb

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/1sqykgz_IkAOnroLacly1KMVl8ahTwb8E
"""
## Friend Recommendation

# Set up

!pip install pyspark
!pip install -U -q PyDrive
!apt-get update 
!apt-get install openjdk-8-jdk-headless -qq
import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"

"""Now we authenticate a Google Drive client to download the file we will be processing in our Spark job.

"""

from pydrive.auth import GoogleAuth
from pydrive.drive import GoogleDrive
from google.colab import auth
from oauth2client.client import GoogleCredentials

# Authenticate and create the PyDrive client
auth.authenticate_user()
gauth = GoogleAuth()
gauth.credentials = GoogleCredentials.get_application_default()
drive = GoogleDrive(gauth)

id='google-drive-file-id'
downloaded = drive.CreateFile({'id': id})
downloaded.GetContentFile('test-data.txt')


# Commented out IPython magic to ensure Python compatibility.
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
# %matplotlib inline

import pyspark
from pyspark.sql import *
from pyspark.sql.functions import *
from pyspark import SparkContext, SparkConf

"""initialize the Spark context."""

# create the session
conf = SparkConf().set("spark.ui.port", "4050")

# create the context
sc = pyspark.SparkContext(conf=conf)
spark = SparkSession.builder.getOrCreate()

"""You can easily check the current version and get the link of the web interface. In the Spark UI, you can monitor the progress of your job and debug the performance bottlenecks (if your Colab is running with a **local runtime**)."""

spark

"""If you are running this Colab on the Google hosted runtime, the cell below will create a *ngrok* tunnel which will allow you to still check the Spark UI."""

!wget https://bin.equinox.io/c/4VmDzA7iaHb/ngrok-stable-linux-amd64.zip
!unzip ngrok-stable-linux-amd64.zip

get_ipython().system_raw('./ngrok http 4050 &')

!curl -s http://localhost:4040/api/tunnels | python3 -c \
    "import sys, json; print(json.load(sys.stdin)['tunnels'][0]['public_url'])"

!touch test.txt

# Commented out IPython magic to ensure Python compatibility.
# %pycat test.txt

#txt = sc.textFile("test.txt")
#sep = " "
txt = sc.textFile("test-data.txt")
sep = "\t"

def lineToMapOfListOfFriends(line: str) -> ():
  splittedLine = line.split(sep)
  if len(splittedLine) < 2: return (int(splittedLine[0]), ('L', ''))
  elif (not splittedLine[1].strip()): return (int(splittedLine[0]), ('L', ''))
  else:
    friends = [int(id) for id in splittedLine[1].split(",")]
    friends.sort()
    return (int(splittedLine[0]), ('L', friends))

def lineToLinks(line: str) -> []:
  splittedLine = line.split(sep)
  if len(splittedLine) < 2: return []
  elif (not splittedLine[1].strip()): return []
  friends = splittedLine[1].split(",")
  return([(int(splittedLine[0]), int(f)) for f in friends])

# (id0, ('L', [id1,id2,...idn]))
firstDegreeListOfFriendsRDD = txt.filter(lambda line: line).map(lambda line: lineToMapOfListOfFriends(line))
# (id0, idx)
firstDegreeRDD = txt.filter(lambda line: line.strip()).flatMap(lambda line: lineToLinks(line))
#firstDegreeListOfFriendsRDD.lookup(11)
firstDegreeRDD.lookup(11)

# (id0  ('2D', idx))
secondDegreeRDD = firstDegreeRDD.join(firstDegreeRDD) \
    .filter(lambda pair: pair[1][0] != pair[1][1]) \
    .map(lambda pair: (pair[1][0], ('2D', pair[1][1]))) \
    .distinct()

secondDegreeRDD.lookup(11)

# (id0, ('L', [id1,id2,...,idn]))
# (id0  ('2D', 'idx'))
tempMergedRDD = firstDegreeListOfFriendsRDD.union(secondDegreeRDD)

def merge2DFriendNListOfFriends(pair):
  id = pair[0]
  seq = pair[1].__iter__()
  friendsList = ''
  secondDegreeFriends = []
  for (flag, v) in seq:
    if flag == 'L':
      friendsList = v
    elif flag == '2D':
      secondDegreeFriends.append(v)
  return [((id, sdf), (id, friendsList)) for sdf in secondDegreeFriends] + [((sdf, id), (id, friendsList)) for sdf in secondDegreeFriends]
# [((id1, id2), (id1, [idx,idy,idz...])),
#  ((id1, id2), (id2, [idm,idn,ido...])), ...]
secondDegreeWithFriendsRDD = tempMergedRDD.groupByKey().flatMap(lambda pair: merge2DFriendNListOfFriends(pair))

tempMergedRDD.lookup(11)

secondDegreeWithFriendsRDD.lookup((11, 27573))

def countCommonFriends(pair):
  seq = pair[1].__iter__()
  data1 = seq.__next__()
  print(data1)
  data2 = seq.__next__()
  id1 = data1[0]
  list1 = data1[1]
  id2 = data2[0]
  list2 = data2[1]

  # return 0 if they are already friends
  if(id1 in list2 or id2 in list1): return (pair[0], 0)

  # make sure the friends list is sorted
  def isSorted(l: list):
    if(len(l) <= 1): return True
    for i in range(1, len(l)):
       if(l[i]<l[i-1]): return False
    return True
  
  if(not isSorted(list1)): 
    raise Exception(id1, data1[1])
  if(not isSorted(list2)):
    raise Exception(id2, data2[1])
  
  # only when the friends list is sorted, can use below algorithm to get common friends count
  i,j,cnt = 0,0,0
  while i<len(list1) and j < len(list2):
    if(list1[i] < list2[j]): i += 1
    elif(list1[i] > list2[j]): j += 1
    else: 
      cnt += 1
      i += 1
      j += 1
  return (pair[0], cnt)

# [((id1, id2), commonFriendsCnt), ...]
secondDegreeWithCommonCountRDD = secondDegreeWithFriendsRDD.groupByKey() \
    .map(lambda x: countCommonFriends(x)) 
#secondDegreeWithCommonCountRDD.take(10)

secondDegreeWithCommonCountRDD.take(5)

def get10Recommendations(pair):
  id = pair[0]
  seq = pair[1]
  recom = []
  for (sdf, cnt) in seq:
    if cnt == 0: continue
    if len(recom) == 0:
      recom.append((sdf, cnt))
    else:
      i = 0
      while i < len(recom):
        if cnt > recom[i][1]: break;
        elif cnt == recom[i][1]:
          if sdf < recom[i][0]: break;
          else: i += 1
        else: i += 1
      if len(recom) < 10 or i < 10: recom.insert(i, (sdf, cnt))
      if len(recom) > 10: del recom[-1]
  return (id, recom)
recommendationRDD = secondDegreeWithCommonCountRDD \
  .map(lambda pair: (pair[0][0], (pair[0][1], pair[1]))) \
  .groupByKey() \
  .map(lambda pair: get10Recommendations(pair)) \
  .mapValues(lambda v: [pair[0] for pair in v])
recommendationRDD.lookup(11)

noFriendRDD = firstDegreeListOfFriendsRDD.subtractByKey(recommendationRDD) \
  .mapValues(lambda v: [])

noFriendRDD.take(10)

def formatOutput(pair):
  return str(pair[0])+"\t"+",".join([str(id) for id in pair[1]])

recommendationRDD.union(noFriendRDD).map(lambda pair: formatOutput(pair)) \
  .saveAsTextFile("soc-LiveJournal1Ajd-res")

!grep ^11 soc-LiveJournal1Ajd-res.txt/*

#from google.colab import drive
#drive.mount('/content/drive')
#!mkdir ./drive/MyDrive/soc-LiveJournal1Ajd-res/
#!cp -r soc-LiveJournal1Ajd-res/* ./drive/MyDrive/soc-LiveJournal1Ajd-res/

# -*- coding: utf-8 -*-
"""Friend Recommendation.ipynb

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/1sqykgz_IkAOnroLacly1KMVl8ahTwb8E
"""
## Friend Recommendation

# Set up

!pip install pyspark
!pip install -U -q PyDrive
!apt-get update 
!apt-get install openjdk-8-jdk-headless -qq
import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"

"""Now we authenticate a Google Drive client to download the file we will be processing in our Spark job.

"""

from pydrive.auth import GoogleAuth
from pydrive.drive import GoogleDrive
from google.colab import auth
from oauth2client.client import GoogleCredentials

# Authenticate and create the PyDrive client
auth.authenticate_user()
gauth = GoogleAuth()
gauth.credentials = GoogleCredentials.get_application_default()
drive = GoogleDrive(gauth)

id='google-drive-file-id'
downloaded = drive.CreateFile({'id': id})
downloaded.GetContentFile('test-data.txt')


# Commented out IPython magic to ensure Python compatibility.
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
# %matplotlib inline

import pyspark
from pyspark.sql import *
from pyspark.sql.functions import *
from pyspark import SparkContext, SparkConf

"""initialize the Spark context."""

# create the session
conf = SparkConf().set("spark.ui.port", "4050")

# create the context
sc = pyspark.SparkContext(conf=conf)
spark = SparkSession.builder.getOrCreate()

"""You can easily check the current version and get the link of the web interface. In the Spark UI, you can monitor the progress of your job and debug the performance bottlenecks (if your Colab is running with a **local runtime**)."""

spark

"""If you are running this Colab on the Google hosted runtime, the cell below will create a *ngrok* tunnel which will allow you to still check the Spark UI."""

!wget https://bin.equinox.io/c/4VmDzA7iaHb/ngrok-stable-linux-amd64.zip
!unzip ngrok-stable-linux-amd64.zip

get_ipython().system_raw('./ngrok http 4050 &')

!curl -s http://localhost:4040/api/tunnels | python3 -c \
    "import sys, json; print(json.load(sys.stdin)['tunnels'][0]['public_url'])"

!touch test.txt

# Commented out IPython magic to ensure Python compatibility.
# %pycat test.txt

#txt = sc.textFile("test.txt")
#sep = " "
txt = sc.textFile("test-data.txt")
sep = "\t"

def lineToMapOfListOfFriends(line: str) -> ():
  splittedLine = line.split(sep)
  if len(splittedLine) < 2: return (int(splittedLine[0]), ('L', ''))
  elif (not splittedLine[1].strip()): return (int(splittedLine[0]), ('L', ''))
  else:
    friends = [int(id) for id in splittedLine[1].split(",")]
    friends.sort()
    return (int(splittedLine[0]), ('L', friends))

def lineToLinks(line: str) -> []:
  splittedLine = line.split(sep)
  if len(splittedLine) < 2: return []
  elif (not splittedLine[1].strip()): return []
  friends = splittedLine[1].split(",")
  return([(int(splittedLine[0]), int(f)) for f in friends])

# (id0, ('L', [id1,id2,...idn]))
firstDegreeListOfFriendsRDD = txt.filter(lambda line: line).map(lambda line: lineToMapOfListOfFriends(line))
# (id0, idx)
firstDegreeRDD = txt.filter(lambda line: line.strip()).flatMap(lambda line: lineToLinks(line))
#firstDegreeListOfFriendsRDD.lookup(11)
firstDegreeRDD.lookup(11)

# (id0  ('2D', idx))
secondDegreeRDD = firstDegreeRDD.join(firstDegreeRDD) \
    .filter(lambda pair: pair[1][0] != pair[1][1]) \
    .map(lambda pair: (pair[1][0], ('2D', pair[1][1]))) \
    .distinct()

secondDegreeRDD.lookup(11)

# (id0, ('L', [id1,id2,...,idn]))
# (id0  ('2D', 'idx'))
tempMergedRDD = firstDegreeListOfFriendsRDD.union(secondDegreeRDD)

def merge2DFriendNListOfFriends(pair):
  id = pair[0]
  seq = pair[1].__iter__()
  friendsList = ''
  secondDegreeFriends = []
  for (flag, v) in seq:
    if flag == 'L':
      friendsList = v
    elif flag == '2D':
      secondDegreeFriends.append(v)
  return [((id, sdf), (id, friendsList)) for sdf in secondDegreeFriends] + [((sdf, id), (id, friendsList)) for sdf in secondDegreeFriends]
# [((id1, id2), (id1, [idx,idy,idz...])),
#  ((id1, id2), (id2, [idm,idn,ido...])), ...]
secondDegreeWithFriendsRDD = tempMergedRDD.groupByKey().flatMap(lambda pair: merge2DFriendNListOfFriends(pair))

tempMergedRDD.lookup(11)

secondDegreeWithFriendsRDD.lookup((11, 27573))

def countCommonFriends(pair):
  seq = pair[1].__iter__()
  data1 = seq.__next__()
  print(data1)
  data2 = seq.__next__()
  id1 = data1[0]
  list1 = data1[1]
  id2 = data2[0]
  list2 = data2[1]

  # return 0 if they are already friends
  if(id1 in list2 or id2 in list1): return (pair[0], 0)

  # make sure the friends list is sorted
  def isSorted(l: list):
    if(len(l) <= 1): return True
    for i in range(1, len(l)):
       if(l[i]<l[i-1]): return False
    return True
  
  if(not isSorted(list1)): 
    raise Exception(id1, data1[1])
  if(not isSorted(list2)):
    raise Exception(id2, data2[1])
  
  # only when the friends list is sorted, can use below algorithm to get common friends count
  i,j,cnt = 0,0,0
  while i<len(list1) and j < len(list2):
    if(list1[i] < list2[j]): i += 1
    elif(list1[i] > list2[j]): j += 1
    else: 
      cnt += 1
      i += 1
      j += 1
  return (pair[0], cnt)

# [((id1, id2), commonFriendsCnt), ...]
secondDegreeWithCommonCountRDD = secondDegreeWithFriendsRDD.groupByKey() \
    .map(lambda x: countCommonFriends(x)) 
#secondDegreeWithCommonCountRDD.take(10)

secondDegreeWithCommonCountRDD.take(5)

def get10Recommendations(pair):
  id = pair[0]
  seq = pair[1]
  recom = []
  for (sdf, cnt) in seq:
    if cnt == 0: continue
    if len(recom) == 0:
      recom.append((sdf, cnt))
    else:
      i = 0
      while i < len(recom):
        if cnt > recom[i][1]: break;
        elif cnt == recom[i][1]:
          if sdf < recom[i][0]: break;
          else: i += 1
        else: i += 1
      if len(recom) < 10 or i < 10: recom.insert(i, (sdf, cnt))
      if len(recom) > 10: del recom[-1]
  return (id, recom)
recommendationRDD = secondDegreeWithCommonCountRDD \
  .map(lambda pair: (pair[0][0], (pair[0][1], pair[1]))) \
  .groupByKey() \
  .map(lambda pair: get10Recommendations(pair)) \
  .mapValues(lambda v: [pair[0] for pair in v])
recommendationRDD.lookup(11)

noFriendRDD = firstDegreeListOfFriendsRDD.subtractByKey(recommendationRDD) \
  .mapValues(lambda v: [])

noFriendRDD.take(10)

def formatOutput(pair):
  return str(pair[0])+"\t"+",".join([str(id) for id in pair[1]])

recommendationRDD.union(noFriendRDD).map(lambda pair: formatOutput(pair)) \
  .saveAsTextFile("soc-LiveJournal1Ajd-res")

!grep ^11 soc-LiveJournal1Ajd-res.txt/*

#from google.colab import drive
#drive.mount('/content/drive')
#!mkdir ./drive/MyDrive/soc-LiveJournal1Ajd-res/
#!cp -r soc-LiveJournal1Ajd-res/* ./drive/MyDrive/soc-LiveJournal1Ajd-res/
# -*- coding: utf-8 -*- Friend Recommendation.ipynb  Automatically generated by Colaboratory.  Original file is located at     https://colab.research.google.com/drive/1sqykgz_IkAOnroLacly1KMVl8ahTwb8E ## Friend Recommendation # Set up Now we authenticate a Google Drive client to download the file we will be processing in our Spark job. # Authenticate and create the PyDrive client # Commented out IPython magic to ensure Python compatibility. # %matplotlib inline initialize the Spark context. # create the session # create the context You can easily check the current version and get the link of the web interface. In the Spark UI, you can monitor the progress of your job and debug the performance bottlenecks (if your Colab is running with a **local runtime**). If you are running this Colab on the Google hosted runtime, the cell below will create a *ngrok* tunnel which will allow you to still check the Spark UI. # Commented out IPython magic to ensure Python compatibility. # %pycat test.txt #txt = sc.textFile("test.txt") #sep = " " # (id0, ('L', [id1,id2,...idn])) # (id0, idx) #firstDegreeListOfFriendsRDD.lookup(11) # (id0  ('2D', idx)) # (id0, ('L', [id1,id2,...,idn])) # (id0  ('2D', 'idx')) # [((id1, id2), (id1, [idx,idy,idz...])), #  ((id1, id2), (id2, [idm,idn,ido...])), ...] # return 0 if they are already friends # make sure the friends list is sorted # only when the friends list is sorted, can use below algorithm to get common friends count # [((id1, id2), commonFriendsCnt), ...] #secondDegreeWithCommonCountRDD.take(10) #from google.colab import drive #drive.mount('/content/drive') #!mkdir ./drive/MyDrive/soc-LiveJournal1Ajd-res/ #!cp -r soc-LiveJournal1Ajd-res/* ./drive/MyDrive/soc-LiveJournal1Ajd-res/