diff options
author | Reynold Xin <rxin@databricks.com> | 2016-04-26 10:58:56 -0700 |
---|---|---|
committer | Yin Huai <yhuai@databricks.com> | 2016-04-26 10:58:56 -0700 |
commit | 5cb03220a02c70d343e82d69cfd30edb894595a1 (patch) | |
tree | 41232454ab88f2c9c1feadd13d43b8cc46873cf5 /mllib | |
parent | 92f66331b4ba3634f54f57ddb5e7962b14aa4ca1 (diff) | |
download | spark-5cb03220a02c70d343e82d69cfd30edb894595a1.tar.gz spark-5cb03220a02c70d343e82d69cfd30edb894595a1.tar.bz2 spark-5cb03220a02c70d343e82d69cfd30edb894595a1.zip |
[SPARK-14912][SQL] Propagate data source options to Hadoop configuration
## What changes were proposed in this pull request?
We currently have no way for users to propagate options to the underlying library that rely in Hadoop configurations to work. For example, there are various options in parquet-mr that users might want to set, but the data source API does not expose a per-job way to set it. This patch propagates the user-specified options also into Hadoop Configuration.
## How was this patch tested?
Used a mock data source implementation to test both the read path and the write path.
Author: Reynold Xin <rxin@databricks.com>
Closes #12688 from rxin/SPARK-14912.
Diffstat (limited to 'mllib')
-rw-r--r-- | mllib/src/main/scala/org/apache/spark/ml/source/libsvm/LibSVMRelation.scala | 10 |
1 files changed, 5 insertions, 5 deletions
diff --git a/mllib/src/main/scala/org/apache/spark/ml/source/libsvm/LibSVMRelation.scala b/mllib/src/main/scala/org/apache/spark/ml/source/libsvm/LibSVMRelation.scala index f07374ac0c..ba2e1e2bc2 100644 --- a/mllib/src/main/scala/org/apache/spark/ml/source/libsvm/LibSVMRelation.scala +++ b/mllib/src/main/scala/org/apache/spark/ml/source/libsvm/LibSVMRelation.scala @@ -181,20 +181,20 @@ class DefaultSource extends FileFormat with DataSourceRegister { partitionSchema: StructType, requiredSchema: StructType, filters: Seq[Filter], - options: Map[String, String]): (PartitionedFile) => Iterator[InternalRow] = { + options: Map[String, String], + hadoopConf: Configuration): (PartitionedFile) => Iterator[InternalRow] = { verifySchema(dataSchema) val numFeatures = options("numFeatures").toInt assert(numFeatures > 0) val sparse = options.getOrElse("vectorType", "sparse") == "sparse" - val broadcastedConf = sparkSession.sparkContext.broadcast( - new SerializableConfiguration( - new Configuration(sparkSession.sparkContext.hadoopConfiguration))) + val broadcastedHadoopConf = + sparkSession.sparkContext.broadcast(new SerializableConfiguration(hadoopConf)) (file: PartitionedFile) => { val points = - new HadoopFileLinesReader(file, broadcastedConf.value.value) + new HadoopFileLinesReader(file, broadcastedHadoopConf.value.value) .map(_.toString.trim) .filterNot(line => line.isEmpty || line.startsWith("#")) .map { line => |