Алгоритм поиска кратчайшего пути от одного пользователя к другому BFS

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
spark = SparkSession.builder.appName("TwitterShortestPath").getOrCreate()
 
df = spark.read.format("text").options(header=True, inferSchema=True).load("/data/twitter/twitter_sample_small.txt")
df = df.selectExpr("split(value, '\t')[0] as user_id", "split(value, '\t')[1] as follower_id")

from pyspark.sql import SparkSession
from pyspark.sql.functions import col
 
spark = SparkSession.builder.appName("TwitterBFS").getOrCreate()
 
df = spark.read.format("text").options(header=False, inferSchema=True).load("/data/twitter/twitter_sample_small.txt")
df = df.selectExpr("split(value, '\t')[0] as user_id", "split(value, '\t')[1] as follower_id")
 
start_user_id = 12
end_user_id = 34
 
previous_users_table = spark.createDataFrame([(start_user_id,)], ["user_id"])
 
previous_users_table.show(10, False)
 
path_length = 0
 
while previous_users_table.filter(col("user_id") == end_user_id).count() == 0 and previous_users_table.count() > 0:
    path_length += 1
    next_level_users = df.join(previous_users_table, df.follower_id == previous_users_table.user_id, "inner").select(df.user_id.alias("user_id"))
 
    next_level_users.createOrReplaceTempView("next_level_users")
 
    previous_users_table = next_level_users
    
print(path_length)