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:
- If first time writing, write dataframe to CSV in s3 path
- If previous data existed, union the data, then write CSV to s3 path
Both are using overwrite as shown.