如何将 PySpark Dataframe 列的类型指定为 JSON

Question

以下是我们的 pyspark 应用程序代码 sn-p。

schema = StructType(
    [
        StructField('name', StringType(), True),
        StructField('version', StringType(), True),
        StructField('requestBody', StringType(), True),
        StructField('id', StringType(), True),
    ]
)

df_new = df.withColumn('value', from_json('value', schema)) \
    .where(col('value.version') == '1') \
    .select(col('value.*'))\
    .na.drop() \
    .withColumn('requestBody', decrypt_udf(col('requestBody')))

df_new.show()

+-------+--------+---------------------------------------------+---+
|   name| version|                                  requestBody| id|
+-------+--------+---------------------------------------------+---+
|kj-test|       1|{"data": {"score": 130, "group": "silver"}}  |  1|
|kj-test|       1|{"data": {"score": 250, "group": "gold"}}    |  2|
|kj-test|       1|{"data": {"score": 330, "group": "platinum"}}|  3|
+-------+--------+---------------------------------------------+---+

decrypt_udf UDF 函数 sn-p：

@udf(returnType=StringType())
def decrypt_udf(encrypted_string: str):
   ...
   ...
   return decrypted_json_str

当我将 spark 数据帧写入 S3 存储桶时，如下所示

df_new.write.mode('overwrite').json(path=s3outputpath)

生成的文件内容如下，这里requestBody 的值写为String，因此用双引号和转义内部双引号。

{"name":"kj-test","version":"1","requestBody":"{\"data\": {\"score\": 130, \"group\": \"silver\"}}","id":"1"}
{"name":"kj-test","version":"2","requestBody":"{\"data\": {\"score\": 250, \"group\": \"gold\"}}","id":"1"}
{"name":"kj-test","version":"3","requestBody":"{\"data\": {\"score\": 330, \"group\": \"platinum\"}}","id":"1"}

但是，我希望 requestBody 的值可以写成如下的 json。

{"name":"kj-test","version":"1","requestBody":{"data": {"score": 130, "group": "silver"}},"id":"1"}

我知道我已将 requestBody 的类型指定为架构 StructField('requestBody', StringType(), True) 中的字符串，因此我以这种方式看到输出。我怎样才能达到我期望的输出？没有JsonType这样的类型

---

编辑：

请注意，我的 requestBody 架构不会总是这样 {"data": {"score": 130, "group": "silver"}}。对于给定的运行，它是固定的，但另一次运行可能具有完全不同的架构。

本质上，需要一种从 json 字符串推断模式的方法。找到一些可能有用的 SO 帖子，将尝试这些：

https://stackoverflow.com/a/45880574/948268
Spark from_json with dynamic schema

Answer 1

试试下面的代码。 （我没有测试过）

使用from_json 函数将requestBody json 字符串转换为结构体。

schema = StructType(
    [
        StructField('name', StringType(), True),
        StructField('version', StringType(), True),
        StructField('requestBody', StringType(), True),
        StructField('id', StringType(), True),
    ]
)

为requestBody准备架构

requestSchema=StructType(
    [
        StructField('data', StructType([StructField('group',StringType(),True),StructField('score',LongType(),True)])),
    ]
)

df_new = df.withColumn('value', from_json('value', schema)) \
    .where(col('value.version') == '1') \
    .select(col('value.*'))\
    .withColumn()
    .na.drop() \
    .withColumn('requestBody', from_json('requestBody',requestSchema))

df_new.write.mode('overwrite').json(path=s3outputpath)

Answer 2

在您的 udf 中，添加以下将 python 对象转换为 JSON 字符串的方法：

import json   
@udf(returnType=StringType())
def decrypt_udf(encrypted_string: str):
   ...
   ...
   return json.dumps(decrypted_json_str)

Answer 3

较新的解决方案（我认为这是一个更好的解决方案）

我们最终使用的另一个聪明的解决方案。在此，我们定义了一个 udf get_combined_json，它结合了给定 Row 的所有列，然后返回一个 json 字符串。导致我们的最终数据框只有一列，这样我们就可以将数据框写为文本文件，这样整个 json 字符串就可以按原样写入而没有任何转义。以下是代码sn-p：

df_new = df.withColumn('value', from_json('value', schema)) \
    .where(col('value.version') == '1') \
    .select(col('value.*'))\
    .na.drop() \
    .withColumn('requestBody', decrypt_udf(col('requestBody')))

df_new.withColumn('combinedColumns', get_combined_json(struct([df_new[x] for x in df_new.columns]))) \
    .select(col('combinedColumns'))\
    .write.mode('overwrite').text(path=output_s3_bucket_path)

...

@udf(returnType=StringType())
def get_combined_json(row: Row):
    return json.dumps({"requestBody": json.loads(row.requestBody),
                       "name": row.name,
                       "version": row.version,
                       "id": row.id})

---

---

较旧的解决方案

以下是我们从requestBody json 字符串派生/推断架构的方式：

request_body_schema = spark_session.read.json(df_new.rdd.map(lambda r: r.requestBody)).schema

然后使用架构更新数据框。这是有效的最终代码：

df_new = df.withColumn('value', from_json('value', schema)) \
    .where(col('value.version') == '1') \
    .select(col('value.*'))\
    .na.drop() \
    .withColumn('requestBody', decrypt_udf(col('requestBody')))

request_body_schema = spark_session.read.json(df_new.rdd.map(lambda r: r.requestBody)).schema

df_new = df_new.withColumn('requestBody', from_json(col('requestBody'), request_body_schema))

df_new.write.mode('overwrite').json(path=output_s3_bucket_path)

以下是写入 S3 存储桶的输出格式：

{"name":"kj-test","version":"1","requestBody":{"data": {"score": 130, "group": "silver"}},"id":"1"}