我正在尝试将实际包含 double 值的 csv 字符串转换为 spark-ml 兼容数据集。因为我事先不知道预期的功能数量,所以我决定使用一个辅助类“Instance”,它已经包含了分类器使用的正确数据类型,并且在其他一些情况下已经按预期工作:
public class Instance implements Serializable {
/**
*
*/
private static final long serialVersionUID = 6091606543088855593L;
private Vector indexedFeatures;
private double indexedLabel;
...getters and setters for both fields...
}
我得到意外行为的部分是:
Encoder<Instance> encoder = Encoders.bean(Instance.class);
System.out.println("encoder.schema()");
encoder.schema().printTreeString();
Dataset<Instance> dfInstance = df.select("value").as(Encoders.STRING())
.flatMap(s -> {
String[] splitted = s.split(",");
int length = splitted.length;
double[] features = new double[length-1];
for (int i=0; i<length-1; i++) {
features[i] = Double.parseDouble(splitted[i]);
}
if (length < 2) {
return Collections.emptyIterator();
} else {
return Collections.singleton(new Instance(
Vectors.dense(features),
Double.parseDouble(splitted[length-1])
)).iterator();
}
}, encoder);
System.out.println("dfInstance");
dfInstance.printSchema();
dfInstance.show(5);
然后我在控制台上得到以下输出:
encoder.schema()
root
|-- indexedFeatures: vector (nullable = true)
|-- indexedLabel: double (nullable = false)
dfInstance
root
|-- indexedFeatures: struct (nullable = true)
|-- indexedLabel: double (nullable = true)
+---------------+------------+
|indexedFeatures|indexedLabel|
+---------------+------------+
| []| 0.0|
| []| 0.0|
| []| 1.0|
| []| 0.0|
| []| 1.0|
+---------------+------------+
only showing top 5 rows
编码器模式正确地将 indexedFeatures 行数据类型显示为 vector 。但是当我应用编码器并进行转换时,它会给我一行结构类型,不包含任何真实对象。
我想了解,为什么 Spark 为我提供结构类型而不是正确的 vector 类型。
最佳答案
实际上,我的回答并不能解释为什么你会得到一个结构类型。但基于previous question ,我或许可以提供解决方法。
原始输入用DataFrameReader's csv function解析,然后又是一个VectorAssembler使用:
Dataset<Row> csv = spark.read().option("inferSchema", "true")
.csv(inputDf.select("value").as(Encoders.STRING()));
String[] fieldNames = csv.schema().fieldNames();
VectorAssembler assembler = new VectorAssembler().setInputCols(
Arrays.copyOfRange(fieldNames, 0, fieldNames.length-1))
.setOutputCol("indexedFeatures");
Dataset<Row> result = assembler.transform(csv)
.withColumn("indexedLabel", functions.col(fieldNames[fieldNames.length-1]))
.select("indexedFeatures", "indexedLabel");
关于java - 在 Apache Spark Dataset<Row> 上应用 flatMap 操作时出现意外的编码器行为,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/49676184/