Coverage for pyspark/sql/readwriter.py : 80%

# 

# Licensed to the Apache Software Foundation (ASF) under one or more 

# contributor license agreements. See the NOTICE file distributed with 

# this work for additional information regarding copyright ownership. 

# The ASF licenses this file to You under the Apache License, Version 2.0 

# (the "License"); you may not use this file except in compliance with 

# the License. You may obtain a copy of the License at 

# 

# http://www.apache.org/licenses/LICENSE-2.0 

# 

# Unless required by applicable law or agreed to in writing, software 

# distributed under the License is distributed on an "AS IS" BASIS, 

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

# See the License for the specific language governing permissions and 

# limitations under the License. 

# 

import sys 

from py4j.java_gateway import JavaClass 

from pyspark import RDD, since 

from pyspark.sql.column import _to_seq, _to_java_column 

from pyspark.sql.types import StructType 

from pyspark.sql import utils 

from pyspark.sql.utils import to_str 

__all__ = ["DataFrameReader", "DataFrameWriter"] 

class OptionUtils(object): 

def _set_opts(self, schema=None, **options): 

""" 

Set named options (filter out those the value is None) 

""" 

if schema is not None: 

self.schema(schema) 

for k, v in options.items(): 

if v is not None: 

self.option(k, v) 

class DataFrameReader(OptionUtils): 

""" 

Interface used to load a :class:`DataFrame` from external storage systems 

(e.g. file systems, key-value stores, etc). Use :attr:`SparkSession.read` 

to access this. 

.. versionadded:: 1.4 

""" 

def __init__(self, spark): 

self._jreader = spark._ssql_ctx.read() 

self._spark = spark 

def _df(self, jdf): 

from pyspark.sql.dataframe import DataFrame 

return DataFrame(jdf, self._spark) 

def format(self, source): 

"""Specifies the input data source format. 

.. versionadded:: 1.4.0 

Parameters 

---------- 

source : str 

string, name of the data source, e.g. 'json', 'parquet'. 

Examples 

-------- 

>>> df = spark.read.format('json').load('python/test_support/sql/people.json') 

>>> df.dtypes 

[('age', 'bigint'), ('name', 'string')] 

""" 

self._jreader = self._jreader.format(source) 

return self 

def schema(self, schema): 

"""Specifies the input schema. 

Some data sources (e.g. JSON) can infer the input schema automatically from data. 

By specifying the schema here, the underlying data source can skip the schema 

inference step, and thus speed up data loading. 

.. versionadded:: 1.4.0 

Parameters 

---------- 

schema : :class:`pyspark.sql.types.StructType` or str 

a :class:`pyspark.sql.types.StructType` object or a DDL-formatted string 

(For example ``col0 INT, col1 DOUBLE``). 

>>> s = spark.read.schema("col0 INT, col1 DOUBLE") 

""" 

from pyspark.sql import SparkSession 

spark = SparkSession.builder.getOrCreate() 

if isinstance(schema, StructType): 

jschema = spark._jsparkSession.parseDataType(schema.json()) 

self._jreader = self._jreader.schema(jschema) 

102 ↛ 105line 102 didn't jump to line 105, because the condition on line 102 was never false elif isinstance(schema, str): 

self._jreader = self._jreader.schema(schema) 

else: 

raise TypeError("schema should be StructType or string") 

return self 

@since(1.5) 

def option(self, key, value): 

"""Adds an input option for the underlying data source. 

""" 

self._jreader = self._jreader.option(key, to_str(value)) 

return self 

@since(1.4) 

def options(self, **options): 

"""Adds input options for the underlying data source. 

""" 

for k in options: 

self._jreader = self._jreader.option(k, to_str(options[k])) 

return self 

def load(self, path=None, format=None, schema=None, **options): 

"""Loads data from a data source and returns it as a :class:`DataFrame`. 

.. versionadded:: 1.4.0 

Parameters 

---------- 

path : str or list, optional 

optional string or a list of string for file-system backed data sources. 

format : str, optional 

optional string for format of the data source. Default to 'parquet'. 

schema : :class:`pyspark.sql.types.StructType` or str, optional 

optional :class:`pyspark.sql.types.StructType` for the input schema 

or a DDL-formatted string (For example ``col0 INT, col1 DOUBLE``). 

**options : dict 

all other string options 

Examples 

-------- 

>>> df = spark.read.format("parquet").load('python/test_support/sql/parquet_partitioned', 

... opt1=True, opt2=1, opt3='str') 

>>> df.dtypes 

[('name', 'string'), ('year', 'int'), ('month', 'int'), ('day', 'int')] 

>>> df = spark.read.format('json').load(['python/test_support/sql/people.json', 

... 'python/test_support/sql/people1.json']) 

>>> df.dtypes 

[('age', 'bigint'), ('aka', 'string'), ('name', 'string')] 

""" 

