java - 使用 Java API 对数据集中所有列进行数据操作

Question

读取Dataset中的csv文件后，想要使用Java API从String类型数据中删除空格。

Apache Spark 2.0.0

Dataset<Row> dataset = sparkSession.read().format("csv").option("header", "true").load("/pathToCsv/data.csv");
Dataset<String> dataset2 = dataset.map(new MapFunction<Row,String>() {

    @Override
    public String call(Row value) throws Exception {

        return value.getString(0).replace(" ", ""); 
        // But this will remove space from only first column
    }
}, Encoders.STRING());

通过使用MapFunction，无法删除所有列中的空格。

但是在Scala中，通过在spark-shell中使用以下方式能够执行所需的操作。

val ds = spark.read.format("csv").option("header", "true").load("/pathToCsv/data.csv")
val opds = ds.select(ds.columns.map(c => regexp_replace(col(c), " ", "").alias(c)): _*)

数据集opds有没有空格的数据。想在Java中实现同样的目标。但在Java API中columns方法返回String[]并且无法对Dataset执行函数式编程。

输入数据

+----------------+----------+-----+---+---+
|               x|         y|    z|  a|  b|
+----------------+----------+-----+---+---+
|     Hello World|John Smith|There|  1|2.3|
|Welcome to world| Bob Alice|Where|  5|3.6|
+----------------+----------+-----+---+---+

预期输出数据

+--------------+---------+-----+---+---+
|             x|        y|    z|  a|  b|
+--------------+---------+-----+---+---+
|    HelloWorld|JohnSmith|There|  1|2.3|
|Welcometoworld| BobAlice|Where|  5|3.6|
+--------------+---------+-----+---+---+

Answer 1

尝试:

for (String col: dataset.columns) {
  dataset = dataset.withColumn(col, regexp_replace(dataset.col(col), " ", ""));
}