如何让Spark跳过失败任务？HDFS畸形数据致任务启动失败求解决方案

Great question! You’re absolutely correct that Spark doesn’t currently offer a direct equivalent to MapReduce’s mapreduce.map.failures.maxpercent parameter — as you noted, this feature hasn’t been implemented yet. But don’t worry, there are several practical workarounds to handle malformed records and keep your Spark job running without crashing during the initial read.

Here are the most reliable approaches:

1. Catch Exceptions Manually During RDD Transformations #

The key issue here is that context.newAPIHadoopFile() fails during the read phase, before you get a chance to process records. To get around this, first read the data in its raw, unprocessed form (like byte streams), then handle deserialization and error-catching in subsequent map/flatMap operations.

For example, if you’re reading a SequenceFile:

// Read raw bytes first to avoid early failure
val rawRDD = sc.newAPIHadoopFile[BytesWritable, BytesWritable, SequenceFileInputFormat]("hdfs://your/data/path")

// Process each record, skipping malformed ones
val validRecordsRDD = rawRDD.flatMap { case (keyBytes, valueBytes) =>
  try {
    // Your custom deserialization logic here
    val deserializedRecord = yourDeserializationFunction(keyBytes, valueBytes)
    Some(deserializedRecord)
  } catch {
    case e: Exception =>
      // Log the error for debugging (optional but recommended)
      println(s"Skipping malformed record: ${e.getMessage}")
      None
  }
}

This shifts the error handling from the read phase to the transformation phase, letting you explicitly skip bad records instead of crashing the entire job.

2. Use Spark SQL’s Built-in Bad Record Handling (If Applicable) #

If your data format is supported by Spark SQL (CSV, JSON, Parquet, etc.), Spark provides built-in parameters to handle malformed records:

• Set mode to DROPMALFORMED to automatically discard any records that can’t be parsed.
• Use PERMISSIVE (default) to set invalid fields to null instead of failing.
• Specify spark.sql.csv.badRecordsPath or spark.sql.json.badRecordsPath to write malformed records to a separate HDFS path for later analysis.

Example for CSV data:

val validDF = spark.read
  .option("mode", "DROPMALFORMED")
  .option("header", "true")
  .csv("hdfs://your/csv/data/path")

Even if you’re using a custom Hadoop InputFormat, you can convert the raw RDD to a DataFrame first and apply these settings.

3. Wrap the Hadoop InputFormat with a Custom RecordReader #

For more low-level control, you can create a custom InputFormat that wraps your original InputFormat and handles exceptions in the RecordReader. Here’s the core idea:

• Extend your target InputFormat and override createRecordReader.
• Implement a custom RecordReader where the nextKeyValue() method catches exceptions, skips bad records, and continues reading until a valid record is found or the input is exhausted.

This approach decouples error handling from your Spark business logic and is useful if you need to reuse the fault-tolerant InputFormat across multiple jobs.

4. Configure Task Retries (As a Complementary Step) #

While this doesn’t directly skip bad records, increasing task retry counts can help with transient failures. If a task fails due to a malformed record, Spark will retry it up to the specified limit. Combine this with one of the above methods for best results:

# Set in spark-defaults.conf or when submitting the job
spark.task.maxFailures=5

Note: This won’t fix permanent failures from consistently malformed records, but it’s a good safety net for occasional issues.

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