Spark 性能问题（可能由“基本”错误引起）

Question

我对 Apache Spark（1.6 版）比较陌生，我觉得自己碰壁了：我浏览了 SE 上与 Spark 相关的大部分问题，但到目前为止我没有发现任何对我有帮助的问题。我相信我在基本层面上做错了什么，但我无法指出它到底是什么，特别是因为我编写的其他代码运行良好。

我会尽量具体地解释我的情况，但我会简化我的任务以便更好地理解。请记住，由于我仍在学习它，我正在使用 Spark 的 本地模式 运行此代码；还值得注意的是，我一直在使用 DataFrames（而不是 RDDs）。最后，请注意以下代码是使用 Pyspark 用 Python 编写的，但我确实欢迎使用 Scala 或 Java 的可能解决方案，因为我相信这个问题是一个非常基本的问题。

我有一个通用 JSON 文件，其结构类似于以下内容：

{"events":[ 
    {"Person":"Alex","Shop":"Burger King","Timestamp":"100"},
    {"Person":"Alex","Shop":"McDonalds","Timestamp":"101"},
    {"Person":"Alex","Shop":"McDonalds","Timestamp":"104"},
    {"Person":"Nathan","Shop":"KFC","Timestamp":"100"},
    {"Person":"Nathan","Shop":"KFC","Timestamp":"120"},
    {"Person":"Nathan","Shop":"Burger King","Timestamp":"170"}]}

我需要做的是计算同一个人两次访问同一家商店之间的时间间隔。输出应该是至少有一位顾客每 5 秒至少访问一次的商店列表，以及满足此要求的顾客数量。在上面的例子中，输出应该是这样的：

{"Shop":"McDonalds","PeopleCount":1}

我的想法是为每个 pair（Person、Shop）分配相同的标识符，然后继续验证该对是否满足要求。可以使用窗口函数 ROW_NUMBER() 来分配标识符，这需要在Spark 中使用hiveContext。分配标识符后，上面的文件应该是这样的：

{"events":[ 
    {"Person":"Alex","Shop":"Burger King","Timestamp":"100","ID":1},
    {"Person":"Alex","Shop":"McDonalds","Timestamp":"101", "ID":2},
    {"Person":"Alex","Shop":"McDonalds","Timestamp":"104", "ID":2},
    {"Person":"Nathan","Shop":"KFC","Timestamp":"100","ID":3},
    {"Person":"Nathan","Shop":"KFC","Timestamp":"120","ID":3},
    {"Person":"Nathan","Shop":"Burger King","Timestamp":"170","ID":4}]}

在得出结论之前，我需要为每一对执行几个步骤（其中一些需要使用自连接），因此我使用了临时表。

我写的代码是这样的（当然，我只包含了相关部分——“df”代表“dataframe”）：

t1_df = hiveContext.read.json(inputFileName)
t1_df.registerTempTable("events")
t2_df = hiveContext.sql("SELECT Person, Shop, ROW_NUMBER() OVER (order by Person asc, Shop asc) as ID FROM events group by Person, Shop HAVING count(*)>1") #if there are less than 2 entries for the same pair, then we can discard this pair
t2_df.write.mode("overwrite").saveAsTable("orderedIDs")
n_pairs = t2_df.count() #used to determine how many pairs I need to inspect
i=1
while i<=n_pairs:
    #now I perform several operations, each one displaying this structure
    #first operation...
    query="SELECT ... FROM orderedIDs WHERE ID=%d" %i
    t3_df = hiveContext.sql(query)
    t3_df.write.mode("overwrite").saveAsTable("table1")
    #...second operation...
    query2="SELECT ... FROM table1 WHERE ..."
    t4_df = hiveContext.sql(query2)
    temp3_df.write.mode("overwrite").saveAsTable("table2")
    #...and so on. Let us skip to the last operation in this loop, which consists of the "saving" of the shop if it met the requirements:
    t8_df = hiveContext.sql("SELECT Shop from table7")
    t8_df.write.mode("append").saveAsTable("goodShops")
    i=i+1

#then we only need to write the table to a proper file
output_df = hiveContext.sql("SELECT Shop, count(*) as PeopleCount from goodShops group by Shop")
output_df.write.json('output')

现在，问题来了：输出是正确的。在这方面，我尝试了几种输入，并且程序运行良好。然而，它非常慢：分析每对大约需要 15-20 秒，不管每对有什么条目。因此，例如，如果有 10 对，大约需要 3 分钟，如果有 100 对，则需要 30 分钟，依此类推。我在几台硬件相对不错的机器上运行了这段代码，但没有任何改变。 我还尝试缓存我使用的部分（甚至全部）表，但问题仍然存在（在某些情况下甚至会增加所需的时间）。更具体地说，循环的最后一个操作（使用“追加”的那个）需要几秒钟才能完成（从 5 到 10），而前 6 只需要 1-2 秒（考虑到任务范围，但肯定更易于管理）。

我认为问题可能在于以下一项（或多项）：

1. 使用循环，可能会导致并行问题；
2. 使用“saveAsTable”方法，强制写入 I/O
3. 缓存使用不当或使用不当

这三个是我唯一想到的，因为我使用 Spark 编写的其他软件（我没有遇到任何性能问题）没有使用上述技术，因为我基本上执行简单的 JOIN 操作并使用 registerTempTable 方法来使用临时表（据我了解，它不能在 loop 中使用）而不是 saveAsTable em> 方法。

我试图尽可能清楚，但如果您确实需要更多详细信息，我愿意提供更多信息。

编辑：感谢 zero323 的回答，我设法解决了我的问题。事实上，使用 LAG 功能是我真正需要做的事情。另一方面，我了解到不应该使用“saveAsTable”方法——尤其是在循环中——因为每次调用它都会导致性能大幅下降。除非绝对必要，否则我将避免使用它。

Answer 1

同一个人两次访问同一家商店之间的时间间隔。输出应该是至少有一位顾客每 5 秒至少光顾一次的商店列表，以及满足此要求的顾客数量。

使用聚合的简单lag 怎么样：

from pyspark.sql.window import Window
from pyspark.sql.functions import col, lag, sum

df = (sc
    .parallelize([
        ("Alex", "Burger King", "100"), ("Alex", "McDonalds", "101"),
        ("Alex", "McDonalds", "104"), ("Nathan", "KFC", "100"),
        ("Nathan", "KFC", "120"), ("Nathan", "Burger King", "170")
    ]).toDF(["Person", "Shop", "Timestamp"])
    .withColumn("Timestamp", col("timestamp").cast("long")))

w = (Window()
    .partitionBy("Person", "Shop")
    .orderBy("timestamp"))

ind = ((
    # Difference between current and previous timestamp le 5
    col("Timestamp") - lag("Timestamp", 1).over(w)) <= 5
 ).cast("long") # Cast so we can sum

(df
    .withColumn("ind", ind)
    .groupBy("Shop")
    .agg(sum("ind").alias("events"))
    .where(col("events") > 0)
    .show())

## +---------+------+
## |     Shop|events|
## +---------+------+
## |McDonalds|     1|
## +---------+------+