if format is not None: 

self.format(format) 

154 ↛ 155line 154 didn't jump to line 155, because the condition on line 154 was never true if schema is not None: 

self.schema(schema) 

self.options(**options) 

if isinstance(path, str): 

return self._df(self._jreader.load(path)) 

elif path is not None: 

160 ↛ 161line 160 didn't jump to line 161, because the condition on line 160 was never true if type(path) != list: 

path = [path] 

return self._df(self._jreader.load(self._spark._sc._jvm.PythonUtils.toSeq(path))) 

else: 

return self._df(self._jreader.load()) 

def json(self, path, schema=None, primitivesAsString=None, prefersDecimal=None, 

allowComments=None, allowUnquotedFieldNames=None, allowSingleQuotes=None, 

allowNumericLeadingZero=None, allowBackslashEscapingAnyCharacter=None, 

mode=None, columnNameOfCorruptRecord=None, dateFormat=None, timestampFormat=None, 

multiLine=None, allowUnquotedControlChars=None, lineSep=None, samplingRatio=None, 

dropFieldIfAllNull=None, encoding=None, locale=None, pathGlobFilter=None, 

recursiveFileLookup=None, allowNonNumericNumbers=None, 

modifiedBefore=None, modifiedAfter=None): 

""" 

Loads JSON files and returns the results as a :class:`DataFrame`. 

`JSON Lines <http://jsonlines.org/>`_ (newline-delimited JSON) is supported by default. 

For JSON (one record per file), set the ``multiLine`` parameter to ``true``. 

If the ``schema`` parameter is not specified, this function goes 

through the input once to determine the input schema. 

.. versionadded:: 1.4.0 

Parameters 

---------- 

path : str, list or :class:`RDD` 

string represents path to the JSON dataset, or a list of paths, 

or RDD of Strings storing JSON objects. 

schema : :class:`pyspark.sql.types.StructType` or str, optional 

an optional :class:`pyspark.sql.types.StructType` for the input schema or 

a DDL-formatted string (For example ``col0 INT, col1 DOUBLE``). 

Other Parameters 

---------------- 

Extra options 

For the extra options, refer to 

`Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-json.html#data-source-option>`_ 

in the version you use. 

.. # noqa 

Examples 

-------- 

>>> df1 = spark.read.json('python/test_support/sql/people.json') 

>>> df1.dtypes 

[('age', 'bigint'), ('name', 'string')] 

>>> rdd = sc.textFile('python/test_support/sql/people.json') 

>>> df2 = spark.read.json(rdd) 

>>> df2.dtypes 

[('age', 'bigint'), ('name', 'string')] 

""" 

self._set_opts( 

schema=schema, primitivesAsString=primitivesAsString, prefersDecimal=prefersDecimal, 

allowComments=allowComments, allowUnquotedFieldNames=allowUnquotedFieldNames, 

allowSingleQuotes=allowSingleQuotes, allowNumericLeadingZero=allowNumericLeadingZero, 

allowBackslashEscapingAnyCharacter=allowBackslashEscapingAnyCharacter, 

mode=mode, columnNameOfCorruptRecord=columnNameOfCorruptRecord, dateFormat=dateFormat, 

timestampFormat=timestampFormat, multiLine=multiLine, 

allowUnquotedControlChars=allowUnquotedControlChars, lineSep=lineSep, 

samplingRatio=samplingRatio, dropFieldIfAllNull=dropFieldIfAllNull, encoding=encoding, 

locale=locale, pathGlobFilter=pathGlobFilter, recursiveFileLookup=recursiveFileLookup, 

modifiedBefore=modifiedBefore, modifiedAfter=modifiedAfter, 

allowNonNumericNumbers=allowNonNumericNumbers) 

if isinstance(path, str): 

path = [path] 

if type(path) == list: 

return self._df(self._jreader.json(self._spark._sc._jvm.PythonUtils.toSeq(path))) 

230 ↛ 243line 230 didn't jump to line 243, because the condition on line 230 was never false elif isinstance(path, RDD): 

def func(iterator): 

for x in iterator: 

if not isinstance(x, str): 

x = str(x) 

