diff --git a/python/pyspark/sql/readwriter.py b/python/pyspark/sql/readwriter.py
index b1453c637f79e2c75b61724ac7ab14964188f712..7f5368d8bdbb2fd25567b30db1246a110234d50b 100644
--- a/python/pyspark/sql/readwriter.py
+++ b/python/pyspark/sql/readwriter.py
@@ -233,6 +233,23 @@ class DataFrameReader(object):
paths = [paths]
return self._df(self._jreader.text(self._sqlContext._sc._jvm.PythonUtils.toSeq(paths)))
+ @since(2.0)
+ def csv(self, paths):
+ """Loads a CSV file and returns the result as a [[DataFrame]].
+
+ This function goes through the input once to determine the input schema. To avoid going
+ through the entire data once, specify the schema explicitly using [[schema]].
+
+ :param paths: string, or list of strings, for input path(s).
+
+ >>> df = sqlContext.read.csv('python/test_support/sql/ages.csv')
+ >>> df.dtypes
+ [('C0', 'string'), ('C1', 'string')]
+ """
+ if isinstance(paths, basestring):
+ paths = [paths]
+ return self._df(self._jreader.csv(self._sqlContext._sc._jvm.PythonUtils.toSeq(paths)))
+
@since(1.5)
def orc(self, path):
"""Loads an ORC file, returning the result as a :class:`DataFrame`.
@@ -448,6 +465,11 @@ class DataFrameWriter(object):
* ``ignore``: Silently ignore this operation if data already exists.
* ``error`` (default case): Throw an exception if data already exists.
+ You can set the following JSON-specific option(s) for writing JSON files:
+ * ``compression`` (default ``None``): compression codec to use when saving to file.
+ This can be one of the known case-insensitive shorten names
+ (``bzip2``, ``gzip``, ``lz4``, and ``snappy``).
+
>>> df.write.json(os.path.join(tempfile.mkdtemp(), 'data'))
"""
self.mode(mode)._jwrite.json(path)
@@ -476,11 +498,39 @@ class DataFrameWriter(object):
def text(self, path):
"""Saves the content of the DataFrame in a text file at the specified path.
+ :param path: the path in any Hadoop supported file system
+
The DataFrame must have only one column that is of string type.
Each row becomes a new line in the output file.
+
+ You can set the following option(s) for writing text files:
+ * ``compression`` (default ``None``): compression codec to use when saving to file.
+ This can be one of the known case-insensitive shorten names
+ (``bzip2``, ``gzip``, ``lz4``, and ``snappy``).
"""
self._jwrite.text(path)
+ @since(2.0)
+ def csv(self, path, mode=None):
+ """Saves the content of the [[DataFrame]] in CSV format at the specified path.
+
+ :param path: the path in any Hadoop supported file system
+ :param mode: specifies the behavior of the save operation when data already exists.
+
+ * ``append``: Append contents of this :class:`DataFrame` to existing data.
+ * ``overwrite``: Overwrite existing data.
+ * ``ignore``: Silently ignore this operation if data already exists.
+ * ``error`` (default case): Throw an exception if data already exists.
+
+ You can set the following CSV-specific option(s) for writing CSV files:
+ * ``compression`` (default ``None``): compression codec to use when saving to file.
+ This can be one of the known case-insensitive shorten names
+ (``bzip2``, ``gzip``, ``lz4``, and ``snappy``).
+
+ >>> df.write.csv(os.path.join(tempfile.mkdtemp(), 'data'))
+ """
+ self.mode(mode)._jwrite.csv(path)
+
@since(1.5)
def orc(self, path, mode=None, partitionBy=None):
"""Saves the content of the :class:`DataFrame` in ORC format at the specified path.
diff --git a/python/test_support/sql/ages.csv b/python/test_support/sql/ages.csv
new file mode 100644
index 0000000000000000000000000000000000000000..18991feda788aadd1fc7fb8c6943baeb54139986
--- /dev/null
+++ b/python/test_support/sql/ages.csv
@@ -0,0 +1,4 @@
+Joe,20
+Tom,30
+Hyukjin,25
+
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
index d6bdd3d825565416a1d9ea1c8e4f29371a237333..093504c765ee863b791c146c0d1cb1eaf423af0e 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
@@ -453,6 +453,10 @@ final class DataFrameWriter private[sql](df: DataFrame) {
* format("json").save(path)
* }}}
*
+ * You can set the following JSON-specific option(s) for writing JSON files:
+ * <li>`compression` (default `null`): compression codec to use when saving to file. This can be
+ * one of the known case-insensitive shorten names (`bzip2`, `gzip`, `lz4`, and `snappy`). </li>
+ *
* @since 1.4.0
*/
def json(path: String): Unit = format("json").save(path)
@@ -492,10 +496,29 @@ final class DataFrameWriter private[sql](df: DataFrame) {
* df.write().text("/path/to/output")
* }}}
*
+ * You can set the following option(s) for writing text files:
+ * <li>`compression` (default `null`): compression codec to use when saving to file. This can be
+ * one of the known case-insensitive shorten names (`bzip2`, `gzip`, `lz4`, and `snappy`). </li>
+ *
* @since 1.6.0
*/
def text(path: String): Unit = format("text").save(path)
+ /**
+ * Saves the content of the [[DataFrame]] in CSV format at the specified path.
+ * This is equivalent to:
+ * {{{
+ * format("csv").save(path)
+ * }}}
+ *
+ * You can set the following CSV-specific option(s) for writing CSV files:
+ * <li>`compression` (default `null`): compression codec to use when saving to file. This can be
+ * one of the known case-insensitive shorten names (`bzip2`, `gzip`, `lz4`, and `snappy`). </li>
+ *
+ * @since 2.0.0
+ */
+ def csv(path: String): Unit = format("csv").save(path)
+
///////////////////////////////////////////////////////////////////////////////////////
// Builder pattern config options
///////////////////////////////////////////////////////////////////////////////////////
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JSONOptions.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JSONOptions.scala
index 31a95ed461215a540ee9279bc2b15cc428082e5a..e59dbd6b3d438175e53ca48b546f4dc4e5e73e09 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JSONOptions.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JSONOptions.scala
@@ -48,10 +48,7 @@ private[sql] class JSONOptions(
parameters.get("allowNonNumericNumbers").map(_.toBoolean).getOrElse(true)
val allowBackslashEscapingAnyCharacter =
parameters.get("allowBackslashEscapingAnyCharacter").map(_.toBoolean).getOrElse(false)
- val compressionCodec = {
- val name = parameters.get("compression").orElse(parameters.get("codec"))
- name.map(CompressionCodecs.getCodecClassName)
- }
+ val compressionCodec = parameters.get("compression").map(CompressionCodecs.getCodecClassName)
/** Sets config options on a Jackson [[JsonFactory]]. */
def setJacksonOptions(factory: JsonFactory): Unit = {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/text/DefaultSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/text/DefaultSource.scala
index 60155b32349a752ca7eeef868364e9765c9fce1f..8f3f6335e4282149a909dd6c599aafe5e234308c 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/text/DefaultSource.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/text/DefaultSource.scala
@@ -115,10 +115,7 @@ private[sql] class TextRelation(
/** Write path. */
override def prepareJobForWrite(job: Job): OutputWriterFactory = {
val conf = job.getConfiguration
- val compressionCodec = {
- val name = parameters.get("compression").orElse(parameters.get("codec"))
- name.map(CompressionCodecs.getCodecClassName)
- }
+ val compressionCodec = parameters.get("compression").map(CompressionCodecs.getCodecClassName)
compressionCodec.foreach { codec =>
CompressionCodecs.setCodecConfiguration(conf, codec)
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala
index 5d57d77ab054998fd4a21af9c79a2d9bd540cbdd..3ecbb14f2ea656e36fff6fd11225e839590a6e9a 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala
@@ -268,9 +268,8 @@ class CSVSuite extends QueryTest with SharedSQLContext with SQLTestUtils {
.load(testFile(carsFile))
cars.coalesce(1).write
- .format("csv")
.option("header", "true")
- .save(csvDir)
+ .csv(csvDir)
val carsCopy = sqlContext.read
.format("csv")