Skip to content

[SPARK-8398][CORE] Hadoop input/output format advanced control #6848

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
koertkuipers wants to merge 24 commits into from
Closed

[SPARK-8398][CORE] Hadoop input/output format advanced control #6848

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
24 commits
Select commit Hold shift + click to select a range
db24636
make hadoop configuration available to user for all hadoop input form…
koertkuipers May 26, 2015
333d943
add JobConf to all RDD saveAs... methods
koertkuipers Jun 3, 2015
1f82a33
actually use conf in saveAsSequenceFile
koertkuipers Jun 16, 2015
135b96e
merge from master
koertkuipers Jun 16, 2015
425a578
expose hadoop Configuration or JobConf for all methods that use hadoo…
koertkuipers Jun 19, 2015
9230543
address issues raised by andrewor14
koertkuipers Jun 19, 2015
2bfa320
merge from master
koertkuipers Jun 19, 2015
2122160
really simple tests that the Configuration provided gets used in (New…
koertkuipers Jun 19, 2015
df2c2ae
fix scalastyle errors
koertkuipers Jun 19, 2015
e2f7023
dont break binary compatibility (make MiMa happy)
koertkuipers Jun 20, 2015
1cbd95a
silence mima for JavaRDDLike.saveAsTextFile and JavaRDDLike.saveAsObj…
koertkuipers Jun 24, 2015
c5cf6b0
merge from master
koertkuipers Jul 2, 2015
96ccee0
merge from master
koertkuipers Jul 2, 2015
3097312
merge from master
koertkuipers Aug 16, 2015
cfe3f0c
merge from master
koertkuipers Sep 28, 2015
7ca662c
move mima excludes to section for right version
koertkuipers Oct 2, 2015
5483148
merge from master
koertkuipers Oct 13, 2015
470b3d9
Merge branch 'master' into feat-hadoop-input-format-advanced-control
koertkuipers Oct 18, 2015
208c019
merge from master
koertkuipers Nov 11, 2015
5e0b89c
merge from master
koertkuipers Apr 20, 2016
0ea4e5c
fix scalastyle errors
koertkuipers Apr 20, 2016
60c34e1
move mima exclusions for SPARK-8398 to version 2.0
koertkuipers Apr 20, 2016
c06548d
merge from master
koertkuipers Apr 20, 2016
34f97d4
merge from master
koertkuipers Apr 21, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
add JobConf to all RDD saveAs... methods
  • Loading branch information
@koertkuipers
koertkuipers committed Jun 3, 2015
commit 333d943b77533fa162a51460820fdac7423a54fb
10 changes: 5 additions & 5 deletions core/src/main/scala/org/apache/spark/rdd/RDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ import scala.reflect.{classTag, ClassTag}
import com.clearspring.analytics.stream.cardinality.HyperLogLogPlus
import org.apache.hadoop.io.{BytesWritable, NullWritable, Text}
import org.apache.hadoop.io.compress.CompressionCodec
import org.apache.hadoop.mapred.TextOutputFormat
import org.apache.hadoop.mapred.{ JobConf, TextOutputFormat }
Copy link
Contributor

@andrewor14 andrewor14 Jun 18, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

remove space around the {}


import org.apache.spark._
import org.apache.spark.Partitioner._
Expand Down Expand Up @@ -1376,7 +1376,7 @@ abstract class RDD[T: ClassTag](
/**
* Save this RDD as a text file, using string representations of elements.
*/
def saveAsTextFile(path: String): Unit = withScope {
def saveAsTextFile(path: String, conf: JobConf = new JobConf(sc.hadoopConfiguration)): Unit = withScope {
// https://issues.apache.org/jira/browse/SPARK-2075
//
// NullWritable is a `Comparable` in Hadoop 1.+, so the compiler cannot find an implicit
Expand All @@ -1397,7 +1397,7 @@ abstract class RDD[T: ClassTag](
}
}
RDD.rddToPairRDDFunctions(r)(nullWritableClassTag, textClassTag, null)
.saveAsHadoopFile[TextOutputFormat[NullWritable, Text]](path)
.saveAsHadoopFile(path, classOf[NullWritable], classOf[Text], classOf[TextOutputFormat[NullWritable, Text]], conf)
Copy link
Contributor

@andrewor14 andrewor14 Jun 18, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this line is almost certainly too long. Can you break it at (?

}

/**
Expand All @@ -1421,10 +1421,10 @@ abstract class RDD[T: ClassTag](
/**
* Save this RDD as a SequenceFile of serialized objects.
*/
def saveAsObjectFile(path: String): Unit = withScope {
def saveAsObjectFile(path: String, conf: JobConf = new JobConf(sc.hadoopConfiguration)): Unit = withScope {
this.mapPartitions(iter => iter.grouped(10).map(_.toArray))
.map(x => (NullWritable.get(), new BytesWritable(Utils.serialize(x))))
.saveAsSequenceFile(path)
.saveAsSequenceFile(path, None, conf)
Copy link
Contributor

@andrewor14 andrewor14 Jun 18, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: I would do saveAsSequenceFile(path, conf = conf) instead

}

/**
Expand Down
3 changes: 2 additions & 1 deletion core/src/main/scala/org/apache/spark/rdd/SequenceFileRDDFunctions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,8 @@ class SequenceFileRDDFunctions[K <% Writable: ClassTag, V <% Writable : ClassTag
*/
def saveAsSequenceFile(
path: String,
codec: Option[Class[_ <: CompressionCodec]] = None): Unit = self.withScope {
codec: Option[Class[_ <: CompressionCodec]] = None,
conf: JobConf = new JobConf(self.context.hadoopConfiguration)): Unit = self.withScope {
def anyToWritable[U <% Writable](u: U): Writable = u

// TODO We cannot force the return type of `anyToWritable` be same as keyWritableClass and
Expand Down