235 ↛ 237line 235 didn't jump to line 237, because the condition on line 235 was never false if isinstance(x, str): 

x = x.encode("utf-8") 

yield x 

keyed = path.mapPartitions(func) 

keyed._bypass_serializer = True 

jrdd = keyed._jrdd.map(self._spark._jvm.BytesToString()) 

return self._df(self._jreader.json(jrdd)) 

else: 

raise TypeError("path can be only string, list or RDD") 

def table(self, tableName): 

"""Returns the specified table as a :class:`DataFrame`. 

.. versionadded:: 1.4.0 

Parameters 

---------- 

tableName : str 

string, name of the table. 

Examples 

-------- 

>>> df = spark.read.parquet('python/test_support/sql/parquet_partitioned') 

>>> df.createOrReplaceTempView('tmpTable') 

>>> spark.read.table('tmpTable').dtypes 

[('name', 'string'), ('year', 'int'), ('month', 'int'), ('day', 'int')] 

""" 

return self._df(self._jreader.table(tableName)) 

def parquet(self, *paths, **options): 

""" 

Loads Parquet files, returning the result as a :class:`DataFrame`. 

.. versionadded:: 1.4.0 

Parameters 

---------- 

paths : str 

Other Parameters 

---------------- 

**options 

For the extra options, refer to 

`Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-parquet.html#data-source-option>`_ 

in the version you use. 

.. # noqa 

Examples 

-------- 

>>> df = spark.read.parquet('python/test_support/sql/parquet_partitioned') 

>>> df.dtypes 

[('name', 'string'), ('year', 'int'), ('month', 'int'), ('day', 'int')] 

""" 

mergeSchema = options.get('mergeSchema', None) 

pathGlobFilter = options.get('pathGlobFilter', None) 

modifiedBefore = options.get('modifiedBefore', None) 

modifiedAfter = options.get('modifiedAfter', None) 

recursiveFileLookup = options.get('recursiveFileLookup', None) 

datetimeRebaseMode = options.get('datetimeRebaseMode', None) 

int96RebaseMode = options.get('int96RebaseMode', None) 

self._set_opts(mergeSchema=mergeSchema, pathGlobFilter=pathGlobFilter, 

recursiveFileLookup=recursiveFileLookup, modifiedBefore=modifiedBefore, 

modifiedAfter=modifiedAfter, datetimeRebaseMode=datetimeRebaseMode, 

int96RebaseMode=int96RebaseMode) 

return self._df(self._jreader.parquet(_to_seq(self._spark._sc, paths))) 

def text(self, paths, wholetext=False, lineSep=None, pathGlobFilter=None, 

recursiveFileLookup=None, modifiedBefore=None, 

modifiedAfter=None): 

""" 

Loads text files and returns a :class:`DataFrame` whose schema starts with a 

string column named "value", and followed by partitioned columns if there 

are any. 

The text files must be encoded as UTF-8. 

By default, each line in the text file is a new row in the resulting DataFrame. 

.. versionadded:: 1.6.0 

Parameters 

---------- 

paths : str or list 

string, or list of strings, for input path(s). 

Other Parameters 

---------------- 

Extra options 

For the extra options, refer to 

`Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-text.html#data-source-option>`_ 

in the version you use. 

.. # noqa 

Examples 

-------- 

>>> df = spark.read.text('python/test_support/sql/text-test.txt') 

>>> df.collect() 

[Row(value='hello'), Row(value='this')] 

>>> df = spark.read.text('python/test_support/sql/text-test.txt', wholetext=True) 

>>> df.collect() 

[Row(value='hello\\nthis')] 

""" 

self._set_opts( 

wholetext=wholetext, lineSep=lineSep, pathGlobFilter=pathGlobFilter, 

recursiveFileLookup=recursiveFileLookup, modifiedBefore=modifiedBefore, 

modifiedAfter=modifiedAfter) 

if isinstance(paths, str): 

paths = [paths] 

return self._df(self._jreader.text(self._spark._sc._jvm.PythonUtils.toSeq(paths))) 

def csv(self, path, schema=None, sep=None, encoding=None, quote=None, escape=None, 

comment=None, header=None, inferSchema=None, ignoreLeadingWhiteSpace=None, 

ignoreTrailingWhiteSpace=None, nullValue=None, nanValue=None, positiveInf=None, 

negativeInf=None, dateFormat=None, timestampFormat=None, maxColumns=None, 

maxCharsPerColumn=None, maxMalformedLogPerPartition=None, mode=None, 

columnNameOfCorruptRecord=None, multiLine=None, charToEscapeQuoteEscaping=None, 

samplingRatio=None, enforceSchema=None, emptyValue=None, locale=None, lineSep=None, 

pathGlobFilter=None, recursiveFileLookup=None, modifiedBefore=None, modifiedAfter=None, 

unescapedQuoteHandling=None): 

