Spark Scala中如何判断CSV文件是否为空并执行对应逻辑？

Absolutely! You can definitely implement this empty CSV check logic in Spark Scala — we need to cover two key scenarios: when the file is missing/completely empty, and when it has a header row but no actual data records. Here's a step-by-step solution tailored to your needs:

Full Implementation Code #

import org.apache.spark.sql.SparkSession
import org.apache.hadoop.fs.{FileSystem, Path}

object CsvProcessor {
  def main(args: Array[String]): Unit = {
    // Initialize Spark Session (adjust master for cluster mode)
    val spark = SparkSession.builder()
      .appName("CsvEmptyValidation")
      .master("local[*]") // Remove this line when running on a cluster
      .getOrCreate()

    val csvFilePath = "/path/to/your/target.csv" // Replace with your actual CSV path
    val hadoopFs = FileSystem.get(spark.sparkContext.hadoopConfiguration)
    val path = new Path(csvFilePath)

    // 1. Check if file exists or is completely empty (0 bytes)
    if (!hadoopFs.exists(path) || hadoopFs.getFileStatus(path).getLen == 0) {
      println("CSV file is missing or has no content. Exiting program.")
      spark.stop()
      sys.exit(0)
    }

    // 2. Read CSV and check for actual data rows (beyond header)
    val csvDf = spark.read
      .option("header", "true") // Remove this if your CSV has no header
      .option("inferSchema", "true")
      .csv(csvFilePath)

    // Efficient check: instead of full count, just check if there's at least one row
    val hasDataRecords = csvDf.take(1).nonEmpty

    if (!hasDataRecords) {
      println("CSV file only contains a header (no data records). Exiting program.")
      spark.stop()
      sys.exit(0)
    }

    // If we reach here, proceed with your data processing logic
    println("CSV has valid data records. Starting processing...")
    // Example processing steps:
    // csvDf.printSchema()
    // csvDf.show(5)

    spark.stop()
  }
}

Key Details to Note #

• Hadoop FileSystem Check: This handles the edge case where the file doesn't exist or is entirely empty (0 bytes). Spark uses Hadoop's FileSystem abstraction, so this works for both local files and distributed storage like HDFS/S3.
• Efficient Data Row Check: Using take(1).nonEmpty is better than count() for large files, as it only reads the first row instead of scanning the entire dataset.
• Header Handling: If your CSV doesn't have a header, remove the option("header", "true") line — the take(1).nonEmpty check will then directly verify if there are any data rows.

Alternative for Multiple Files #

If you're dealing with a directory of CSV files, adjust the check to iterate over the files or use Spark's built-in methods to check the combined dataset:

val csvDf = spark.read.csv("/path/to/csv/directory")
if (csvDf.isEmpty) {
  // Exit logic
}

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