Write BFS search function in Apache Spark and pandas dataframe:

# This is a pairs notation of the edges, for simplicity of visualization graph $=[(\text{'}$A$\text{'},\text{'}$B$\text{'}),(\text{'}$A$\text{'},\text{'}$C$\text{'}),(\text{'}$A$\text{'},\text{'}$D$\text{'}),(\text{'}$C$\text{'},\text{'}$F$\text{'}),(\text{'}$F$\text{'},\text{'}$A$\text{'}),(\text{'}$B$\text{'},\text{'}$G$\text{'}),(\text{'}$G$\text{'},\text{'}$H$\text{'}),(\text{'}$D$\text{'},\text{'}$E$\text{'})]$ # Here's an equivalent dictionary representation that we can use for a # Pandas DataFrame... simple$\_$dict $=\{\text{'}$from$\_$node': $[\text{'}$A$\text{'},\text{'}$A$\text{'},\text{'}$A$\text{'},\text{'}$C$\text{'},\text{'}$F$\text{'},\text{'}$B$\text{'},\text{'}$G$\text{'},\text{'}$D$\text{'}],\text{'}$to$\_$node': $[\text{'}$B$\text{'},\text{'}$C$\text{'},\text{'}$D$\text{'},\text{'}$F$\text{'},\text{'}$A$\text{'},\text{'}$G$\text{'},\text{'}$H$\text{'},\text{'}$E$\text{'}]\}$ simple$\_$graph$\_$df $=$ pd$.$DataFrame.from$\_$dict$($simple$\_$dict$)$ simple$\_$graph$\_$sdf $=$ spark.createDataFrame$($simple$\_$graph$\_$df$)$ simple$\_$graph$\_$sdf$.$show$()$

As you can see, each row of this dataframe represents an edge between two nodes Although the nodes are labeled "from" and $"$to$",$ the edges are actually undirected, meaning that A$-->$B represents the same edge as B$-->$A$.$

Let's define our starting node as follows:

smallOrig $=[\{\text{'}$node$\text{'}$: $\text{'}$A$\text{'}\}]$

TODO: Write spark$\_$bfs$\_1\_$round$($visted$\_$nodes$)$ that takes the currently dataframe of visited$\_$nodes, performs one round of BFS$,$ and returns an updated visited nodes dataframe. You should assume that a temporary sdf G already exists.

def spark$\_$bfs$\_1\_$round$($visited$\_$nodes$)$: $"""$ :param visited$\_$nodes: dataframe with columns node and distance :return: dataframe of updated visuted nodes, with columns node and distance $"""$ #TODO

Now, run the inner function on simple$\_1\_$round$\_$bfs$\_$sdf result of $1$ round of BFS on simple graph and store the results in simple$\_$bfs$\_$result. This is ultimately what the output of BFS to depth $2$ should look like.

simple$\_$graph$\_$sdf$.$createOrReplaceTempView$(\text{'}$G$\text{'})$ simple$\_$bfs$\_$result $=$ #TODO simple$\_$bfs$\_$result.show$()$

Convert this result to Pandas, sorted by the node.

simple$\_$bfs$\_$test $=$ #TODO

Now, we will fully implement spark$\_$bfs$.$ This function should iteratively call your implemented version of spark$\_$bfs$\_1\_$round and ultimately return the output of this function at max$\_$depth.

You are also responsible for initializing the starting dataframe, that is converting the list of origin nodes into a spark dataframe with the nodes logged at distance $0.$

Consider the following:

schema $=$ StructType$([$ StructField$("$node$",$ StringType$(),$ True$)])$ my$\_$sdf $=$ spark.read.format$("$csv$").$schema$($schema$).$load$("$my$.$csv$")$

The schema ultimately specifies the structure of the Spark DataFrame with a string node column. It then calls spark.load to read the CSV with this schema. Also, you are responsible for ensuring that a view of your graph is available within this function. $($Note: you will also need to add in a distance column$)$

TODO: implement spark$\_$bfs$($G$,$origins,max$\_$depth$)$ and run on review$\_$graph$\_$sdf initalized in $4\mathrm{.}3.$ Note: you may want to run tests on the simple$\_$graphexample as the review$\_$graph$\_$sdf will take quite some time to run.

# TODO: iterative search over undirected graph # Worth $5$ points directly, but will be needed later def spark$\_$bfs$($G$,$ origins, max$\_$depth$)$: $"""$ runs distributed BFS to a specified max depth :param G: graph dataframe from $4\mathrm{.}3$ :param origins: list of origin nodes stored as $\{"$node$"$: nodeValue$\}$ :param max$\_$depth: integer value of max depth to run BFS to :return: dataframe with columns node, distance of all visited nodes $"""$ #TODO # Remember that if you want to go from Pandas dataframes to Spark dataframes, # you may need to write to a CSV and read it back.