r"""Loads a CSV file and returns the result as a :class:`DataFrame`. 

This function will go through the input once to determine the input schema if 

``inferSchema`` is enabled. To avoid going through the entire data once, disable 

``inferSchema`` option or specify the schema explicitly using ``schema``. 

.. versionadded:: 2.0.0 

Parameters 

---------- 

path : str or list 

string, or list of strings, for input path(s), 

or RDD of Strings storing CSV rows. 

schema : :class:`pyspark.sql.types.StructType` or str, optional 

an optional :class:`pyspark.sql.types.StructType` for the input schema 

or a DDL-formatted string (For example ``col0 INT, col1 DOUBLE``). 

Other Parameters 

---------------- 

Extra options 

For the extra options, refer to 

`Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-csv.html#data-source-option>`_ 

in the version you use. 

.. # noqa 

Examples 

-------- 

>>> df = spark.read.csv('python/test_support/sql/ages.csv') 

>>> df.dtypes 

[('_c0', 'string'), ('_c1', 'string')] 

>>> rdd = sc.textFile('python/test_support/sql/ages.csv') 

>>> df2 = spark.read.csv(rdd) 

>>> df2.dtypes 

[('_c0', 'string'), ('_c1', 'string')] 

""" 

self._set_opts( 

schema=schema, sep=sep, encoding=encoding, quote=quote, escape=escape, comment=comment, 

header=header, inferSchema=inferSchema, ignoreLeadingWhiteSpace=ignoreLeadingWhiteSpace, 

ignoreTrailingWhiteSpace=ignoreTrailingWhiteSpace, nullValue=nullValue, 

nanValue=nanValue, positiveInf=positiveInf, negativeInf=negativeInf, 

dateFormat=dateFormat, timestampFormat=timestampFormat, maxColumns=maxColumns, 

maxCharsPerColumn=maxCharsPerColumn, 

maxMalformedLogPerPartition=maxMalformedLogPerPartition, mode=mode, 

columnNameOfCorruptRecord=columnNameOfCorruptRecord, multiLine=multiLine, 

charToEscapeQuoteEscaping=charToEscapeQuoteEscaping, samplingRatio=samplingRatio, 

enforceSchema=enforceSchema, emptyValue=emptyValue, locale=locale, lineSep=lineSep, 

pathGlobFilter=pathGlobFilter, recursiveFileLookup=recursiveFileLookup, 

modifiedBefore=modifiedBefore, modifiedAfter=modifiedAfter, 

unescapedQuoteHandling=unescapedQuoteHandling) 

if isinstance(path, str): 

path = [path] 

if type(path) == list: 

return self._df(self._jreader.csv(self._spark._sc._jvm.PythonUtils.toSeq(path))) 

411 ↛ 431line 411 didn't jump to line 431, because the condition on line 411 was never false elif isinstance(path, RDD): 

def func(iterator): 

for x in iterator: 

414 ↛ 415line 414 didn't jump to line 415, because the condition on line 414 was never true if not isinstance(x, str): 

x = str(x) 

416 ↛ 418line 416 didn't jump to line 418, because the condition on line 416 was never false if isinstance(x, str): 

x = x.encode("utf-8") 

yield x 

keyed = path.mapPartitions(func) 

keyed._bypass_serializer = True 

jrdd = keyed._jrdd.map(self._spark._jvm.BytesToString()) 

# see SPARK-22112 

# There aren't any jvm api for creating a dataframe from rdd storing csv. 

# We can do it through creating a jvm dataset firstly and using the jvm api 

# for creating a dataframe from dataset storing csv. 

jdataset = self._spark._ssql_ctx.createDataset( 

jrdd.rdd(), 

self._spark._jvm.Encoders.STRING()) 

return self._df(self._jreader.csv(jdataset)) 

else: 

raise TypeError("path can be only string, list or RDD") 

def orc(self, path, mergeSchema=None, pathGlobFilter=None, recursiveFileLookup=None, 

modifiedBefore=None, modifiedAfter=None): 

