Skip to main content
 首页 » 编程设计

python之PySpark 在嵌套数组中反转 StringIndexer

2025年02月15日32开发

我正在使用 PySpark 通过 ALS 进行协同过滤。我的原始用户和项目 ID 是字符串,所以我使用了 StringIndexer将它们转换为数字索引(PySpark 的 ALS 模型要求我们这样做)。

在我拟合模型后,我可以像这样为每个用户获得前 3 个推荐:

recs = ( 
    model 
    .recommendForAllUsers(3) 
) 

recs数据框看起来像这样:

+-----------+--------------------+ 
|userIdIndex|     recommendations| 
+-----------+--------------------+ 
|       1580|[[10096,3.6725707...| 
|       4900|[[10096,3.0137873...| 
|       5300|[[10096,2.7274625...| 
|       6620|[[10096,2.4493625...| 
|       7240|[[10096,2.4928937...| 
+-----------+--------------------+ 
only showing top 5 rows 
 
root 
 |-- userIdIndex: integer (nullable = false) 
 |-- recommendations: array (nullable = true) 
 |    |-- element: struct (containsNull = true) 
 |    |    |-- productIdIndex: integer (nullable = true) 
 |    |    |-- rating: float (nullable = true) 

我想用这个数据框创建一个巨大的 JSOM 转储,我可以这样:

( 
    recs 
    .toJSON() 
    .saveAsTextFile("name_i_must_hide.recs") 
) 

这些 json 的示例是:

{ 
  "userIdIndex": 1580, 
  "recommendations": [ 
    { 
      "productIdIndex": 10096, 
      "rating": 3.6725707 
    }, 
    { 
      "productIdIndex": 10141, 
      "rating": 3.61542 
    }, 
    { 
      "productIdIndex": 11591, 
      "rating": 3.536216 
    } 
  ] 
} 

userIdIndex和 productIdIndex key 归因于 StringIndexer转型。

如何取回这些列的原始值?我怀疑我必须使用 IndexToString变压器,但我不太清楚如何,因为数据嵌套在 recs 内的数组中数据框。

我尝试使用 Pipeline求值器 ( stages=[StringIndexer, ALS, IndexToString] ),但看起来这个求值器不支持这些索引器。

干杯!

请您参考如下方法:

在这两种情况下,您都需要访问标签列表。这可以使用 StringIndexerModel

访问
user_indexer_model = ...  # type: StringIndexerModel 
user_labels = user_indexer_model.labels 
 
product_indexer_model = ...  # type: StringIndexerModel 
product_labels = product_indexer_model.labels 

或列元数据。

对于userIdIndex,你可以只应用IndexToString:

from pyspark.ml.feature import IndexToString 
 
user_id_to_label = IndexToString( 
    inputCol="userIdIndex", outputCol="userId", labels=user_labels) 
user_id_to_label.transform(recs) 

对于建议,您需要 udf 或像这样的表达式:

from pyspark.sql.functions import array, col, lit, struct 
 
n = 3  # Same as numItems 
 
product_labels_ = array(*[lit(x) for x in product_labels]) 
recommendations = array(*[struct( 
    product_labels_[col("recommendations")[i]["productIdIndex"]].alias("productId"), 
    col("recommendations")[i]["rating"].alias("rating") 
) for i in range(n)]) 
 
recs.withColumn("recommendations", recommendations)