如何使用map在pyspark中进行rest api调用

Question

我正在尝试使用 map 对 rdd 进行 put rest api 调用，如下所示：

def put(params, payload):
  
  url = "https://{}/{}".format(server, params)
  headers = {
    'Content-Type': 'application/json' }
  
  response = requests.request("PUT", url, headers=headers, data = payload) 
  return response.status_code

df.select("params", "payload").rdd.map(lambda x, y: put(x, y)).collect()

但我收到一个错误：

org.apache.spark.api.python.PythonException: 'TypeError: () 缺少 1 个必需的位置参数：'payload''

似乎lambda函数没有得到第二个参数payload，不知道为什么。有谁能帮帮我吗？

Answer 1

下面这段代码是罪魁祸首：

df.select("params", "payload").rdd.map(lambda x, y: put(x, y)).collect()

您不能在 lambda 函数中解压数据帧。这是正确的：

df.select("params", "payload").rdd.map(lambda row: put(row[0], row[1])).collect()

Answer 2

我无法对 Nihad 的回答添加评论。因此，为 Nihad 的答案添加一个答案，以及如何将响应合并回 DF 的后续问题。

你可以做如下的事情

df_new = df.select("params", "payload").rdd.map(lambda row: [row[0],
                                                             row[1],
                                                             put(row[0], row[1])]
                                                )
                                       .toDF()

这样您可以将原始数据链接到响应。

Answer 3

为了向数据框添加响应，您必须使用 udf 注册 put 方法并在 withColumn 方法中将其用于数据框。

from pyspark.sql.types import StringType
from pyspark.sql.functions import udf

putUdf = udf(put, StringType())

df = df.withColumn("response", putUdf(df.params, df.payload))

这将创建一个名为 response 的新列并在其中填充 put 输出。