Let's break down your problem first: you used CUBE on gender and department to get 4 distinct aggregation groups, but your current approach with partitionBy creates extra folders/files with Spark's default naming, which doesn't match your requirement of 4 custom-named files.

Why Your Current Code Isn't Working #

When you use partitionBy("combination"), Spark creates a subfolder for each unique combination value, and inside each folder, you get Spark's default part-*.csv files. This doesn't let you directly set the exact filename you want, and it's not aligned with your goal of having 4 standalone files.

Solution: Filter & Export Each Group Separately #

The right approach is to filter the CubeData DataFrame to isolate each of the 4 groups, then export each filtered DataFrame to a custom-named file. Since Spark doesn't let you directly specify the output filename, we'll use a temporary directory and then rename the generated part-* file to your desired name.

Step 1: Filter Each Target Group

First, isolate each of the 4 groups with explicit filters:

import org.apache.spark.sql.functions.col

// 1. Records with non-null gender AND non-null department
val genderDepartmentsDF = CubeData.filter(col("gender").isNotNull && col("department").isNotNull)

// 2. Records with non-null gender AND null department
val genderNullDF = CubeData.filter(col("gender").isNotNull && col("department").isNull)

// 3. Records with null gender AND non-null department
val departmentsNullDF = CubeData.filter(col("gender").isNull && col("department").isNotNull)

// 4. Records with null gender AND null department
val nullNullDF = CubeData.filter(col("gender").isNull && col("department").isNull)

Step 2: Create a Reusable Export Function

To avoid repeating code for each export, we'll create a helper function that handles writing to a temp directory, renaming the file, and cleaning up:

import org.apache.hadoop.fs.{FileSystem, Path}
import java.net.URI

def exportToCustomFile(df: org.apache.spark.sql.DataFrame, baseOutputDir: String, fileName: String): Unit = {
  // Step 1: Write to a temporary directory
  val tempDir = s"$baseOutputDir/temp_$fileName"
  df.coalesce(1) // Combine data into a single partition to get one part file
    .write
    .format("csv")
    .option("header", "true")
    .mode("overwrite")
    .save(tempDir)

  // Step 2: Get the file system (use "file:///" for local filesystem instead of HDFS)
  val fs = FileSystem.get(new URI("hdfs:///"), spark.sparkContext.hadoopConfiguration)
  
  // Step 3: Find the generated part file
  val partFiles = fs.listStatus(new Path(tempDir))
    .filter(status => status.getPath.getName.startsWith("part-"))
  
  if (partFiles.nonEmpty) {
    // Rename the part file to your custom name
    val sourcePath = partFiles(0).getPath
    val targetPath = new Path(s"$baseOutputDir/$fileName.csv")
    
    // Delete target file if it already exists
    if (fs.exists(targetPath)) fs.delete(targetPath, true)
    
    fs.rename(sourcePath, targetPath)
  }

  // Step 4: Clean up the temporary directory
  fs.delete(new Path(tempDir), true)
}

Step 3: Export All 4 Files

Call the helper function for each filtered DataFrame:

val outputDir = "final"

exportToCustomFile(genderDepartmentsDF, outputDir, "gender_departments")
exportToCustomFile(genderNullDF, outputDir, "gender_null")
exportToCustomFile(departmentsNullDF, outputDir, "departments_null")
exportToCustomFile(nullNullDF, outputDir, "null_null")

Key Notes #

• coalesce(1): This combines all data into one partition, ensuring only one part-* file is generated (easy to rename). If you're working with large datasets, this could cause performance issues—consider skipping it if you don't strictly need a single file, but adjust the renaming logic to handle multiple part files.
• Filesystem URI: If you're using a local filesystem instead of HDFS, replace "hdfs:///" with "file:///".
• Permissions: Ensure your Spark job has read/write permissions for the outputDir and temp directories.

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