diff options
Diffstat (limited to 'python/pyspark')
-rw-r--r-- | python/pyspark/sql/readwriter.py | 44 |
1 files changed, 41 insertions, 3 deletions
diff --git a/python/pyspark/sql/readwriter.py b/python/pyspark/sql/readwriter.py index 882a03090e..dea8bad79e 100644 --- a/python/pyspark/sql/readwriter.py +++ b/python/pyspark/sql/readwriter.py @@ -146,14 +146,28 @@ class DataFrameReader(object): return self._df(self._jreader.table(tableName)) @since(1.4) - def parquet(self, *path): + def parquet(self, *paths): """Loads a Parquet file, returning the result as a :class:`DataFrame`. >>> df = sqlContext.read.parquet('python/test_support/sql/parquet_partitioned') >>> df.dtypes [('name', 'string'), ('year', 'int'), ('month', 'int'), ('day', 'int')] """ - return self._df(self._jreader.parquet(_to_seq(self._sqlContext._sc, path))) + return self._df(self._jreader.parquet(_to_seq(self._sqlContext._sc, paths))) + + @since(1.5) + def orc(self, path): + """ + Loads an ORC file, returning the result as a :class:`DataFrame`. + + ::Note: Currently ORC support is only available together with + :class:`HiveContext`. + + >>> df = hiveContext.read.orc('python/test_support/sql/orc_partitioned') + >>> df.dtypes + [('a', 'bigint'), ('b', 'int'), ('c', 'int')] + """ + return self._df(self._jreader.orc(path)) @since(1.4) def jdbc(self, url, table, column=None, lowerBound=None, upperBound=None, numPartitions=None, @@ -378,6 +392,29 @@ class DataFrameWriter(object): self.partitionBy(partitionBy) self._jwrite.parquet(path) + def orc(self, path, mode=None, partitionBy=None): + """Saves the content of the :class:`DataFrame` in ORC format at the specified path. + + ::Note: Currently ORC support is only available together with + :class:`HiveContext`. + + :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. + :param partitionBy: names of partitioning columns + + >>> orc_df = hiveContext.read.orc('python/test_support/sql/orc_partitioned') + >>> orc_df.write.orc(os.path.join(tempfile.mkdtemp(), 'data')) + """ + self.mode(mode) + if partitionBy is not None: + self.partitionBy(partitionBy) + self._jwrite.orc(path) + @since(1.4) def jdbc(self, url, table, mode=None, properties={}): """Saves the content of the :class:`DataFrame` to a external database table via JDBC. @@ -408,7 +445,7 @@ def _test(): import os import tempfile from pyspark.context import SparkContext - from pyspark.sql import Row, SQLContext + from pyspark.sql import Row, SQLContext, HiveContext import pyspark.sql.readwriter os.chdir(os.environ["SPARK_HOME"]) @@ -420,6 +457,7 @@ def _test(): globs['os'] = os globs['sc'] = sc globs['sqlContext'] = SQLContext(sc) + globs['hiveContext'] = HiveContext(sc) globs['df'] = globs['sqlContext'].read.parquet('python/test_support/sql/parquet_partitioned') (failure_count, test_count) = doctest.testmod( |