将列添加到数据帧并在pyspark中更新

Question

我在pyspark:中有一个数据帧

ratings = spark.createDataFrame(
sc.textFile("transactions.json").map(lambda l: json.loads(l)),
)
ratings.show()
+--------+-------------------+------------+----------+-------------+-------+
|click_id|         created_at|          ip|product_id|product_price|user_id|
+--------+-------------------+------------+----------+-------------+-------+
|     123|2016-10-03 12:50:33| 10.10.10.10|     98373|        220.5|      1|
|     124|2017-02-03 11:51:33| 10.13.10.10|     97373|        320.5|      1|
|     125|2017-10-03 12:52:33| 192.168.2.1|     96373|         20.5|      1|
|     126|2017-10-03 13:50:33|172.16.11.10|     88373|        220.5|      2|
|     127|2017-10-03 13:51:33| 10.12.15.15|     87373|        320.5|      2|
|     128|2017-10-03 13:52:33|192.168.1.10|     86373|         20.5|      2|
|     129|2017-08-03 14:50:33| 10.13.10.10|     78373|        220.5|      3|
|     130|2017-10-03 14:51:33| 12.168.1.60|     77373|        320.5|      3|
|     131|2017-10-03 14:52:33| 10.10.30.30|     76373|         20.5|      3|
+--------+-------------------+------------+----------+-------------+-------+
ratings.registerTempTable("transactions")
final_df = sqlContext.sql("select * from transactions");

我想在此数据帧中添加一个名为status的新列，然后基于created_at和user_id更新状态列。

从给定的表transations读取created_at和user_id，并将其传递给返回status的函数get_status(user_id,created_at)。该status需要作为对应的user_id和created_at的新列放入事务表中

我可以在pyspark中运行alter和update命令吗？如何使用pyspark实现这一点？

Answer 1

不清楚您到底想做什么。你应该看看window functions，他们允许你比较，求和。。。框架中的行。

例如

import pyspark.sql.functions as psf
from pyspark.sql import Window
w = Window.partitionBy("user_id").orderBy(psf.desc("created_at"))
ratings.withColumn(
"status", 
psf.when(psf.row_number().over(w) == 1, "active").otherwise("inactive")).sort("click_id").show()
+--------+-------------------+------------+----------+-------------+-------+--------+
|click_id|         created_at|          ip|product_id|product_price|user_id|  status|
+--------+-------------------+------------+----------+-------------+-------+--------+
|     123|2016-10-03 12:50:33| 10.10.10.10|     98373|        220.5|      1|inactive|
|     124|2017-02-03 11:51:33| 10.13.10.10|     97373|        320.5|      1|inactive|
|     125|2017-10-03 12:52:33| 192.168.2.1|     96373|         20.5|      1|  active|
|     126|2017-10-03 13:50:33|172.16.11.10|     88373|        220.5|      2|inactive|
|     127|2017-10-03 13:51:33| 10.12.15.15|     87373|        320.5|      2|inactive|
|     128|2017-10-03 13:52:33|192.168.1.10|     86373|         20.5|      2|  active|
|     129|2017-08-03 14:50:33| 10.13.10.10|     78373|        220.5|      3|inactive|
|     130|2017-10-03 14:51:33| 12.168.1.60|     77373|        320.5|      3|inactive|
|     131|2017-10-03 14:52:33| 10.10.30.30|     76373|         20.5|      3|  active|
+--------+-------------------+------------+----------+-------------+-------+--------+

它为您提供每个用户的最后一次点击

如果要传递一个UDF以从两个现有列创建一个新列。假设您有一个函数，它将user_id和created_at作为参数

from pyspark.sql.types import *
def get_status(user_id,created_at): 
...
get_status_udf = psf.udf(get_status, StringType())

StringType()或函数输出的任何数据类型

ratings.withColumn("status", get_status_udf("user_id", "created_at"))