使用复杂函数更新 pyspark 数据框列

Question

是否可以使用 UDF 中不可用的复杂函数更新 pyspark 中的 hiveContext 数据框列？

我有一个包含许多列的数据框，其中 2 列称为时间戳和数据。如果数据中的时间戳满足某些条件，我需要从数据中的 JSON 字符串中检索时间戳并更新时间戳列。我知道数据框是不可变的，但是可以以某种方式构建一个新的数据框，保留旧数据框的所有列但更新 timstamp 列？

说明我想做的代码：

def updateTime(row):
    import json

    THRESHOLD_TIME = 60 * 30
    client_timestamp = json.loads(row['data'])
    client_timestamp = float(client_timestamp['timestamp'])
    server_timestamp = float(row['timestamp'])
    if server_timestamp - client_timestamp <= THRESHOLD_TIME:
        new_row = .....  # copy contents of row
        new_row['timestamp'] = client_timestamp
        return new_row
    else:
        return row

df = df.map(updateTime)

我想过将行内容映射到元组，然后使用 .toDF() 将其转换回数据帧，但我找不到将行内容复制到元组然后取回列名的方法。

Answer 1

如果您调整 updateTime 函数以接收时间戳作为参数并返回新处理的时间戳，您可以创建一个 UDF 并直接在 DataFrame 的列上使用它：

from pyspark.sql.functions import *
from pyspark.sql.types import TimestampType

myUDF = udf(updateTime, TimestampType())
df = df.withColumn("timestamp", myUDF(col("timestamp"))

但是，在你的情况下，我认为它有点复杂：

from pyspark.sql.functions import *
from pyspark.sql.types import TimestampType

myUDF = udf(getClientTime, TimestampType())
client_timestamp = myUDF(col("data"))
server_timestamp = col("timestamp")
condition = server_timestamp.cast("float") - client_timestamp.cast("float") <= THRESHOLD_TIME    

newCol =  when(condition, client_timestamp).otherwise(server_timestamp) 
newDF = df.withColumn("new_timestamp", newCol)

使用第二种方法，函数getClientTime 从data 列接收一个值，并返回该值的客户端时间戳。然后，您可以使用它来创建一个包含此信息的新列 (client_timestamp)。最后，您可以根据server_timestamp 列和新创建的client_timestamp 列的值，使用when 有条件地创建新列。

参考：

• 有关在 Spark 中使用 UDF 的更多信息，您可以查看以下文章：http://www.sparktutorials.net/using-sparksql-udfs-to-create-date-times-in-spark-1.5

• 我使用的函数的主要文档可以在pyspark API docs找到