Skip to content

[SPARK-19018][SQL] Add support for custom encoding on csv writer #20949

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
crafty-coder wants to merge 8 commits into from
Closed

[SPARK-19018][SQL] Add support for custom encoding on csv writer #20949

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
b9a7bf0
[SPARK-19018][SQL] Add support for custom encoding on csv writer
crafty-coder Mar 30, 2018
8fd3e0e
Merge remote-tracking branch 'upstream/master'
crafty-coder May 6, 2018
0d0addf
Fix checkstyle issue
crafty-coder May 6, 2018
91f4750
Add test to check UTF-16 and UTF-32
crafty-coder Jul 16, 2018
349e132
Add jira ticked (SPARK-19018:) as prefix on the related tests
crafty-coder Jul 17, 2018
fd857b0
Add styling improvements
crafty-coder Jul 18, 2018
b6311e3
Merge branch 'master' into master
crafty-coder Jul 18, 2018
025958a
Fix import order on CSVSuite
crafty-coder Jul 18, 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
7 changes: 5 additions & 2 deletions python/pyspark/sql/readwriter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -826,7 +826,7 @@ def text(self, path, compression=None, lineSep=None):
def csv(self, path, mode=None, compression=None, sep=None, quote=None, escape=None,
header=None, nullValue=None, escapeQuotes=None, quoteAll=None, dateFormat=None,
timestampFormat=None, ignoreLeadingWhiteSpace=None, ignoreTrailingWhiteSpace=None,
charToEscapeQuoteEscaping=None):
charToEscapeQuoteEscaping=None, encoding=None):
"""Saves the content of the :class:`DataFrame` in CSV format at the specified path.

:param path: the path in any Hadoop supported file system
Expand Down Expand Up @@ -876,6 +876,8 @@ def csv(self, path, mode=None, compression=None, sep=None, quote=None, escape=No
the quote character. If None is set, the default value is
escape character when escape and quote characters are
different, ``\0`` otherwise..
:param encoding: sets encoding used for encoding the file. If None is set, it

@MaxGekk MaxGekk Jun 24, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Could you reformulate this encoding used for encoding

uses the default value, ``UTF-8``.

@HyukjinKwon HyukjinKwon Jul 18, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Likewise, let's match the doc to JSON's.


>>> df.write.csv(os.path.join(tempfile.mkdtemp(), 'data'))
"""
Expand All @@ -885,7 +887,8 @@ def csv(self, path, mode=None, compression=None, sep=None, quote=None, escape=No
dateFormat=dateFormat, timestampFormat=timestampFormat,
ignoreLeadingWhiteSpace=ignoreLeadingWhiteSpace,
ignoreTrailingWhiteSpace=ignoreTrailingWhiteSpace,
charToEscapeQuoteEscaping=charToEscapeQuoteEscaping)
charToEscapeQuoteEscaping=charToEscapeQuoteEscaping,
encoding=encoding)
self._jwrite.csv(path)

@since(1.5)
Expand Down
1 change: 1 addition & 0 deletions sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -623,6 +623,7 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) {
* enclosed in quotes. Default is to only escape values containing a quote character.</li>
* <li>`header` (default `false`): writes the names of columns as the first line.</li>
* <li>`nullValue` (default empty string): sets the string representation of a null value.</li>
* <li>`encoding` (default `UTF-8`): encoding to use when saving to file.</li>

@HyukjinKwon HyukjinKwon Jul 18, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

I think we should match the doc with JSON's

spark/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala

Lines 525 to 526 in 6ea582e

* <li>`encoding` (by default it is not set): specifies encoding (charset) of saved json
* files. If it is not set, the UTF-8 charset will be used. </li>

* <li>`compression` (default `null`): compression codec to use when saving to file. This can be
* one of the known case-insensitive shorten names (`none`, `bzip2`, `gzip`, `lz4`,
* `snappy` and `deflate`). </li>
Expand Down
10 changes: 9 additions & 1 deletion 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 @@ -17,6 +17,8 @@

package org.apache.spark.sql.execution.datasources.csv

import java.nio.charset.Charset

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileStatus, Path}
import org.apache.hadoop.mapreduce._
Expand Down Expand Up @@ -146,7 +148,13 @@ private[csv] class CsvOutputWriter(
context: TaskAttemptContext,
params: CSVOptions) extends OutputWriter with Logging {

private val writer = CodecStreams.createOutputStreamWriter(context, new Path(path))
private val charset = Charset.forName(params.charset)

private val writer = CodecStreams.createOutputStreamWriter(
context,

@HyukjinKwon HyukjinKwon Jul 18, 2018 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

tiny nit:

private val writer = CodecStreams.createOutputStreamWriter(context, new Path(path), charset)

new Path(path),
charset
)

@MaxGekk MaxGekk Jun 24, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Move the ) up like charset). See https://github.com/databricks/scala-style-guide