"""Loads ORC files, returning the result as a :class:`DataFrame`. 

.. versionadded:: 1.5.0 

Parameters 

---------- 

path : str or list 

Other Parameters 

---------------- 

Extra options 

For the extra options, refer to 

`Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-orc.html#data-source-option>`_ 

in the version you use. 

.. # noqa 

Examples 

-------- 

>>> df = spark.read.orc('python/test_support/sql/orc_partitioned') 

>>> df.dtypes 

[('a', 'bigint'), ('b', 'int'), ('c', 'int')] 

""" 

self._set_opts(mergeSchema=mergeSchema, pathGlobFilter=pathGlobFilter, 

modifiedBefore=modifiedBefore, modifiedAfter=modifiedAfter, 

recursiveFileLookup=recursiveFileLookup) 

if isinstance(path, str): 

path = [path] 

return self._df(self._jreader.orc(_to_seq(self._spark._sc, path))) 

def jdbc(self, url, table, column=None, lowerBound=None, upperBound=None, numPartitions=None, 

predicates=None, properties=None): 

""" 

Construct a :class:`DataFrame` representing the database table named ``table`` 

accessible via JDBC URL ``url`` and connection ``properties``. 

Partitions of the table will be retrieved in parallel if either ``column`` or 

``predicates`` is specified. ``lowerBound``, ``upperBound`` and ``numPartitions`` 

is needed when ``column`` is specified. 

If both ``column`` and ``predicates`` are specified, ``column`` will be used. 

.. versionadded:: 1.4.0 

Parameters 

---------- 

table : str 

the name of the table 

column : str, optional 

alias of ``partitionColumn`` option. Refer to ``partitionColumn`` in 

`Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-jdbc.html#data-source-option>`_ 

in the version you use. 

predicates : list, optional 

a list of expressions suitable for inclusion in WHERE clauses; 

each one defines one partition of the :class:`DataFrame` 

properties : dict, optional 

a dictionary of JDBC database connection arguments. Normally at 

least properties "user" and "password" with their corresponding values. 

For example { 'user' : 'SYSTEM', 'password' : 'mypassword' } 

Other Parameters 

---------------- 

Extra options 

For the extra options, refer to 

`Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-jdbc.html#data-source-option>`_ 

in the version you use. 

.. # noqa 

Notes 

----- 

Don't create too many partitions in parallel on a large cluster; 

otherwise Spark might crash your external database systems. 

Returns 

------- 

:class:`DataFrame` 

""" 

if properties is None: 

properties = dict() 

jprop = JavaClass("java.util.Properties", self._spark._sc._gateway._gateway_client)() 

for k in properties: 

jprop.setProperty(k, properties[k]) 

if column is not None: 

assert lowerBound is not None, "lowerBound can not be None when ``column`` is specified" 

assert upperBound is not None, "upperBound can not be None when ``column`` is specified" 

assert numPartitions is not None, \ 

"numPartitions can not be None when ``column`` is specified" 

return self._df(self._jreader.jdbc(url, table, column, int(lowerBound), int(upperBound), 

int(numPartitions), jprop)) 

if predicates is not None: 

gateway = self._spark._sc._gateway 

jpredicates = utils.toJArray(gateway, gateway.jvm.java.lang.String, predicates) 

return self._df(self._jreader.jdbc(url, table, jpredicates, jprop)) 

return self._df(self._jreader.jdbc(url, table, jprop)) 

class DataFrameWriter(OptionUtils): 

""" 

Interface used to write a :class:`DataFrame` to external storage systems 

(e.g. file systems, key-value stores, etc). Use :attr:`DataFrame.write` 

to access this. 

.. versionadded:: 1.4 

""" 

def __init__(self, df): 

self._df = df 

self._spark = df.sql_ctx 

self._jwrite = df._jdf.write() 

def _sq(self, jsq): 

from pyspark.sql.streaming import StreamingQuery 

return StreamingQuery(jsq) 

def mode(self, saveMode): 

"""Specifies the behavior when data or table already exists. 

Options include: 

* `append`: Append contents of this :class:`DataFrame` to existing data. 

* `overwrite`: Overwrite existing data. 

* `error` or `errorifexists`: Throw an exception if data already exists. 

* `ignore`: Silently ignore this operation if data already exists. 

.. versionadded:: 1.4.0 

Examples 

-------- 

>>> df.write.mode('append').parquet(os.path.join(tempfile.mkdtemp(), 'data')) 

""" 

# At the JVM side, the default value of mode is already set to "error". 

# So, if the given saveMode is None, we will not call JVM-side's mode method. 

if saveMode is not None: 

self._jwrite = self._jwrite.mode(saveMode) 

