Java-Spark：Dataset<Row>文本"null"转"\N"时when参数问题解决

First, let's break down the issues in your current code:

• Overcomplicating column management with selectExpr and manual list updates, which can lead to incorrect column references
• Using testingDS.col(col) which refers to the dataset state before your selectExpr call, potentially causing mismatches
• A duplicate Dataset<Row> testingDS declaration at the end (invalid Java syntax)

Here's a simplified, working approach that achieves your goal:

Step-by-Step Explanation #

We'll create new columns with the _nulled suffix, replace the string "null" with "\N", and either keep the new columns in the original order or drop old columns incrementally.

Approach 1: Add New Columns, Then Select Only the New Ones (Preserves Original Order) #

This method is clean and ensures your final dataset has columns in the same order as the original:

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.functions;
import java.util.ArrayList;
import java.util.List;

// Start by selecting the required columns
List<String> reqColListCopy = Arrays.asList(reqCols);
Dataset<Row> testingDS = DS.selectExpr(JavaConverters.asScalaIteratorConverter(reqColListCopy.iterator()).asScala().toSeq());

List<String> newColumnNames = new ArrayList<>();

// Iterate over each column to create the "_nulled" version
for (String colName : reqColListCopy) {
    String newColName = colName + "_nulled";
    newColumnNames.add(newColName);
    
    // Create the new column: replace "null" with "\N", else keep original value
    testingDS = testingDS.withColumn(
        newColName,
        functions.when(functions.col(colName).equalTo("null"), functions.lit("\\N"))
                 .otherwise(functions.col(colName))
    );
}

// Select only the new columns (in original order)
testingDS = testingDS.select(newColumnNames.stream().map(functions::col).toArray(Column[]::new));

testingDS.show(false);

Approach 2: Incrementally Replace Columns #

If you prefer to drop original columns as you go (instead of keeping them temporarily):

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.functions;

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

for (String colName : reqColListCopy) {
    String newColName = colName + "_nulled";
    
    testingDS = testingDS.withColumn(
        newColName,
        functions.when(functions.col(colName).equalTo("null"), functions.lit("\\N"))
                 .otherwise(functions.col(colName))
    )
    .drop(colName); // Remove the original column after creating the new one
}

testingDS.show(false);

Key Fixes & Improvements #

• Using functions.col(colName): This references the column from the current state of testingDS, avoiding the confusion of using the pre-selectExpr dataset reference.
• Simplified Column Management: No need to manually track and update column lists with add/remove calls—Spark handles preserving other columns automatically with withColumn.
• Correct otherwise Usage: otherwise expects a Column type (not a String), which your original code had right, but the column reference was potentially incorrect.

Content of the question来源于stack exchange,提问作者aiman