ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Merge pull request #782 from WANdisco/master

Browse source 
SHARK-94 Log the files computed by HadoopRDD and NewHadoopRDD
This commit is contained in:
Reynold Xin 2013-08-05 22:33:00 -07:00
parent 1b63dea816 a30866438b
commit d031f73679
2 changed files with 2 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
core/src/main/scala/spark/rdd/HadoopRDD.scala
View file

@ -88,6 +88,7 @@ class HadoopRDD[K, V](
override def compute(theSplit: Partition, context: TaskContext) = new NextIterator[(K, V)] { override def compute(theSplit: Partition, context: TaskContext) = new NextIterator[(K, V)] {
val split = theSplit.asInstanceOf[HadoopPartition] val split = theSplit.asInstanceOf[HadoopPartition]
logInfo("Input split: " + split.inputSplit)
var reader: RecordReader[K, V] = null var reader: RecordReader[K, V] = null
val conf = confBroadcast.value.value val conf = confBroadcast.value.value

1
core/src/main/scala/spark/rdd/NewHadoopRDD.scala
View file

@ -73,6 +73,7 @@ class NewHadoopRDD[K, V](
override def compute(theSplit: Partition, context: TaskContext) = new Iterator[(K, V)] { override def compute(theSplit: Partition, context: TaskContext) = new Iterator[(K, V)] {
val split = theSplit.asInstanceOf[NewHadoopPartition] val split = theSplit.asInstanceOf[NewHadoopPartition]
logInfo("Input split: " + split.serializableHadoopSplit)
val conf = confBroadcast.value.value val conf = confBroadcast.value.value
val attemptId = newTaskAttemptID(jobtrackerId, id, true, split.index, 0) val attemptId = newTaskAttemptID(jobtrackerId, id, true, split.index, 0)
val hadoopAttemptContext = newTaskAttemptContext(conf, attemptId) val hadoopAttemptContext = newTaskAttemptContext(conf, attemptId)