return self 

def format(self, source): 

"""Specifies the underlying output data source. 

.. versionadded:: 1.4.0 

Parameters 

---------- 

source : str 

string, name of the data source, e.g. 'json', 'parquet'. 

Examples 

-------- 

>>> df.write.format('json').save(os.path.join(tempfile.mkdtemp(), 'data')) 

""" 

self._jwrite = self._jwrite.format(source) 

return self 

@since(1.5) 

def option(self, key, value): 

"""Adds an output option for the underlying data source. 

""" 

self._jwrite = self._jwrite.option(key, to_str(value)) 

return self 

@since(1.4) 

def options(self, **options): 

"""Adds output options for the underlying data source. 

""" 

for k in options: 

self._jwrite = self._jwrite.option(k, to_str(options[k])) 

return self 

def partitionBy(self, *cols): 

"""Partitions the output by the given columns on the file system. 

If specified, the output is laid out on the file system similar 

to Hive's partitioning scheme. 

.. versionadded:: 1.4.0 

Parameters 

---------- 

cols : str or list 

name of columns 

Examples 

-------- 

>>> df.write.partitionBy('year', 'month').parquet(os.path.join(tempfile.mkdtemp(), 'data')) 

""" 

620 ↛ 621line 620 didn't jump to line 621, because the condition on line 620 was never true if len(cols) == 1 and isinstance(cols[0], (list, tuple)): 

cols = cols[0] 

self._jwrite = self._jwrite.partitionBy(_to_seq(self._spark._sc, cols)) 

return self 

def bucketBy(self, numBuckets, col, *cols): 

"""Buckets the output by the given columns. If specified, 

the output is laid out on the file system similar to Hive's bucketing scheme, 

but with a different bucket hash function and is not compatible with Hive's bucketing. 

.. versionadded:: 2.3.0 

Parameters 

---------- 

numBuckets : int 

the number of buckets to save 

col : str, list or tuple 

a name of a column, or a list of names. 

cols : str 

additional names (optional). If `col` is a list it should be empty. 

Notes 

----- 

Applicable for file-based data sources in combination with 

:py:meth:`DataFrameWriter.saveAsTable`. 

Examples 

-------- 

>>> (df.write.format('parquet') # doctest: +SKIP 

... .bucketBy(100, 'year', 'month') 

... .mode("overwrite") 

... .saveAsTable('bucketed_table')) 

""" 

653 ↛ 654line 653 didn't jump to line 654, because the condition on line 653 was never true if not isinstance(numBuckets, int): 

raise TypeError("numBuckets should be an int, got {0}.".format(type(numBuckets))) 

if isinstance(col, (list, tuple)): 

657 ↛ 658line 657 didn't jump to line 658, because the condition on line 657 was never true if cols: 

raise ValueError("col is a {0} but cols are not empty".format(type(col))) 

col, cols = col[0], col[1:] 

662 ↛ 663line 662 didn't jump to line 663, because the condition on line 662 was never true if not all(isinstance(c, str) for c in cols) or not(isinstance(col, str)): 

raise TypeError("all names should be `str`") 

self._jwrite = self._jwrite.bucketBy(numBuckets, col, _to_seq(self._spark._sc, cols)) 

return self 

def sortBy(self, col, *cols): 

"""Sorts the output in each bucket by the given columns on the file system. 

.. versionadded:: 2.3.0 

Parameters 

---------- 

col : str, tuple or list 

a name of a column, or a list of names. 

cols : str 

additional names (optional). If `col` is a list it should be empty. 

Examples 

-------- 

>>> (df.write.format('parquet') # doctest: +SKIP 

... .bucketBy(100, 'year', 'month') 

... .sortBy('day') 

... .mode("overwrite") 

... .saveAsTable('sorted_bucketed_table')) 

""" 

if isinstance(col, (list, tuple)): 

689 ↛ 690line 689 didn't jump to line 690, because the condition on line 689 was never true if cols: 

raise ValueError("col is a {0} but cols are not empty".format(type(col))) 

col, cols = col[0], col[1:] 

694 ↛ 695line 694 didn't jump to line 695, because the condition on line 694 was never true if not all(isinstance(c, str) for c in cols) or not(isinstance(col, str)): 

raise TypeError("all names should be `str`") 

self._jwrite = self._jwrite.sortBy(col, _to_seq(self._spark._sc, cols)) 

return self 

def save(self, path=None, format=None, mode=None, partitionBy=None, **options): 

