如何在DataFrame中使用Jinja模板生成Score_HTML列

我有如下PySpark DataFrame：

FirstName  LastName  Score
Hello      World     [('Math', 90), ('Eng', 80)]
ABC        XYZ       [('Math', 90)]

其中Score列是Spark中的struct数组类型，实际存储格式为：

[Row(sub='Math', score=90), Row(sub='Eng', score=80)]

希望将Score列转换为Score_HTML列，预期输出如下：

FirstName  LastName  Score_HTML
Hello      World     "<b>FullName:</b>Hello World <br><br> <table border="1"><tr><td>Sub</td><td>Score</td></tr><tr><td>Math</td><td>90</td></tr><tr><td>Eng</td><td>80</td></tr></table>"
ABC        XYZ       "<b>FullName:</b>ABC XYZ <br><br> <table border="1"><tr><td>Sub</td><td>Score</td></tr><tr><td>Math</td><td>90</td></tr></table>"

我尝试将Spark DataFrame转为Pandas DataFrame后使用Jinja模板实现，但代码存在问题，需要修正模板并完成需求。

方案一：Pandas + Jinja2 实现 #

步骤1：定义正确的Jinja模板 #

原模板的循环逻辑错误（所有单元格会被放在同一行），修正后的模板会为每个科目生成独立的表格行：

import jinja2

# 初始化Jinja环境
env = jinja2.Environment()
template_str = """
<b>FullName:</b>{{ first_name }} {{ last_name }} <br><br>
<table border="1">
    <tr><td>Sub</td><td>Score</td></tr>
    {% for item in score_list %}
    <tr><td>{{ item.sub }}</td><td>{{ item.score }}</td></tr>
    {% endfor %}
</table>
"""
template = env.from_string(template_str)

步骤2：PySpark转Pandas后逐行渲染 #

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, ArrayType

# 初始化Spark会话（已有则跳过）
spark = SparkSession.builder.appName("HTMLGenerator").getOrCreate()

# 构建示例Spark DataFrame（替换为你的实际数据）
data = [
    ("Hello", "World", [("Math", 90), ("Eng", 80)]),
    ("ABC", "XYZ", [("Math", 90)])
]
schema = StructType([
    StructField("FirstName", StringType()),
    StructField("LastName", StringType()),
    StructField("Score", ArrayType(StructType([
        StructField("sub", StringType()),
        StructField("score", IntegerType())
    ])))
])
spark_df = spark.createDataFrame(data, schema=schema)

# 转换为Pandas DataFrame
pandas_df = spark_df.toPandas()

# 定义逐行渲染函数
def generate_html(row):
    return template.render(
        first_name=row["FirstName"],
        last_name=row["LastName"],
        score_list=row["Score"]
    )

# 生成Score_HTML列
pandas_df["Score_HTML"] = pandas_df.apply(generate_html, axis=1)

# 可选：转回Spark DataFrame
spark_result_df = spark.createDataFrame(pandas_df)
spark_result_df.show(truncate=False)

方案二：纯Spark UDF 实现（大数据场景推荐） #

如果数据量较大，转Pandas可能存在内存限制，直接用Spark UDF处理更高效：

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

def score_to_html(first_name, last_name, score_list):
    # 拼接FullName部分
    html = f"<b>FullName:</b>{first_name} {last_name} <br><br>"
    # 拼接表格头部
    html += '<table border="1"><tr><td>Sub</td><td>Score</td></tr>'
    # 循环拼接每个科目行
    for item in score_list:
        html += f'<tr><td>{item.sub}</td><td>{item.score}</td></tr>'
    # 闭合表格标签
    html += '</table>'
    # 如需带外层引号，可改为 return f'"{html}"'
    return html

# 注册UDF
html_udf = udf(score_to_html, StringType())

# 生成Score_HTML列
spark_result_df = spark_df.withColumn(
    "Score_HTML",
    html_udf(spark_df["FirstName"], spark_df["LastName"], spark_df["Score"])
)

spark_result_df.show(truncate=False)

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