Apache Spark Java实现：遍历Row数据集并移除空值字段

Hey there! Let's work through this problem together. Your goal is to remove null/blank fields per individual record when writing JSON output (not drop entire columns), while avoiding the expensive collect() operation that risks OOM with large datasets. Here's how to solve this efficiently:

Core Idea #

Spark's default JSON writer retains null fields, but we need to dynamically filter out empty values (null, empty strings, all-whitespace strings) for each row. Using distributed operations (not driver-side collection) is key—we can convert each row to a Map<String, Object> where only non-empty fields are kept, since Spark serializes maps to JSON without including missing keys.

Solution 1: Use a UDF for Generic Row Filtering #

This approach works directly with your Dataset<Row> and handles all columns dynamically, no need for a specific bean class upfront:

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import static org.apache.spark.sql.functions.*;
import java.util.HashMap;
import java.util.Map;

public class ExtractionExample {
    public static void main(String[] args) throws Exception {
        SparkSession spark = SparkSession.builder()
                .appName("ExtractionExample")
                .config("spark.sql.warehouse.dir", "/user/hive/warehouse/")
                .enableHiveSupport()
                .getOrCreate();

        // Read Hive table data
        Dataset<Row> sqlDF = spark.sql("SELECT person_id as personId, person_name as personName, email_id as email FROM person");

        // Register UDF to filter null/blank fields per row
        spark.udf().register("filterEmptyFields", (Row row) -> {
            Map<String, Object> filteredMap = new HashMap<>();
            String[] fieldNames = row.schema().fieldNames();
            
            for (int i = 0; i < fieldNames.length; i++) {
                Object value = row.get(i);
                if (value != null) {
                    // Handle string values: trim and check if non-empty
                    if (value instanceof String) {
                        String trimmedStr = ((String) value).trim();
                        if (!trimmedStr.isEmpty()) {
                            filteredMap.put(fieldNames[i], trimmedStr);
                        }
                    } else {
                        // Keep non-string values (numbers, dates, etc.) if not null
                        filteredMap.put(fieldNames[i], value);
                    }
                }
            }
            return filteredMap;
        }, Encoders.javaSerialization(Map.class));

        // Apply UDF and write JSON output
        sqlDF.select(callUDF("filterEmptyFields", struct(sqlDF.columns())))
                .write()
                .mode("overwrite") // Adjust mode as needed: append, ignore, etc.
                .json("/data/testdb/test/person_json");

        spark.stop();
    }
}

Solution 2: Use Bean Class with Distributed Map Operation #

If you prefer working with your Person bean class, you can use Dataset.map() (distributed, not driver-side) to convert each bean to a filtered map:

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import com.fdc.model.Person;
import java.util.HashMap;
import java.util.Map;

public class ExtractionExample {
    public static void main(String[] args) throws Exception {
        SparkSession spark = SparkSession.builder()
                .appName("ExtractionExample")
                .config("spark.sql.warehouse.dir", "/user/hive/warehouse/")
                .enableHiveSupport()
                .getOrCreate();

        Dataset<Row> sqlDF = spark.sql("SELECT person_id as personId, person_name as personName, email_id as email FROM person");
        Dataset<Person> personDS = sqlDF.as(Encoders.bean(Person.class));

        // Convert each Person to a filtered map
        Dataset<Map<String, Object>> filteredDS = personDS.map(person -> {
            Map<String, Object> filteredMap = new HashMap<>();
            
            if (person.getPersonId() != null) {
                filteredMap.put("personId", person.getPersonId());
            }
            if (person.getPersonName() != null && !person.getPersonName().trim().isEmpty()) {
                filteredMap.put("personName", person.getPersonName().trim());
            }
            if (person.getEmail() != null && !person.getEmail().trim().isEmpty()) {
                filteredMap.put("email", person.getEmail().trim());
            }
            
            return filteredMap;
        }, Encoders.javaSerialization(Map.class));

        // Write filtered data to JSON
        filteredDS.write()
                .mode("overwrite")
                .json("/data/testdb/test/person_json");

        spark.stop();
    }
}

Why Your Previous Approaches Failed #

• First method: drop(column) removes an entire column from the dataset, not just empty values in individual rows. Also, sqlDF.select(column).toString() doesn't get actual column values—it returns a string representation of the Dataset, so your null check never triggered.
• Second method: collect() pulls all data to the driver, which is catastrophic for large datasets. Additionally, Spark Row objects have a fixed schema—you can't dynamically add/remove fields per row, which is why your RowFactory approach threw errors. Maps are the right choice here because they support dynamic key-value pairs.

Key Advantages of These Solutions #

• Distributed processing: All operations run on executors, no driver-side data collection.
• Per-record filtering: Only removes empty fields from the rows where they're empty (other rows keep the field if it has a value).
• Handles multiple data types: Works for strings, numbers, dates, etc., with minimal adjustments.

内容的提问来源于stack exchange，提问作者Srinivas