"""Saves the contents of the :class:`DataFrame` to a data source. 

The data source is specified by the ``format`` and a set of ``options``. 

If ``format`` is not specified, the default data source configured by 

``spark.sql.sources.default`` will be used. 

.. versionadded:: 1.4.0 

Parameters 

---------- 

path : str, optional 

the path in a Hadoop supported file system 

format : str, optional 

the format used to save 

mode : str, optional 

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`` or ``errorifexists`` (default case): Throw an exception if data already \ 

exists. 

partitionBy : list, optional 

names of partitioning columns 

**options : dict 

all other string options 

Examples 

-------- 

>>> df.write.mode("append").save(os.path.join(tempfile.mkdtemp(), 'data')) 

""" 

self.mode(mode).options(**options) 

733 ↛ 734line 733 didn't jump to line 734, because the condition on line 733 was never true if partitionBy is not None: 

self.partitionBy(partitionBy) 

if format is not None: 

self.format(format) 

737 ↛ 738line 737 didn't jump to line 738, because the condition on line 737 was never true if path is None: 

self._jwrite.save() 

else: 

self._jwrite.save(path) 

@since(1.4) 

def insertInto(self, tableName, overwrite=None): 

"""Inserts the content of the :class:`DataFrame` to the specified table. 

It requires that the schema of the :class:`DataFrame` is the same as the 

schema of the table. 

Parameters 

---------- 

overwrite : bool, optional 

If true, overwrites existing data. Disabled by default 

Notes 

----- 

Unlike :meth:`DataFrameWriter.saveAsTable`, :meth:`DataFrameWriter.insertInto` ignores 

the column names and just uses position-based resolution. 

""" 

if overwrite is not None: 

self.mode("overwrite" if overwrite else "append") 

self._jwrite.insertInto(tableName) 

def saveAsTable(self, name, format=None, mode=None, partitionBy=None, **options): 

"""Saves the content of the :class:`DataFrame` as the specified table. 

In the case the table already exists, behavior of this function depends on the 

save mode, specified by the `mode` function (default to throwing an exception). 

When `mode` is `Overwrite`, the schema of the :class:`DataFrame` does not need to be 

the same as that of the existing table. 

* `append`: Append contents of this :class:`DataFrame` to existing data. 

* `overwrite`: Overwrite existing data. 

* `error` or `errorifexists`: Throw an exception if data already exists. 

* `ignore`: Silently ignore this operation if data already exists. 

.. versionadded:: 1.4.0 

Notes 

----- 

When `mode` is `Append`, if there is an existing table, we will use the format and 

options of the existing table. The column order in the schema of the :class:`DataFrame` 

doesn't need to be same as that of the existing table. Unlike 

:meth:`DataFrameWriter.insertInto`, :meth:`DataFrameWriter.saveAsTable` will use the 

column names to find the correct column positions. 

Parameters 

---------- 

name : str 

the table name 

format : str, optional 

the format used to save 

mode : str, optional 

one of `append`, `overwrite`, `error`, `errorifexists`, `ignore` \ 

(default: error) 

partitionBy : str or list 

names of partitioning columns 

**options : dict 

all other string options 

""" 

self.mode(mode).options(**options) 

802 ↛ 803line 802 didn't jump to line 803, because the condition on line 802 was never true if partitionBy is not None: 

self.partitionBy(partitionBy) 

if format is not None: 

self.format(format) 

self._jwrite.saveAsTable(name) 

def json(self, path, mode=None, compression=None, dateFormat=None, timestampFormat=None, 

lineSep=None, encoding=None, ignoreNullFields=None): 

"""Saves the content of the :class:`DataFrame` in JSON format 

(`JSON Lines text format or newline-delimited JSON <http://jsonlines.org/>`_) at the 

specified path. 

.. versionadded:: 1.4.0 

Parameters 

---------- 

path : str 

the path in any Hadoop supported file system 

mode : str, optional 

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`` or ``errorifexists`` (default case): Throw an exception if data already \ 

exists. 

Other Parameters 

---------------- 

Extra options 

For the extra options, refer to 

`Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-json.html#data-source-option>`_ 

in the version you use. 

.. # noqa 

Examples 

-------- 

>>> df.write.json(os.path.join(tempfile.mkdtemp(), 'data')) 

""" 

self.mode(mode) 

self._set_opts( 

compression=compression, dateFormat=dateFormat, timestampFormat=timestampFormat, 

lineSep=lineSep, encoding=encoding, ignoreNullFields=ignoreNullFields) 

