python之使用 Spark 检查日志

我是 Spark 的新手，我正在尝试开发一个 python 脚本来读取带有一些日志的 csv 文件:

userId,timestamp,ip,event 
13,2016-12-29 16:53:44,86.20.90.121,login 
43,2016-12-29 16:53:44,106.9.38.79,login 
66,2016-12-29 16:53:44,204.102.78.108,logoff 
101,2016-12-29 16:53:44,14.139.102.226,login 
91,2016-12-29 16:53:44,23.195.2.174,logoff 

并检查用户是否有一些奇怪的行为，例如他是否连续两次“登录”而没有“注销”。我已将 csv 加载为 Spark 数据帧，我想比较单个用户的日志行，按时间戳排序并检查两个连续事件是否属于同一类型(登录 - 登录，注销 - 注销)。我正在寻找以“map-reduce”方式进行的操作，但目前我无法弄清楚如何使用比较连续行的 reduce 函数。 我编写的代码有效，但性能很差。

sc = SparkContext("local","Data Check") 
sqlContext = SQLContext(sc) 
 
LOG_FILE_PATH = "hdfs://quickstart.cloudera:8020/user/cloudera/flume/events/*" 
RESULTS_FILE_PATH = "hdfs://quickstart.cloudera:8020/user/cloudera/spark/script_results/prova/bad_users.csv" 
N_USERS = 10*1000 
 
dataFrame = sqlContext.read.format("com.databricks.spark.csv").load(LOG_FILE_PATH) 
dataFrame = dataFrame.selectExpr("C0 as userID","C1 as timestamp","C2 as ip","C3 as event") 
 
wrongUsers = [] 
 
for i in range(0,N_USERS): 
 
    userDataFrame = dataFrame.where(dataFrame['userId'] == i) 
    userDataFrame = userDataFrame.sort('timestamp') 
 
    prevEvent = '' 
 
    for row in userDataFrame.rdd.collect(): 
 
        currEvent = row[3] 
        if(prevEvent == currEvent): 
            wrongUsers.append(row[0]) 
 
        prevEvent = currEvent 
 
badUsers = sqlContext.createDataFrame(wrongUsers) 
badUsers.write.format("com.databricks.spark.csv").save(RESULTS_FILE_PATH) 

请您参考如下方法:

首先(不相关但仍然)，确保每个用户的条目数不是那么大，因为 collect in for row in userDataFrame.rdd.collect(): 是危险的。

其次，您无需离开此处的DataFrame区域即可使用经典Python，只需坚持使用Spark。

现在，你的问题。它基本上是“对于每一行，我想从上一行了解一些东西”:属于 Window 函数的概念，准确地说是 lag 函数。这里有两篇关于 Spark 中窗口函数的有趣文章:一篇来自 Databricks使用 Python 代码和来自 Xinh 的代码使用(我认为更容易理解)Scala 中的示例。

我在 Scala 中有一个解决方案，但我认为您可以将其翻译成 Python:

import org.apache.spark.sql.expressions.Window 
import org.apache.spark.sql.functions.lag 
 
import sqlContext.implicits._ 
 
val LOG_FILE_PATH = "hdfs://quickstart.cloudera:8020/user/cloudera/flume/events/*" 
val RESULTS_FILE_PATH = "hdfs://quickstart.cloudera:8020/user/cloudera/spark/script_results/prova/bad_users.csv" 
 
val data = sqlContext 
  .read 
  .format("com.databricks.spark.csv") 
  .option("inferSchema", "true") 
  .option("header", "true") // use the header from your csv 
  .load(LOG_FILE_PATH) 
 
val wSpec = Window.partitionBy("userId").orderBy("timestamp") 
 
val badUsers = data 
  .withColumn("previousEvent", lag($"event", 1).over(wSpec)) 
  .filter($"previousEvent" === $"event") 
  .select("userId") 
  .distinct 
 
badUsers.write.format("com.databricks.spark.csv").save(RESULTS_FILE_PATH) 

基本上，您只需检索上一行的值并将其与当前行的值进行比较，如果匹配则为错误行为，您将保留 userId。对于每个 userId 行“ block ”中的第一行，先前的值将为 null:与当前值比较时， bool 表达式将为 false 所以这里没有问题。