Continuous EMR consistency failures when writing CSV data in Spark

Viewed 108

I'm using Spark to update some CSV metadata.

The structure is that I aggregate a bunch of data into a CSV file and write it into a directory.

Code

    val processedMetadataDf = spark.read.csv(readPath)
    val processCount = processedMetadataDf.count()
    
    // Updated entry to processed metadata records
    val appendedRow = Seq(s3filePath)
    val appendedRowDf = appendedRow.toDF()
    if (processCount == 0) {
        println("Generating New Processed Metadata Enty")
        appendedRowDf.write.option("compression","uncompressed").mode("overwrite").csv("s3://" + metadataPath + "/PROCESSED/" + "month-05")
    } else {
        println("Union Merge With Existing Data")
        val mergedFrame = processedMetadataDf.union(appendedRowDf).toDF
        mergedFrame.toDF.show()
        mergedFrame.write.option("compression","uncompressed").mode("overwrite").csv("s3://" + metadataPath + "/PROCESSED/" + "month-05")      
    }
  }

month-05 in the above example is a directory and there should ideally just be 1 single CSV written into it from Spark. The first run of this works, the second run hits me with an inconsistency failure.

com.amazon.ws.emr.hadoop.fs.consistency.exception.ConsistencyException: 1 items inconsistent (no s3 object for associated metadata item).

I've ran EMRFS Sync, but my real question is why is this happening? I'm not modifying these files at all outside this script shown above.

My high-level logic is:

  1. If first time writing, write dataframe to CSV in s3 path
  2. If previous data existed, union the data, then write CSV to s3 path

Both are using overwrite as shown.

0 Answers
Related