PySpark UDF Returns 状态代码和响应在单独的 withColumn 中

Question

我有一个可以引用 status_code 和 return 正文的 udf。

def Api(a):
  path = endpoint
  headers = {'sample-Key': sample}
  body = [{'text': body }]
  res = None
  try:
      req = requests.post(path, params=params, headers=headers, json=body)
      req = req.json()
      dumps=json.dumps(req)
  except Exception as e:
    return e
  if res != None and req.status_code == 200:
    return json.loads(dumps)
  return None

udf_Api = udf(Api)
newDF=df.withColumn("output", udf_Api(col("input")))

我可以 return json.loads 并将其放入数据框。但是，我的问题是我还需要将 status_code 保留在单独的列中。所以输出看起来像：

+---------+-----------+----------+
|    input|status_code|    output|
+---------+-----------+----------+
|inputText|        200|outputText|
+---------+-----------+----------+

那么我如何 return req.status_code 和 json.loads() ，但将它们放在数据框中的单独列中？我想到了 return 创建一个数组然后拆分它，但不确定该怎么做。

Answer 1

您可以将 UDF 修改为 return 字典而不是字符串或整数，然后定义输出架构。

from pyspark.sql import functions as F
from pyspark.sql import types as T

def Api(a):
    return {
        'status': 200,
        'data': '{"a": 1}'
    }

schema = T.StructType([
    T.StructField('status', T.IntegerType()),
    T.StructField('data', T.StringType())
])

(df
    .withColumn('output', F.udf(Api, schema)('col'))
    .select('col', 'output.*')
    .show()
)

# +---+------+--------+
# |col|status|    data|
# +---+------+--------+
# | 10|   200|{"a": 1}|
# | 20|   200|{"a": 1}|
# | 30|   200|{"a": 1}|
# +---+------+--------+

PySpark UDF Returns 状态代码和响应在单独的 withColumn 中

PySpark UDF Returns Status Code and Response in Separate withColumn

api

user-defined-functions

pyspark