数据分析之图算法: 基于Spark和Neo4j

46

｜

第

4

章

.drop(new_distances["id"])

.withColumn("visited", new_visited_col)

.withColumn("newDistance", new_distance_col)

.withColumn("newPath", new_path_col)

.drop("aggMess", "distance", "path")

.withColumnRenamed('newDistance', 'distance')

.withColumnRenamed('newPath', 'path'))

cached_new_vertices = AM.getCachedDataFrame(new_vertices)

g2 = GraphFrame(cached_new_vertices, g2.edges)

if

g2.vertices.filter(g2.vertices.id == destination).first().visited:

return

(g2.vertices.filter(g2.vertices.id == destination)

.withColumn("newPath", add_path_udf("path", "id"))

.drop("visited", "path")

.withColumnRenamed("newPath", "path"))

return ...