如何基于含NULL值的多列组合生成产品分类列unique_cat

问题背景 #

现有MySQL 5.7的products表，结构及测试数据如下：

CREATE TABLE products (
    product_id int(11) NOT NULL AUTO_INCREMENT,
    node_0 varchar(400) DEFAULT NULL,
    node_1 varchar(400) DEFAULT NULL,
    node_2 varchar(400) DEFAULT NULL,
    node_3 varchar(400) DEFAULT NULL,
    node_4 varchar(255) DEFAULT NULL,
    PRIMARY KEY (product_id)
);

INSERT INTO products (node_0, node_1, node_2, node_3, node_4) VALUES
('a_0', NULL, NULL, NULL, 'a_1'),
('a_0', NULL, NULL, NULL, 'a_1'),
('a_2', 'a_1', NULL, NULL, 'a_1'),
('a_0', NULL, NULL, 'a_3', 'a_2'),
('a_3', NULL, NULL, 'a_0', 'a_2'),
('a_0', NULL, NULL, NULL, 'a_2'),
('a_2', 'a_1', NULL, NULL, 'a_1')

其中node_0和node_4保证非NULL，其余字段可能为NULL。需求是基于node_0至node_4的唯一组合生成数值型列unique_cat，预期输出如下：

基础可行方案（仅处理node_0和node_4） #

当仅基于node_0和node_4生成唯一ID时，以下PySpark代码可正常工作：

# 创建node_0与node_4的唯一组合并分配ID
unique_cat_df = df \
      .select("node_0", "node_4") \
      .distinct() \
      .withColumn("unique_cat", monotonically_increasing_id())

# 将唯一ID关联回原DataFrame
df_with_cat_ids = df.join(
      unique_cat_df,
      on=["node_0", "node_4"],
      how="left"
)

扩展到全字段时的失效情况 #

当需要包含所有node_0至node_4字段生成唯一组合时，最初尝试的代码出现失效：

placeholder = "___NULL___"
df = df_0 \
    .withColumn("node_2", F.when(col("node_2").isNull(), placeholder).otherwise(col("node_2"))) \
    .withColumn("node_3", F.when(col("node_3").isNull(), placeholder).otherwise(col("node_3")))

# 选择全字段创建唯一组合并分配ID
unique_combinations_df = df \
    .select("node_0", "node_1", "node_2", "node_3", "node_4") \
    .distinct() \
    .withColumn("unique_cat", monotonically_increasing_id())

# 关联回原DataFrame（原代码存在字段名笔误：current_node、root_node应为node_0、node_4）
df_with_ids = lastest_data_df_2.join(
    unique_combinations_df,
    on=["node_0", "node_1", "node_2", "node_3", "node_4"],
    how="left"
)

问题排查与修正 #

经排查，失效核心原因是遗漏了对node_1字段的NULL值替换。PySpark中NULL值在distinct或关联操作时无法被识别为相同值，因此需要对所有可能为NULL的node字段统一替换占位符。

修正后的完整代码如下（补充连续ID生成方式，更贴合预期输出）：

from pyspark.sql import functions as F

placeholder = "___NULL___"
# 对所有可能为NULL的node字段替换占位符
df_processed = df_0 \
    .withColumn("node_1", F.when(F.col("node_1").isNull(), placeholder).otherwise(F.col("node_1"))) \
    .withColumn("node_2", F.when(F.col("node_2").isNull(), placeholder).otherwise(F.col("node_2"))) \
    .withColumn("node_3", F.when(F.col("node_3").isNull(), placeholder).otherwise(F.col("node_3")))

# 创建全字段唯一组合并生成从0开始的连续ID
unique_combinations_df = df_processed \
    .select("node_0", "node_1", "node_2", "node_3", "node_4") \
    .distinct() \
    .withColumn("unique_cat", F.row_number().over(F.orderBy("node_0", "node_4")) - 1)

# 关联回原DataFrame并还原NULL值
df_with_ids = df_processed.join(
    unique_combinations_df,
    on=["node_0", "node_1", "node_2", "node_3", "node_4"],
    how="left"
) \
.withColumn("node_1", F.when(F.col("node_1") == placeholder, F.lit(None)).otherwise(F.col("node_1"))) \
.withColumn("node_2", F.when(F.col("node_2") == placeholder, F.lit(None)).otherwise(F.col("node_2"))) \
.withColumn("node_3", F.when(F.col("node_3") == placeholder, F.lit(None)).otherwise(F.col("node_3")))

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