Java-Spark：如何在循环中迭代时获取 Dataset<Row> 列的值并在 when().otherwise() 中使用它？

Question

我有一个 Dataset<Row>，其列的值为 "null"（空书面文本）。
我正在尝试将 "null" 文本替换为文本：\N.
为此，我使用了一种逻辑，我将添加一个新列，其名称附加“_nulled”，例如列 abc 变为 abc_nulled 并且此新列的值为 "\N"如果当前值为文本 null，否则该值保持不变。
为此，我使用了 withColumn(<new name>, when(col.equalTo("null"), "\N").otherwise(<existing_value>))。我如何获得这个 <existing_value>.
当我通过 otherwise(ds.col(col_nm)) 时它不起作用，可能是因为它在 otherwise() 中期望 String 并找到 Column.

我应该如何解决这个问题？这是代码：

ArrayList<String> newCols = new ArrayList<String>();
List<String> reqColListCopy = Arrays.asList(reqCols);
Dataset<Row> testingDS = DS.selectExpr(JavaConverters.asScalaIteratorConverter(reqColListCopy.iterator()).asScala().toSeq())

//Creating newCols (ArrayList so that I can add/remove column names.
Iterator itrTmp2 = reqColListCopy.iterator();
while(itrTmp2.hasNext()){
    newCols.add((String)itrTmp2.next());
}

//Creating a List reference for newCols ArrayList. This will be used to get Seq(<columns>).
List<String> newColsList = newCols;

Iterator colListItr = reqColListCopy.iterator();
while(colListItr.hasNext())
{
    String col = colListItr.next().toString();
    testingDS = testingDS.selectExpr(convertListToSeq(newColsList))
            .withColumn(col+"_nulled",  functions.when(testingDS.col(col).equalTo("null"), functions.lit("\N")).otherwise(testingDS.col(col))) //'otherwise' needs a string parameter
            .drop(testingDS.col(col));

    newCols.add(col+"_nulled");
    newCols.remove(col);
    newColsList = newCols;
}
Dataset<Row> testingDS = DS.selectExpr(JavaConverters.asScalaIteratorConverter(newColsList.iterator()).asScala().toSeq())

testingDS.show(false);

Answer 1

我通过在 lit():

中传递列来解决它

.withColumn(col+"_nulled",  functions.when(testingDS.col(col).equalTo("null"), functions.lit("\N")).otherwise(functions.lit(testingDS.col(col)))) //'otherwise' needs a string parameter

Java-Spark：如何在循环中迭代时获取 Dataset<Row> 列的值并在 when().otherwise() 中使用它？

Java-Spark: how to get a Dataset<Row> column's value when iterating in a loop and use it in when().otherwise()?

java

apache-spark

apache-spark-sql

apache-spark-dataset