在Palantir Foundry的PySpark中高效清理JSON的HTML实体

在Palantir Foundry中使用PySpark分析30MB的上传JSON文件，该JSON包含4个元素，其中Data字段是一个约60列、20000行的表格。表格中部分字符串列包含HTML实体（其余为数值/布尔型），需要将这些实体替换为对应UTF字符。

当前代码直接用spark_session.read.json()生成带自动推断Schema的DataFrame，但不确定如何在解析前加入html.unescape()处理，同时避免所有字段变为StringType()；也不想手动编写嵌套Schema。另外想了解：是先处理整个JSON文本再解析高效，还是利用Spark并行性逐行逐列处理更高效？

当前读取JSON的代码：

from transforms.verbs.dataframes import sanitize_schema_for_parquet
from transforms.api import transform, Input, Output, Check

@transform(
    parsed_output=Output("out_path"),
    raw_file_input=Input("in_path")
)
def read_json(ctx, parsed_output, raw_file_input):
    filesystem = raw_file_input.filesystem()
    hadoop_path = filesystem.hadoop_path
    paths = [f"{hadoop_path}/{f.path}" for f in filesystem.ls()]

    df = ctx.spark_session.read.json(paths)

    parsed_output.write_dataframe(sanitize_schema_for_parquet(df))

JSON输入示例：

{
    "Data": [
        {"Id": 1, "Lots": "more", "data": "of", "different": "types", "Flag1": true, "RefNumber": 17},
        {"Id": 2, "Lots": "of the string", "data": "includes entities like ≤ in", "different": "places", "Flag1": false, "RefNumber": 17781}
    ],
    "TotalRows":2,
    "Warnings":null,
    "Errors":null
}

期望输出（仅替换HTML实体，保留原数据类型）：

Id | Lots            | data                          | different | Flag1 | RefNumber
---+-----------------+-------------------------------+-----------+-------+-----------
 1 | "more"          | "of"                          | "types"   | true  |        17
 2 | "of the string" | "includes entities like ≤ in" | "places"  | false |     17781

方案一：解析前批量处理JSON文本（保留自动Schema推断） #

先将JSON文件读为文本RDD，对每一行文本执行html.unescape()，再将处理后的RDD解析为DataFrame。这种方式能让Spark自动推断原始Schema，同时提前处理所有HTML实体。

修改后的代码：

from html import unescape
from transforms.verbs.dataframes import sanitize_schema_for_parquet
from transforms.api import transform, Input, Output

@transform(
    parsed_output=Output("out_path"),
    raw_file_input=Input("in_path")
)
def read_json(ctx, parsed_output, raw_file_input):
    filesystem = raw_file_input.filesystem()
    hadoop_path = filesystem.hadoop_path
    paths = [f"{hadoop_path}/{f.path}" for f in filesystem.ls()]

    # 读取原始文本RDD并处理HTML实体
    raw_text_rdd = ctx.spark_session.sparkContext.textFile(",".join(paths))
    unescaped_rdd = raw_text_rdd.map(lambda line: unescape(line))

    # 解析处理后的文本为DataFrame，自动推断Schema
    df = ctx.spark_session.read.json(unescaped_rdd)

    parsed_output.write_dataframe(sanitize_schema_for_parquet(df))

注意：如果JSON文件是单行压缩格式（大文件常见格式），textFile会将整个文件读为单一行，此时map操作依然能高效处理；30MB的文件在Spark中处理完全无性能瓶颈。

方案二：利用Spark并行性高效处理DataFrame字符串列（比普通UDF更高效） #

如果担心先处理整个文本可能出现的编码问题，可采用Pandas UDF（向量化UDF）处理DataFrame中的所有字符串列，这种方式利用Spark的并行计算能力，效率远高于普通Python UDF，且无需手动指定Schema。

代码示例：

import pandas as pd
from html import unescape
from pyspark.sql.functions import pandas_udf, col
from pyspark.sql.types import StringType
from transforms.verbs.dataframes import sanitize_schema_for_parquet
from transforms.api import transform, Input, Output

@transform(
    parsed_output=Output("out_path"),
    raw_file_input=Input("in_path")
)
def read_json(ctx, parsed_output, raw_file_input):
    filesystem = raw_file_input.filesystem()
    hadoop_path = filesystem.hadoop_path
    paths = [f"{hadoop_path}/{f.path}" for f in filesystem.ls()]

    # 读取原始JSON为DataFrame，自动推断Schema
    df = ctx.spark_session.read.json(paths)

    # 定义Pandas UDF用于批量处理字符串
    @pandas_udf(StringType())
    def unescape_udf(s: pd.Series) -> pd.Series:
        return s.apply(unescape)

    # 自动识别所有字符串列并应用处理
    for field in df.schema.fields:
        if field.dataType.typeName() == "string":
            df = df.withColumn(field.name, unescape_udf(col(field.name)))

    # （可选）将Data数组展开为行
    # from pyspark.sql.functions import explode
    # df = df.select(explode(col("Data")).alias("data_row")).select("data_row.*")

    parsed_output.write_dataframe(sanitize_schema_for_parquet(df))

效率对比 #

• 30MB属于小型数据集，两种方案效率差异不大。方案一优势是一次性处理所有文本，减少后续DataFrame操作开销；方案二更灵活，可针对性处理特定列，且Pandas UDF的批量处理能力在数据量更大时优势更明显。
• 普通Python UDF因序列化开销大，效率较低，不推荐使用；Pandas UDF是批量处理数据，性能接近原生Spark函数。

内容的提问来源于stack exchange，提问作者JRI