private val gen = new UnivocityGenerator(dataSchema, writer, params)

Expand Down
37 changes: 37 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 @@ -513,6 +513,43 @@ class CSVSuite extends QueryTest with SharedSQLContext with SQLTestUtils {
}
}

test("Save csv with custom charset") {

@MaxGekk MaxGekk Jul 16, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Could you prepend SPARK-19018 to the test title.

Seq("iso-8859-1", "utf-8", "windows-1250").foreach { encoding =>

@MaxGekk MaxGekk Jun 24, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Could you check the UTF-16 and UTF-32 encoding too. The written csv files must contain BOMs for such encodings. I am not sure that Spark CSV datasource is able to read it in per-line mode (multiLine is set to false). Probably, you need to switch to multLine mode or read the files by Scala's library like in JsonSuite:

spark/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala

Lines 2322 to 2338 in c7e2742

test("SPARK-23723: write json in UTF-16/32 with multiline off") {
Seq("UTF-16", "UTF-32").foreach { encoding =>
withTempPath { path =>
val ds = spark.createDataset(Seq(("a", 1))).repartition(1)
ds.write
.option("encoding", encoding)
.option("multiline", false)
.json(path.getCanonicalPath)
val jsonFiles = path.listFiles().filter(_.getName.endsWith("json"))
jsonFiles.foreach { jsonFile =>
val readback = Files.readAllBytes(jsonFile.toPath)
val expected = ("""{"_1":"a","_2":1}""" + "\n").getBytes(Charset.forName(encoding))
assert(readback === expected)
}
}
}
}

withTempDir { dir =>

@HyukjinKwon HyukjinKwon Jul 18, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

withTempDir -> withTempPath

val csvDir = new File(dir, "csv").getCanonicalPath
// scalastyle:off

@HyukjinKwon HyukjinKwon Jun 25, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Let's ignore the specific rule for this, e.g.:

// scalastyle:off nonascii
...
// scalastyle:on nonascii

val originalDF = Seq("µß áâä ÁÂÄ").toDF("_c0")
// scalastyle:on
originalDF.write
.option("header", "false")

@MaxGekk MaxGekk Jun 24, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

The header flag is disabled by default. Just in case, are there any specific reasons fro testing without CSV header?

@crafty-coder crafty-coder Jul 17, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

My bad, there is no reason. It's fixed on the next commit.

.option("encoding", encoding)
.csv(csvDir)

val df = spark
.read
.option("header", "false")
.option("encoding", encoding)

@HyukjinKwon HyukjinKwon Mar 31, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

I think our CSV read encoding option is incomplete for now .. there are many discussions about this now. I am going to fix the read path soon. Let me revisit this after fixing it.

@HyukjinKwon HyukjinKwon Jun 25, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Now it's fine. I think we decided to support encoding in CSV/JSON datasources. Ignore the comment above. We can proceed separately.

.csv(csvDir)

checkAnswer(df, originalDF)
}
}
}

test("bad encoding name on writer") {
val exception = intercept[SparkException] {
withTempDir { dir =>

@HyukjinKwon HyukjinKwon Jul 18, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

withTempPath

val csvDir = new File(dir, "csv").getCanonicalPath
Seq("a,A,c,A,b,B").toDF().write
.option("header", "false")
.option("encoding", "1-9588-osi")
.csv(csvDir)

@HyukjinKwon HyukjinKwon Jul 25, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

nit: you could use directly path.getCanonicalPath

}
}

assert(exception.getCause.getMessage.contains("1-9588-osi"))
}

test("commented lines in CSV data") {
Seq("false", "true").foreach { multiLine =>

Expand Down