Is there an explanation when spark-csv won't save a DataFrame to file?

Question

dataFrame.coalesce(1).write().save("path") sometimes writes only _SUCCESS and ._SUCCESS.crc files without an expected *.csv.gz even on non-empty input DataFrame

file save code:

private static void writeCsvToDirectory(Dataset<Row> dataFrame, Path directory) {
    dataFrame.coalesce(1)
            .write()
            .format("csv")
            .option("header", "true")
            .option("delimiter", "\t")
            .option("codec", "org.apache.hadoop.io.compress.GzipCodec")
            .mode(SaveMode.Overwrite)
            .save("file:///" + directory);
}

file get code:

static Path getTemporaryCsvFile(Path directory) throws IOException {
    String glob = "*.csv.gz";
    try (DirectoryStream<Path> stream = Files.newDirectoryStream(directory, glob)) {
        return stream.iterator().next();
    } catch (NoSuchElementException e) {
        throw new RuntimeException(getNoSuchElementExceptionMessage(directory, glob), e);
    }
}

file get error example:

java.lang.RuntimeException: directory /tmp/temp5889805853850415940 does not contain a file with glob *.csv.gz. Directory listing:
    /tmp/temp5889805853850415940/_SUCCESS, 
    /tmp/temp5889805853850415940/._SUCCESS.crc

I rely on this expectation, can someone explain me why it work this way?

Redress

Pls feel free to exotic guesses =) I will be glad to read it too, not just complete solutions and full explonation to my problem – Redress

Do you know if the output file contains records? The movement from ._Success to _Success at least is showing that the write action (logistically-speaking) builds the temp files and moved them to their final resting place. – Tortuosity

Output file should (must by logic) contain at least the header line and some data lines. But he does not exist at all – Redress

During your write statement also include a println("#### Number of output records: " + inputDF.count + " ####"). On the output log run this 'cat output.log | grep ####' . This will be able to tell us if the problem is that your inputDF you are trying to write is empty or not. – Tortuosity

@Tortuosity inputDf cant be empty by-design. I omitted some details for brevity. You can read full class implementation at gist.github.com/banyrule/bc8b593de9fcb3fd911742970a890db2 – Redress

writeCsvToTemporaryDirectory method prohibits writing empty dataftames via .format("csv").save. – Redress

can we do dataFrame.coalesce(1).limit(1).collect to eliminate dataframe emptiness as a cause? – Daley

@BanyRule In what deploy-mode are you running your Spark app? – Acrolein

In case you are running this in cluster mode, then the local file output path would mean local path on each and every executor node. Can this be the problem? – Dirty

@VihitShah In OP case, it will be local file system of one executor that dataframe is being coalesced to. – Acrolein

Answer 1

Output file should (must by logic) contain at least the header line and some data lines. But he does not exist at all

This comment was a bit misleading. According to the code on Github, this will happen only if the Dataframe is empty, and won't produce SUCCESS files. Considering that those files are present - Dataframe is not empty and the writeCsvToDirectory from your code is triggered.

I have a couple of questions:

• Does your Spark job finish without errors?
• Does the timestamp of SUCCESS file gets updated?

My two main suspects are:

1. coalesce(1) - if you have a lot of data, this might fail
2. SaveMode.Overwrite - I have a feeling that those SUCCESS files are in that folder from previous runs

Tillis

> if you have a lot of data, this might fail Result file weighs about 0.5GB . > I have a feeling that those SUCCESS files are in that folder from previous runs This is obvious, but I have not noticed for so long. Thanks so much for this comment – Redress

Glad to help. 0.5 Gb should work fine with coalesce, unless you have a really small RAM. – Tillis

Answer 2

It is depend on your storage that you choose to write your csv file. if you write on hdfs everything's ok. but whenever you decide to write in your local files system you must care that nothing will be written in driver local files system and your data will be in worker's files system and you should find them in worker's storage.

two solution's:

1. Run Spark in Local Mode

set mater local[NUMBER_OF_CORES] that you can submit your job by --master local[10] config

2. Write In Distributed File System

write your data in distributed file system like s3,hdfs,...

Latashalatashia

Answer 3

My own solution solved this problem.

I replace .save("file://" with hadoopFileSystem.copyToLocalFile

The thing is .save("file:// works expectedly only with SparkSession.builder().master("local"), where hdfs:// is emulated by master's file://.

I may be wrong in theory, but it works.

static Path writeCsvToTemporaryDirectory(Dataset<Row> dataFrame) throws IOException {
    String temporaryDirectoryName = getTemporaryDirectoryName();

    writeCsvToDirectory(dataFrame, temporaryDirectoryName, sparkContext);

    return Paths.get(temporaryDirectoryName);
}

static void writeCsvToDirectory(Dataset<Row> dataFrame, String directory) throws IOException {
    dataFrame.coalesce(1)
        .write()
        .option("header", "true")
        .option("delimiter", "\t")
        .option("codec", "org.apache.hadoop.io.compress.GzipCodec")
        .mode(SaveMode.Overwrite)
        .csv(directory);

    FileSystem hadoopFileSystem = FileSystem.get(sparkContext.hadoopConfiguration());

    hadoopFileSystem.copyToLocalFile(true,
        new org.apache.hadoop.fs.Path(directory),
        new org.apache.hadoop.fs.Path(directory));
}

static Path getTemporaryCsvFile(Path directory) throws IOException {
    String glob = "*.csv.gz";

    try (DirectoryStream<Path> stream = Files.newDirectoryStream(directory, glob)) {
        return stream.iterator().next();
    } catch (NoSuchElementException e) {
        throw new RuntimeException(getNoSuchElementExceptionMessage(directory, glob), e);
    }
}

Path temporaryDirectory = writeCsvToTemporaryDirectory(dataFrame);
Path temporaryFile = DataFrameIOUtils.getTemporaryCsvFile(temporaryDirectory);

try {
    return otherStorage.upload(temporaryFile, name, fields).join();
} catch (InterruptedException | ExecutionException e) {
    throw new RuntimeException(e);
} finally {
    removeTemporaryDirectory(temporaryDirectory);
}

Redress