<jupyter_start><jupyter_text>Finding seeds from tree (3 different findSeed functions)<jupyter_code>btc_raw = sc.parallelize([(0,1), (1,2), (2,5), (5,8), (7,8), (3,7), (3,4), (3,6), (10,11), (10,12), (12,13)])
G = btc_raw.flatMap(lambda x: [x, (x[1], x[0])]).groupByKey().mapValues(lambda x: set(x))
def Min_Selection_Step(G): #dictionary format RDD
    v_min = G.map(lambda x: (x[0], min(x[1] | {x[0]})))
    NN_G_u = G.map(lambda x: (x[0], (x[1] | {x[0]})))
    
    #Broadcasting
    v_min_bc = sc.broadcast(dict(v_min.collect()))
    addEdge = NN_G_u.map(lambda x: (x[0], (x[1], v_min_bc.value[x[0]]))).flatMap(lambda x: [(y, x[1][1]) for y in x[1][0]])
    #Without broadcasting
    #addEdge = NN_G_u.join(v_min).flatMap(lambda x: [(y, x[1][1]) for y in x[1][0]])
    #H = addEdge.groupByKey().map(lambda x: (x[0], set(x[1])))
    H = addEdge.groupByKey().mapValues(lambda x: set(x))
    return H

def Pruning_Step(H, T):
    
    #minimum node of the neighborhood: shared for following parts
    v_min = H.map(lambda x: (x[0], min(x[1])))
    v_min_bc = sc.broadcast(dict(v_min.collect())) #Broadcasting v_min
    
    #---------------G construction-------------------
    H_filtered = H.filter(lambda x: len(x[1]) > 1)
    #NN_H_u = H_filtered.map(lambda x: (x[0], x[1] - {min(x[1])} ))
    NN_H_u = H_filtered.mapValues(lambda x: x - {min(x)} )
    #With Broadcasting
    addEdge2=NN_H_u.map(lambda x:(x[0],(x[1],v_min_bc.value[x[0]]))).flatMap(lambda x:[(x[1][1],y) for y in x[1][0]])
    #Without broadcasting
    #addEdge2 = NN_H_u.join(v_min).flatMap(lambda x: [(x[1][1], y) for y in x[1][0]])
    G = addEdge2.flatMap(lambda x: [x, (x[1], x[0])]).groupByKey().mapValues(lambda x: set(x))
    
    #---------------Tree construction--------------
    deactiveNodes = H.filter(lambda x: x[0] not in x[1]).mapValues(lambda x: None)
    #Without broadcasting
    #addEdge3 = deactiveNodes.join(v_min).map(lambda x: (x[1][1], x[0]))
    #With Broadcasting
    addEdge3 = deactiveNodes.map(lambda x: (x[0], (x[1], v_min_bc.value[x[0]]))).map(lambda x: (x[1][1], x[0]))
    T = T.union(addEdge3)

    return [G, T]

#Finding seeds
def findSeeds(T): 
    keys = T
    values = T.map(lambda x:(x[1], x[0]))
    return keys.subtractByKey(values).keys().distinct()

def findSeeds1(T):
    keys = T.keys().distinct().map(lambda x:(x,1))
    values = T.values().distinct().map(lambda x:(x,1))
    return keys.subtractByKey(values).keys()

def findSeeds2(T):
    T_inv = T.map(lambda x:(x[1], x[0]))
    A = T.keys().distinct().map(lambda x:(x,1))        #Each distinct is a reduceByKey
    B = T_inv.keys().distinct().map(lambda x:(x,1))
    return A.leftOuterJoin(B).filter(lambda x: not x[1][1]).keys()

def Cracker(G):
    n = 0
    T = sc.parallelize([])
    while G.take(1):
        n += 1
        H = Min_Selection_Step(G)
        G, T = Pruning_Step(H, T)
    
    return findSeeds2(T)