self._jwrite.json(path) 

def parquet(self, path, mode=None, partitionBy=None, compression=None): 

"""Saves the content of the :class:`DataFrame` in Parquet format at the specified path. 

.. versionadded:: 1.4.0 

Parameters 

---------- 

path : str 

the path in any Hadoop supported file system 

mode : str, optional 

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`` or ``errorifexists`` (default case): Throw an exception if data already \ 

exists. 

partitionBy : str or list, optional 

names of partitioning columns 

Other Parameters 

---------------- 

Extra options 

For the extra options, refer to 

`Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-parquet.html#data-source-option>`_ 

in the version you use. 

.. # noqa 

Examples 

-------- 

>>> df.write.parquet(os.path.join(tempfile.mkdtemp(), 'data')) 

""" 

self.mode(mode) 

882 ↛ 883line 882 didn't jump to line 883, because the condition on line 882 was never true if partitionBy is not None: 

self.partitionBy(partitionBy) 

self._set_opts(compression=compression) 

self._jwrite.parquet(path) 

def text(self, path, compression=None, lineSep=None): 

"""Saves the content of the DataFrame in a text file at the specified path. 

The text files will be encoded as UTF-8. 

.. versionadded:: 1.6.0 

Parameters 

---------- 

path : str 

the path in any Hadoop supported file system 

Other Parameters 

---------------- 

Extra options 

For the extra options, refer to 

`Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-text.html#data-source-option>`_ 

in the version you use. 

.. # noqa 

The DataFrame must have only one column that is of string type. 

Each row becomes a new line in the output file. 

""" 

self._set_opts(compression=compression, lineSep=lineSep) 

self._jwrite.text(path) 

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, encoding=None, emptyValue=None, lineSep=None): 

r"""Saves the content of the :class:`DataFrame` in CSV format at the specified path. 

.. versionadded:: 2.0.0 

Parameters 

---------- 

path : str 

the path in any Hadoop supported file system 

mode : str, optional 

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`` or ``errorifexists`` (default case): Throw an exception if data already \ 

exists. 

Other Parameters 

---------------- 

Extra options 

For the extra options, refer to 

`Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-csv.html#data-source-option>`_ 

in the version you use. 

.. # noqa 

Examples 

-------- 

>>> df.write.csv(os.path.join(tempfile.mkdtemp(), 'data')) 

""" 

self.mode(mode) 

self._set_opts(compression=compression, sep=sep, quote=quote, escape=escape, header=header, 

nullValue=nullValue, escapeQuotes=escapeQuotes, quoteAll=quoteAll, 

dateFormat=dateFormat, timestampFormat=timestampFormat, 

ignoreLeadingWhiteSpace=ignoreLeadingWhiteSpace, 

ignoreTrailingWhiteSpace=ignoreTrailingWhiteSpace, 

charToEscapeQuoteEscaping=charToEscapeQuoteEscaping, 

encoding=encoding, emptyValue=emptyValue, lineSep=lineSep) 

self._jwrite.csv(path) 

def orc(self, path, mode=None, partitionBy=None, compression=None): 

"""Saves the content of the :class:`DataFrame` in ORC format at the specified path. 

.. versionadded:: 1.5.0 

Parameters 

---------- 

path : str 

the path in any Hadoop supported file system 

mode : str, optional 

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`` or ``errorifexists`` (default case): Throw an exception if data already \ 

exists. 

partitionBy : str or list, optional 

names of partitioning columns 

Other Parameters 

---------------- 

Extra options 

For the extra options, refer to 

`Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-orc.html#data-source-option>`_ 

in the version you use. 

.. # noqa 

Examples 

-------- 

>>> orc_df = spark.read.orc('python/test_support/sql/orc_partitioned') 

>>> orc_df.write.orc(os.path.join(tempfile.mkdtemp(), 'data')) 

""" 

self.mode(mode) 

992 ↛ 993line 992 didn't jump to line 993, because the condition on line 992 was never true if partitionBy is not None: 

self.partitionBy(partitionBy) 

self._set_opts(compression=compression) 

self._jwrite.orc(path) 

def jdbc(self, url, table, mode=None, properties=None): 

"""Saves the content of the :class:`DataFrame` to an external database table via JDBC. 

.. versionadded:: 1.4.0 

Parameters 

---------- 

table : str 

Name of the table in the external database. 

mode : str, optional 

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`` or ``errorifexists`` (default case): Throw an exception if data already \ 

exists. 

properties : dict 

a dictionary of JDBC database connection arguments. Normally at