Skip to content

[SPARK-26208][SQL] add headers to empty csv files when header=true #23173

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 14 commits into from
Closed

[SPARK-26208][SQL] add headers to empty csv files when header=true #23173

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
3192656
write headers to empty csv files when header=true
koertkuipers Nov 28, 2018
aad5f09
Merge branch 'master' into feat-empty-csv-with-header
koertkuipers Nov 28, 2018
1f897be
add missing init overrides
koertkuipers Nov 28, 2018
bfadbf9
add missing init override
koertkuipers Nov 28, 2018
fd1ee62
Merge branch 'master' into feat-empty-csv-with-header
koertkuipers Nov 29, 2018
258a1e4
provide default do-nothing implementation for OutputWriter.init
koertkuipers Nov 29, 2018
088c710
cleanup test
koertkuipers Nov 29, 2018
879be71
add repartition to test
koertkuipers Nov 30, 2018
6f498a0
Merge branch 'master' into feat-empty-csv-with-header
koertkuipers Nov 30, 2018
29fc6b8
remove init
koertkuipers Nov 30, 2018
6165e1a
move code up so its clearer its part of constructor
koertkuipers Nov 30, 2018
5e87e7e
dont check number of partitions
koertkuipers Dec 1, 2018
238efa5
use lazy val for univocity generator
koertkuipers Dec 1, 2018
9d5cb7b
Revert "use lazy val for univocity generator"
koertkuipers Dec 1, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 4 additions & 5 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityGenerator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,6 @@ class UnivocityGenerator(
private val writerSettings = options.asWriterSettings
writerSettings.setHeaders(schema.fieldNames: _*)
private val gen = new CsvWriter(writer, writerSettings)
private var printHeader = options.headerFlag

// A `ValueConverter` is responsible for converting a value of an `InternalRow` to `String`.
// When the value is null, this converter should not be called.
Expand Down Expand Up @@ -72,15 +71,15 @@ class UnivocityGenerator(
values
}

def writeHeaders(): Unit = {
gen.writeHeaders()
}

/**
* Writes a single InternalRow to CSV using Univocity.
*/
def write(row: InternalRow): Unit = {
if (printHeader) {
gen.writeHeaders()
}
gen.writeRow(convertRow(row): _*)
printHeader = false
}

def writeToString(row: InternalRow): String = {
Expand Down
22 changes: 14 additions & 8 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -171,15 +171,21 @@ private[csv] class CsvOutputWriter(

private var univocityGenerator: Option[UnivocityGenerator] = None

override def write(row: InternalRow): Unit = {
val gen = univocityGenerator.getOrElse {
val charset = Charset.forName(params.charset)
val os = CodecStreams.createOutputStreamWriter(context, new Path(path), charset)
val newGen = new UnivocityGenerator(dataSchema, os, params)
univocityGenerator = Some(newGen)
newGen
}
if (params.headerFlag) {
val gen = getGen()
gen.writeHeaders()
}

private def getGen(): UnivocityGenerator = univocityGenerator.getOrElse {
val charset = Charset.forName(params.charset)
val os = CodecStreams.createOutputStreamWriter(context, new Path(path), charset)
val newGen = new UnivocityGenerator(dataSchema, os, params)
univocityGenerator = Some(newGen)
newGen
}

override def write(row: InternalRow): Unit = {
val gen = getGen()
Copy link
Member

@HyukjinKwon HyukjinKwon Dec 1, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Wait .. is this going to create UnivocityGenerator for each record?

Copy link
Member

@HyukjinKwon HyukjinKwon Dec 1, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, it's getOrElse. Okay but still can we simplify this logic? Looks a bit confusing. For instance, I think we can do this with lazy val.

Copy link
Member

@MaxGekk MaxGekk Dec 1, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@HyukjinKwon Do you mean creating generator in lazy val?

lazy val univocityGenerator = {
    val charset = Charset.forName(params.charset)
    val os = CodecStreams.createOutputStreamWriter(context, new Path(path), charset)
    new UnivocityGenerator(dataSchema, os, params)
}

The problem is in the close method, you will have to call univocityGenerator.close() in the method. If the lazy val wasn't instantiated before (empty partition and the header option is false), the generator will be created and closed immediately. And as a result, you will get an empty file for the empty partition. That's why I prefer the approach with Option[UnivocityGenerator] in #23052.

Copy link
Member

@HyukjinKwon HyukjinKwon Dec 1, 2018 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see the problem. OrcFileFormat uses a flag approach. For instance:

private var isGeneratorInitiated = false

lazy val univocityGenerator = {
  isGeneratorInitiated = true
  val charset = Charset.forName(params.charset)
  val os = CodecStreams.createOutputStreamWriter(context, new Path(path), charset)
  new UnivocityGenerator(dataSchema, os, params)
}

if (isGeneratorInitiated) {
  univocityGenerator.close()
}

Should be okay to stick to it.

Copy link
Contributor Author

@koertkuipers koertkuipers Dec 1, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok i changed it to lazy val and flag

Copy link
Member

@MaxGekk MaxGekk Dec 1, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Frankly speaking I don't see any reasons for this. For now we have 2 flags actually - isGeneratorInitiated and another one inside of lazy val. And 2 slightly different approaches - with the Option type in Json and Text, and lazy val + flag in Orc and CSV.

Copy link
Member

@srowen srowen Dec 1, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah we have two different approaches, both of which are fine IMHO. I think it's reasonable to clean that up in a follow-up if desired. WDYT @HyukjinKwon ?

Copy link
Contributor Author

@koertkuipers koertkuipers Dec 1, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i will revert this change to lazy val for now since it doesnt have anything to do wit this pullreq or jira: the Option approach was created in another pullreq.

Copy link
Member

@HyukjinKwon HyukjinKwon Dec 2, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OK. Shouldn't be a big deal.

gen.write(row)
}

Expand Down
13 changes: 13 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1987,6 +1987,19 @@ class CSVSuite extends QueryTest with SharedSQLContext with SQLTestUtils with Te
assert(errMsg2.contains("'lineSep' can contain only 1 character"))
}

test("SPARK-26208: write and read empty data to csv file with headers") {
withTempPath { path =>
val df1 = spark.range(10).repartition(2).filter(_ < 0).map(_.toString).toDF
// we have 2 partitions but they are both empty and will be filtered out upon writing
// thanks to SPARK-23271 one new empty partition will be inserted
df1.write.format("csv").option("header", true).save(path.getAbsolutePath)
val df2 = spark.read.format("csv").option("header", true).option("inferSchema", false)
.load(path.getAbsolutePath)
assert(df1.schema === df2.schema)
checkAnswer(df1, df2)
}
}

test("do not produce empty files for empty partitions") {
withTempPath { dir =>
val path = dir.getCanonicalPath
Expand Down