如何在Java中分解特征向量?

Question

我有一个数据帧，如下所示：

+---------------+--------------------+
|IndexedArtistID|     recommendations|
+---------------+--------------------+
|           1580|[[919, 0.00249262...|
|           4900|[[41749, 7.143963...|
|           5300|[[0, 2.0147272E-4...|
|           6620|[[208780, 9.81092...|
+---------------+--------------------+

我想拆分建议列，以便有一个数据帧，如下所示：

+---------------+--------------------+
|IndexedArtistID|     recommendations|
+---------------+--------------------+
|           1580|919                 |
|           1580|0.00249262          |
|           4900|41749               |
|           4900|7.143963            |
|           5300|0                   |
|           5300|2.0147272E-4        |
|           6620|208780              |
|           6620|9.81092             |
+---------------+--------------------+

所以基本上，我想将特征向量拆分为列，然后将这些列合并为单个列。合并部分在：如何使用 Java 在 Spark DataFrame 中将单行拆分为多行。 现在，如何使用java进行拆分部分？ 对于scala，这里解释了：Spark Scala：如何将Dataframe[vector]转换为DataFrame[f1：Double，...，fn：Double(]，但我无法找到一种方法在java中以与链接中给出的相同的方式进行。

数据帧的架构如下所示，IndexedUserID 的值将纳入新创建的建议列：

root
|-- IndexedArtistID: integer (nullable = false)
|-- recommendations: array (nullable = true)
|    |-- element: struct (containsNull = true)
|    |    |-- IndexedUserID: integer (nullable = true)
|    |    |-- rating: float (nullable = true)

Answer 1

我试图找到这个问题的解决方案，我必须说，在 python 和 scala for Spark 中，人们面临的问题有很多内容可用，但在 Java 中可用的内容很少。 因此，解决方案如下：

List<ElementStruct> structElements = dataFrameWithFeatures.javaRDD().map(row -> {
int artistId = row.getInt(0);
List<Object> recommendations = row.getList(1);
return new ElementStruct(artistId, recommendations);
}).collect();
List<Recommendation> recommendations = new ArrayList<>();
for (ElementStruct element : structElements) {
List<Object> features = element.getFeatures();
int artistId = element.getArtistId();
for (int i = 0; i < features.size(); i++) {
Object o = ((GenericRowWithSchema) features.get(i)).get(0);
recommendations.add(new Recommendation(artistId, (int) o));
}
}
SparkSession sparkSession = SessionCreator.getOrCreateSparkSession();
Dataset<Row> decomposedDataframe = sparkSession.createDataFrame(recommendations, Recommendation.class);

元素结构类

import java.io.Serializable;
import java.util.List;
public class ElementStruct implements Serializable {
private int artistId;
private List<Object> features;
public ElementStruct(int artistId, List<Object> features) {
this.artistId = artistId;
this.features = features;
}
public int getArtistId() {
return artistId;
}
public void setArtistId(int artistId) {
this.artistId = artistId;
}
public List<Object> getFeatures() {
return features;
}
public void setFeatures(List<Object> features) {
this.features = features;
}
}

推荐类

import java.io.Serializable;
public class Recommendation implements Serializable {
private int artistId;
private int userId;
public Recommendation(int artistId, int userId){
this.artistId = artistId;
this.userId = userId;
}
public int getArtistId() {
return artistId;
}
public void setArtistId(int artistId) {
this.artistId = artistId;
}
public int getUserId() {
return userId;
}
public void setUserId(int userId) {
this.userId = userId;
}
}

解释： 1. 对于数据帧中的每一行，将艺术家和要素作为列表获取，以便于进一步处理。将这些艺术家和功能列表存储为 java 对象(在本例中为 Element struct(。

2. 对于功能列表中的每个艺术家和元素，创建一个新的对象列表(在本例中为"建议"(，并将每个对象存储在该列表中。

3. 最后，从第 2 步中获得的对象列表中创建一个数据帧。

结果：

root
|-- artistId: integer (nullable = false)
|-- userId: integer (nullable = false)
+---------------+----------------+
|       artistId|          userId|
+---------------+----------------+
|           1580|919             |
|           1580|0.00249262      |
|           4900|41749           |
|           4900|7.143963        |
|           5300|0               |
|           5300|2.0147272E-4    |
|           6620|208780          |
|           6620|9.81092         |
+---------------+----------------+