如何编写一个转换函数来转换RDD引用到一个Graphframe对象

Question

我有一个Graphframe对象:g和一个RDD对象:candidate:

g = GraphFrame(v,e)
candidates_rdd.collect() 
#  [Row(source=u'a', target=u'b'),
#   Row(source=u'a', target=u'c'),
#   Row(source=u'e', target=u'a')]

我想在candidates_rdd中计算从"源"到"目标"的路径，并使用graphframe的广度优先搜索生成具有键，值对((源，目标)，path_list)的结果rdd，其中path_list是从源到目标的路径列表。

输出示例:

(('a','b'),['a-c-b','a-d-e-b']), 
(('f','c'),[]),
(('a',d'),['a-b-e-d']

我写了下面的函数:

def bfs_(row):    
    arg1 = "id = '" + row.source + "'"
    arg2 = "id = '" + row.target + "'"        
    return ((row.source, row.target), g.bfs(arg1,arg2).rdd)
results = candidates_rdd.map(bfs_)

我得到这个错误:

Py4JError: An error occurred while calling o274.__getnewargs__. Trace:
py4j.Py4JException: Method __getnewargs__([]) does not exist

我试过使图形全局或广播它，都不工作。

有谁能帮我一下吗?

非常感谢!

Answer 1

不可能。

Spark不支持这样的嵌套操作。外循环必须是非分布式的:

>>> [g.bfs(arg1, arg2) for arg1, arg2 in candidates_rdd.collect()]