Spark使用Java API读取仅首文件含表头的多CSV文件

Got it, let's tackle this common scenario where you've got a bunch of CSV files, but only the first one includes column headers, and you need to read them all correctly using Spark's Java API. Here's a clean, reliable approach that works every time:

Step 1: Initialize SparkSession #

First, set up your SparkSession as you normally would:

import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.StructType;
import java.io.File;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

SparkSession spark = SparkSession.builder()
        .appName("MultiCSVWithSingleHeader")
        .master("local[*]") // Remove this line for production clusters
        .getOrCreate();

Step 2: Identify Your Files #

If you know the exact path of the header-containing file, great. If not, you can list and sort the CSV files in your directory to grab the first one:

// Path to your directory with CSV files
File csvDir = new File("/path/to/your/csv/files");

// Filter and sort CSV files alphabetically (adjust sorting logic if needed)
File[] csvFiles = csvDir.listFiles((dir, name) -> name.toLowerCase().endsWith(".csv"));
Arrays.sort(csvFiles);

// Split into first file (with headers) and remaining files (no headers)
String headerFile = csvFiles[0].getAbsolutePath();
List<String> dataFiles = new ArrayList<>();
for (int i = 1; i < csvFiles.length; i++) {
    dataFiles.add(csvFiles[i].getAbsolutePath());
}

Step 3: Read the Header File & Extract Schema #

Read the first file with header=true to get the correct column names and data types. We'll reuse this schema for the other files:

// Read the header file to get structured data and schema
Dataset<Row> headerDf = spark.read()
        .option("header", "true")
        .option("inferSchema", "true") // Auto-detects column types; replace with manual schema if needed
        .csv(headerFile);

// Extract the schema to apply to other files
StructType csvSchema = headerDf.schema();

Pro Tip: Use a Manual Schema for Better Performance #

If you know your schema upfront, defining it manually is faster and more reliable than inferring:

import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;

StructType csvSchema = new StructType(new StructField[]{
        DataTypes.createStructField("id", DataTypes.IntegerType, true),
        DataTypes.createStructField("name", DataTypes.StringType, true)
});

// When reading the header file, use this schema instead of inferSchema:
Dataset<Row> headerDf = spark.read()
        .schema(csvSchema)
        .option("header", "true")
        .csv(headerFile);

Step 4: Read Remaining Files & Merge #

Read the rest of the files using the schema we extracted (no need for headers here), then union it with the header file's data:

// Read data files with the pre-defined schema (no headers)
Dataset<Row> dataDf = spark.read()
        .schema(csvSchema)
        .option("header", "false")
        .csv(dataFiles.toArray(new String[0]));

// Merge the two DataFrames using unionByName (safer than union for column order consistency)
Dataset<Row> finalDf = headerDf.unionByName(dataDf);

// Verify the result
finalDf.show();

Key Notes #

• unionByName vs union: Always use unionByName if there's any chance column orders might differ (even if not here, it's a good habit). It matches columns by name instead of position.
• Handling Large Datasets: This approach is efficient because it avoids scanning files multiple times. The schema is determined once from the first file, then applied to all others.
• Edge Cases: If your header file has empty rows or malformed data, add options like .option("skipRows", 1) if needed, but adjust accordingly.

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