%%time 
#Cracker with findSeeds
Cracker(G).collect()<jupyter_output>CPU times: user 176 ms, sys: 26.7 ms, total: 202 ms
Wall time: 1.34 s<jupyter_text>Tracking activeness and seeds<jupyter_code>btc_raw = sc.parallelize([(0,1), (1,2), (2,5), (5,8), (7,8), (3,7), (3,4), (3,6), (10,11), (10,12), (12,13)])
G2 = btc_raw.flatMap(lambda x: [x, (x[1], x[0])]).groupByKey().mapValues(lambda x: (set(x), True))
def Min_Selection_Step(G): #dictionary format RDD
    v_min = G.map(lambda x: (x[0], min(x[1][0] | {x[0]})))
    NN_G_u = G.map(lambda x: (x[0], (x[1][0] | {x[0]}, x[1][1])))
    
    #Broadcasting
    v_min_bc = sc.broadcast(dict(v_min.collect()))
    addEdge = NN_G_u.map(lambda x: (x[0], (x[1][0], v_min_bc.value[x[0]], x[1][1])))
    addEdge1 = addEdge.flatMap(lambda x: [(y, (x[1][1], x[1][2])) for y in x[1][0]])
    
    temp = addEdge1.groupByKey().map(lambda x: (x[0], (list(x[1]))))
    H = temp.map(lambda x: (x[0], list(zip(*x[1])))).mapValues(lambda x: (set(x[0]), all(x[1])))
    return H

def Pruning_Step(H, T, Seeds):
    
    #minimum node of the neighborhood: shared for following parts
    v_min = H.map(lambda x: (x[0], min(x[1][0])))
    v_min_bc = sc.broadcast(dict(v_min.collect())) #Broadcasting v_min

    #---------------G construction-------------------
    H_filtered = H.filter(lambda x: len(x[1][0]) > 1)
    NN_H_u = H_filtered.mapValues(lambda x: (x[0] - {min(x[0])}, x[1] ))
    #With Broadcasting
    addEdge2=NN_H_u.map(lambda x:(x[0],(x[1][0],v_min_bc.value[x[0]], x[1][1]))).flatMap(lambda x:[(x[1][1],(y, x[1][2])) for y in x[1][0]])

    temp = addEdge2.flatMap(lambda x: [x, (x[1][0], (x[0],x[1][1]))]).groupByKey().mapValues(lambda x: list(x))
    G = temp.mapValues(lambda x: list(zip(*x))).mapValues(lambda x: (set(x[0]), all(x[1])))
    
    #---------------Tree construction--------------
    #The deactivated Nodes do not appear in G_{t+1}
    deactiveNodes = H.filter(lambda x: x[0] not in x[1][0]).mapValues(lambda x: False)
    #With Broadcasting
    addEdge3 = deactiveNodes.map(lambda x: (x[0], (x[1], v_min_bc.value[x[0]]))).map(lambda x: (x[1][1], x[0]))
    T = T.union(addEdge3)
    
    #--------------Find Seed-----------------
    #Elements in H with neighborhood from G_{t+1}
    NN_G_H = H.cogroup(G).mapValues(lambda x: arr( (list(x[0]), list(x[1])) ) ) 
    
    #---->Not sure if it is necessary to use True/False
    #deactivated = NN_G_H.cogroup(deactiveNodes).map(lambda x: (x[0], arr2( (list(x[1][0]), list(x[1][1])) ) ) )
    #seed = deactivated.filter(lambda x: (len(x[1][0]) <= 1) & (x[0] in x[1][0]) & x[1][1])
    
    seed = NN_G_H.filter(lambda x: (len(x[1][0]) <= 1) & (x[0] in x[1][0]))
    Seeds = Seeds.union(seed)
    return [G, T, Seeds]

def arr(value):
    if not value[1]:
        return value[0][0]
    else:
        temp = list(zip(*[value[0][0], value[1][0]]))
        return temp[0][0].union(temp[0][1]), all(temp[1])

def arr2(value):
    if not value[1]:
        return value[0][0]
    else:
        return (value[0][0][0], False)

def Cracker(G):
    n = 0
    T = sc.parallelize([])
    Seeds = sc.parallelize([])
    while G.take(1):
        n += 1
        H = Min_Selection_Step(G)
        G, T, Seeds = Pruning_Step(H, T, Seeds)
    
    return Seeds.keys()
