PySpark 在嵌套数组中反转 StringIndexer

我正在使用 PySpark 使用 ALS 进行协作过滤。我原来的用户和项目 ID 是字符串，所以我使用StringIndexer将它们转换为数字索引（PySpark 的 ALS 模型要求我们这样做）。

安装模型后，我可以获得每个用户的前 3 个推荐，如下所示：

recs = (
    model
    .recommendForAllUsers(3)
)

The 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)

我想用这个数据帧创建一个巨大的 JSON 转储，我可以这样：

(
    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
    }
  ]
}

The userIdIndex and productIdIndex键是由于StringIndexer转型。

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

我尝试使用Pipeline评估者（stages=[StringIndexer, ALS, IndexToString]）但看起来这个评估器不支持这些索引器。

Cheers!

在这两种情况下，您都需要访问标签列表。可以使用以下任一方式访问此内容：StringIndexerModel

user_indexer_model = ...  # type: StringIndexerModel
user_labels = user_indexer_model.labels

product_indexer_model = ...  # type: StringIndexerModel
product_labels = product_indexer_model.labels

或列元数据。

For 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)