From c31c36c4a76bd3449696383321332ec95bff7fed Mon Sep 17 00:00:00 2001
From: Thomas Graves <tgraves@apache.org>
Date: Mon, 2 Feb 2015 22:45:55 -0800
Subject: [PATCH] [SPARK-3778] newAPIHadoopRDD doesn't properly pass
credentials for secure hdfs
.this was https://github.com/apache/spark/pull/2676
https://issues.apache.org/jira/browse/SPARK-3778
This affects if someone is trying to access secure hdfs something like:
val lines = {
val hconf = new Configuration()
hconf.set("mapred.input.dir", "mydir")
hconf.set("textinputformat.record.delimiter","\003432\n")
sc.newAPIHadoopRDD(hconf, classOf[TextInputFormat], classOf[LongWritable], classOf[Text])
}
Author: Thomas Graves <tgraves@apache.org>
Closes #4292 from tgravescs/SPARK-3788 and squashes the following commits:
cf3b453 [Thomas Graves] newAPIHadoopRDD doesn't properly pass credentials for secure hdfs on yarn
---
core/src/main/scala/org/apache/spark/SparkContext.scala | 7 ++++++-
1 file changed, 6 insertions(+), 1 deletion(-)
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala
index 228076f01c..6a16a31654 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -804,6 +804,8 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
vClass: Class[V],
conf: Configuration = hadoopConfiguration): RDD[(K, V)] = {
assertNotStopped()
+ // The call to new NewHadoopJob automatically adds security credentials to conf,
+ // so we don't need to explicitly add them ourselves
val job = new NewHadoopJob(conf)
NewFileInputFormat.addInputPath(job, new Path(path))
val updatedConf = job.getConfiguration
@@ -826,7 +828,10 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
kClass: Class[K],
vClass: Class[V]): RDD[(K, V)] = {
assertNotStopped()
- new NewHadoopRDD(this, fClass, kClass, vClass, conf)
+ // Add necessary security credentials to the JobConf. Required to access secure HDFS.
+ val jconf = new JobConf(conf)
+ SparkHadoopUtil.get.addCredentials(jconf)
+ new NewHadoopRDD(this, fClass, kClass, vClass, jconf)
}
/** Get an RDD for a Hadoop SequenceFile with given key and value types.
--
GitLab