%%time
Cracker(G2).collect()<jupyter_output>CPU times: user 207 ms, sys: 25.4 ms, total: 232 ms
Wall time: 1.12 s<jupyter_text>Tracking seeds without activeness<jupyter_code>btc_raw = sc.parallelize([(0,1), (1,2), (2,5), (5,8), (7,8), (3,7), (3,4), (3,6), (10,11), (10,12), (12,13)])
G = btc_raw.flatMap(lambda x: [x, (x[1], x[0])]).groupByKey().mapValues(lambda x: set(x))
def Min_Selection_Step(G): #dictionary format RDD
    v_min = G.map(lambda x: (x[0], min(x[1] | {x[0]})))
    NN_G_u = G.map(lambda x: (x[0], x[1] | {x[0]}))
    #Broadcasting
    v_min_bc = sc.broadcast(dict(v_min.collect()))
    addEdge = NN_G_u.map(lambda x: (x[0], (x[1], v_min_bc.value[x[0]])) )
    addEdge1 = addEdge.flatMap(lambda x: [(y, x[1][1]) for y in x[1][0]])
    #Without broadcasting
    #addEdge = NN_G_u.join(v_min).flatMap(lambda x: [(y, x[1][1]) for y in x[1][0]])

    H = addEdge1.groupByKey().mapValues(lambda x: set(x))
    return H

def Pruning_Step(H, T, Seeds):
    #minimum node of the neighborhood: shared for following parts
    v_min = H.mapValues(lambda x: min(x))
    v_min_bc = sc.broadcast(dict(v_min.collect())) #Broadcasting v_min
    
    #---------------G construction-------------------
    H_filtered = H.filter(lambda x: len(x[1]) > 1)
    NN_H_u = H_filtered.mapValues(lambda x: x - {min(x)} )
    #With Broadcasting
    addEdge2=NN_H_u.map(lambda x:(x[0],(x[1],v_min_bc.value[x[0]]))).flatMap(lambda x:[(x[1][1],y) for y in x[1][0]])
    #Without broadcasting
    #addEdge2 = NN_H_u.join(v_min).flatMap(lambda x: [(x[1][1], y) for y in x[1][0]])
    G = addEdge2.flatMap(lambda x: [x, (x[1], x[0])]).groupByKey().mapValues(lambda x: set(x))
    
    #---------------Tree construction--------------
    #The deactivated Nodes do not appear in G_{t+1}
    deactiveNodes = H.filter(lambda x: x[0] not in x[1]).mapValues(lambda x: False)
    #Without broadcasting
    #addEdge3 = deactiveNodes.join(v_min).map(lambda x: (x[1][1], x[0]))
    #With Broadcasting
    addEdge3 = deactiveNodes.map(lambda x: (x[0], (x[1], v_min_bc.value[x[0]]))).map(lambda x: (x[1][1], x[0]))
    T = T.union(addEdge3)

    #--------------Find Seed-----------------
    #Elements in H with neighborhood from G_{t+1}
    NN_G_H = H.cogroup(G).mapValues(lambda x: (list(x[0]), list(x[1])) ).mapValues(lambda x: set_join(x) )

    #Not sure is necessary to use True/False
    #deactivated = NN_G_H.cogroup(deactiveNodes).map(lambda x: (x[0], (list(x[1][0]), list(x[1][1])) ))
    #seed = deactivated.filter(lambda x: (len(x[1][0]) <= 1) & (x[0] in x[1][0]) & x[1][1]) 
    
    seed = NN_G_H.filter(lambda x: (len(x[1]) <= 1) & (x[0] in x[1]))
    Seeds = Seeds.union(seed)

    return [G, T, Seeds]


def set_join(value):
    if not value[1]:
        return value[0][0]
    else:
        return value[0][0] | value[1][0]

def Cracker(G):
    n = 0
    T = sc.parallelize([])
    Seeds = sc.parallelize([])
    while G.take(1):
        n += 1
        H = Min_Selection_Step(G)
        G, T, Seeds = Pruning_Step(H, T, Seeds)
    
    return Seeds.keys()
%%time
Cracker(G).collect()<jupyter_output>CPU times: user 156 ms, sys: 22.9 ms, total: 179 ms
Wall time: 824 ms
{'fraction_non_alphanumeric': 0.23420034393809114, 'fraction_numerical': 0.04299226139294927, 'mean_word_length': 3.3238847583643123, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 14, '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': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '507587', 'license': 'MIT', 'n_tokens_mistral': 4037, 'n_tokens_neox': 3883, 'n_words': 826, 'path': 'rdd_seeds_optim.ipynb', 'repo_name': 'Miguel-Jiahao-Wang/Connected_Components_in_Graphs', 'size': '9304'}