You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by bl...@apache.org on 2019/08/23 18:32:25 UTC
[incubator-iceberg] branch master updated: Propagate Hadoop config
in SparkTableUtil (#402)
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new d071be7 Propagate Hadoop config in SparkTableUtil (#402)
d071be7 is described below
commit d071be7f974558e06d1b6e0565c65c2f98d47e96
Author: yathindranath <ya...@gmail.com>
AuthorDate: Fri Aug 23 11:32:21 2019 -0700
Propagate Hadoop config in SparkTableUtil (#402)
---
.../org/apache/iceberg/spark/SparkTableUtil.scala | 19 ++++++++++---------
1 file changed, 10 insertions(+), 9 deletions(-)
diff --git a/spark/src/main/scala/org/apache/iceberg/spark/SparkTableUtil.scala b/spark/src/main/scala/org/apache/iceberg/spark/SparkTableUtil.scala
index 3409807..3ad58af 100644
--- a/spark/src/main/scala/org/apache/iceberg/spark/SparkTableUtil.scala
+++ b/spark/src/main/scala/org/apache/iceberg/spark/SparkTableUtil.scala
@@ -90,13 +90,14 @@ object SparkTableUtil {
def listPartition(
partition: Map[String, String],
uri: String,
- format: String): Seq[SparkDataFile] = {
+ format: String,
+ conf: Configuration = new Configuration()): Seq[SparkDataFile] = {
if (format.contains("avro")) {
- listAvroPartition(partition, uri)
+ listAvroPartition(partition, uri, conf)
} else if (format.contains("parquet")) {
- listParquetPartition(partition, uri)
+ listParquetPartition(partition, uri, conf)
} else if (format.contains("orc")) {
- listOrcPartition(partition, uri)
+ listOrcPartition(partition, uri, conf)
} else {
throw new UnsupportedOperationException(s"Unknown partition format: $format")
}
@@ -230,8 +231,8 @@ object SparkTableUtil {
private def listAvroPartition(
partitionPath: Map[String, String],
- partitionUri: String): Seq[SparkDataFile] = {
- val conf = new Configuration()
+ partitionUri: String,
+ conf: Configuration): Seq[SparkDataFile] = {
val partition = new Path(partitionUri)
val fs = partition.getFileSystem(conf)
@@ -253,8 +254,8 @@ object SparkTableUtil {
private def listParquetPartition(
partitionPath: Map[String, String],
partitionUri: String,
+ conf: Configuration,
metricsSpec: MetricsConfig = MetricsConfig.getDefault): Seq[SparkDataFile] = {
- val conf = new Configuration()
val partition = new Path(partitionUri)
val fs = partition.getFileSystem(conf)
@@ -276,8 +277,8 @@ object SparkTableUtil {
private def listOrcPartition(
partitionPath: Map[String, String],
- partitionUri: String): Seq[SparkDataFile] = {
- val conf = new Configuration()
+ partitionUri: String,
+ conf: Configuration): Seq[SparkDataFile] = {
val partition = new Path(partitionUri)
val fs = partition.getFileSystem(conf)