如何对PySpark DataFrame执行行归一化及按userId访问行

我来帮你搞定这个PySpark的行归一化和行访问问题，分两部分一步步来：

一、实现行归一化 #

你的需求是把每行的电影类型计数转换成0-1之间的比例（基于该行的总计数），步骤如下：

1. 计算每行的总计数 #

首先我们需要把除userId外的所有类型列求和，得到每行的总观影计数，同时把总和转为浮点型，避免后续计算出现整数除法问题。

2. 对每个类型列做归一化处理 #

遍历所有类型列，用列值除以该行的总计数；如果总计数为0（用户没有任何观影记录），直接返回0.0，避免除以0的错误。

完整代码示例 #

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

# 假设你的原始DataFrame名为user_preferences
# 提取所有类型列（排除userId）
genre_columns = [col for col in user_preferences.columns if col != "userId"]

# 第一步：计算每行的总计数
df_with_total = user_preferences.withColumn(
    "total",
    F.sum(*genre_columns).cast(DoubleType())
)

# 第二步：对每个类型列执行归一化
normalized_df = df_with_total.select(
    "userId",
    *[
        # 处理总和为0的情况，避免除以0错误
        F.when(F.col("total") == 0, 0.0)
        .otherwise(F.col(col) / F.col("total"))
        .alias(col)
        for col in genre_columns
    ]
).drop("total")  # 移除临时的total列

转换成数组格式（匹配示例输出） #

如果需要把归一化后的类型值整合成一个数组（和你给出的示例格式一致），可以用array函数：

# 将归一化后的类型列打包成数组
normalized_array_df = normalized_df.select(
    "userId",
    F.array(*genre_columns).alias("normalized_preferences")
)

执行后，normalized_preferences列就是你示例中的数组形式，每个元素对应一个类型的归一化值。

二、通过userId访问指定行 #

有几种常用的方式可以根据userId快速定位到目标行，适合不同场景：

1. Spark原生Filter（推荐大数据量） #

直接用filter或where方法筛选，返回的是Spark Row对象：

# 获取userId=65的归一化数据
target_user_row = normalized_df.filter(F.col("userId") == 65).first()

# 如果是数组格式的DataFrame，提取数组值
target_user_preferences = normalized_array_df.filter(F.col("userId") == 65).select("normalized_preferences").first()[0]

2. 注册临时视图用SQL查询 #

如果你更熟悉SQL语法，可以把DataFrame注册成临时视图查询：

# 注册临时视图
normalized_df.createOrReplaceTempView("user_normalized_prefs")

# 用SQL查询指定userId
target_user_sql = spark.sql("SELECT * FROM user_normalized_prefs WHERE userId = 65").first()

3. 转成Pandas DataFrame（适合小数据量） #

如果你的数据量不大，转成Pandas DataFrame后可以用索引快速访问：

# 转成Pandas DataFrame并设置userId为索引
pandas_prefs = normalized_df.toPandas()
pandas_prefs.set_index("userId", inplace=True)

# 直接通过userId索引访问
target_user_pandas = pandas_prefs.loc[65]

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