-
-
- - ID: {app.id}
- - Name: {app.desc.name}
- - User: {app.desc.user}
- - Cores:
- {
- if (app.desc.maxCores.isEmpty) {
- "Unlimited (%s granted)".format(app.coresGranted)
- } else {
- "%s (%s granted, %s left)".format(
- app.desc.maxCores.get, app.coresGranted, app.coresLeft)
- }
- }
-
- -
- Executor Memory:
- {Utils.megabytesToString(app.desc.memoryPerSlave)}
-
- - Submit Date: {app.submitDate}
- - State: {app.state}
- - Application Detail UI
-
-
+
+
+
+ - ID: {app.id}
+ - Name: {app.desc.name}
+ - User: {app.desc.user}
+ - Cores:
+ {
+ if (app.desc.maxCores.isEmpty) {
+ "Unlimited (%s granted)".format(app.coresGranted)
+ } else {
+ "%s (%s granted, %s left)".format(
+ app.desc.maxCores.get, app.coresGranted, app.coresLeft)
+ }
+ }
+
+ -
+ Executor Memory:
+ {Utils.megabytesToString(app.desc.memoryPerSlave)}
+
+ - Submit Date: {app.submitDate}
+ - State: {app.state}
+ - Application Detail UI
+
+
-
-
-
Executor Summary
- {executorTable}
-
-
;
+
+
+
Executor Summary
+ {executorsTable}
+ {
+ if (removedExecutors.nonEmpty) {
+ Removed Executors
++
+ removedExecutorsTable
+ }
+ }
+
+
;
UIUtils.basicSparkPage(content, "Application: " + app.desc.name)
}
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/ExecutorRunner.scala b/core/src/main/scala/org/apache/spark/deploy/worker/ExecutorRunner.scala
index d09136de49807..6433aac1c23e0 100644
--- a/core/src/main/scala/org/apache/spark/deploy/worker/ExecutorRunner.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/ExecutorRunner.scala
@@ -154,11 +154,10 @@ private[spark] class ExecutorRunner(
Files.write(header, stderr, Charsets.UTF_8)
stderrAppender = FileAppender(process.getErrorStream, stderr, conf)
- // Wait for it to exit; this is actually a bad thing if it happens, because we expect to run
- // long-lived processes only. However, in the future, we might restart the executor a few
- // times on the same machine.
+ // Wait for it to exit; executor may exit with code 0 (when driver instructs it to shutdown)
+ // or with nonzero exit code
val exitCode = process.waitFor()
- state = ExecutorState.FAILED
+ state = ExecutorState.EXITED
val message = "Command exited with code " + exitCode
worker ! ExecutorStateChanged(appId, execId, state, Some(message), Some(exitCode))
} catch {
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/ui/WorkerPage.scala b/core/src/main/scala/org/apache/spark/deploy/worker/ui/WorkerPage.scala
index d4513118ced05..327b905032800 100644
--- a/core/src/main/scala/org/apache/spark/deploy/worker/ui/WorkerPage.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/ui/WorkerPage.scala
@@ -46,74 +46,62 @@ private[spark] class WorkerPage(parent: WorkerWebUI) extends WebUIPage("") {
val stateFuture = (workerActor ? RequestWorkerState)(timeout).mapTo[WorkerStateResponse]
val workerState = Await.result(stateFuture, timeout)
- val executorHeaders = Seq("ExecutorID", "Cores", "Memory", "Job Details", "Logs")
+ val executorHeaders = Seq("ExecutorID", "Cores", "State", "Memory", "Job Details", "Logs")
+ val runningExecutors = workerState.executors
val runningExecutorTable =
- UIUtils.listingTable(executorHeaders, executorRow, workerState.executors)
+ UIUtils.listingTable(executorHeaders, executorRow, runningExecutors)
+ val finishedExecutors = workerState.finishedExecutors
val finishedExecutorTable =
- UIUtils.listingTable(executorHeaders, executorRow, workerState.finishedExecutors)
+ UIUtils.listingTable(executorHeaders, executorRow, finishedExecutors)
val driverHeaders = Seq("DriverID", "Main Class", "State", "Cores", "Memory", "Logs", "Notes")
val runningDrivers = workerState.drivers.sortBy(_.driverId).reverse
val runningDriverTable = UIUtils.listingTable(driverHeaders, driverRow, runningDrivers)
val finishedDrivers = workerState.finishedDrivers.sortBy(_.driverId).reverse
- def finishedDriverTable = UIUtils.listingTable(driverHeaders, driverRow, finishedDrivers)
+ val finishedDriverTable = UIUtils.listingTable(driverHeaders, driverRow, finishedDrivers)
// For now we only show driver information if the user has submitted drivers to the cluster.
// This is until we integrate the notion of drivers and applications in the UI.
- def hasDrivers = runningDrivers.length > 0 || finishedDrivers.length > 0
val content =
-
-
-
- - ID: {workerState.workerId}
- -
- Master URL: {workerState.masterUrl}
-
- - Cores: {workerState.cores} ({workerState.coresUsed} Used)
- - Memory: {Utils.megabytesToString(workerState.memory)}
- ({Utils.megabytesToString(workerState.memoryUsed)} Used)
-
-
Back to Master
-
+
+
+
+ - ID: {workerState.workerId}
+ -
+ Master URL: {workerState.masterUrl}
+
+ - Cores: {workerState.cores} ({workerState.coresUsed} Used)
+ - Memory: {Utils.megabytesToString(workerState.memory)}
+ ({Utils.megabytesToString(workerState.memoryUsed)} Used)
+
+
Back to Master
-
-
-
-
Running Executors {workerState.executors.size}
- {runningExecutorTable}
-
-
- // scalastyle:off
-
- {if (hasDrivers)
-
-
-
Running Drivers {workerState.drivers.size}
- {runningDriverTable}
-
-
+
+
+
+
Running Executors ({runningExecutors.size})
+ {runningExecutorTable}
+ {
+ if (runningDrivers.nonEmpty) {
+ Running Drivers ({runningDrivers.size})
++
+ runningDriverTable
+ }
}
-
-
-
-
-
Finished Executors
- {finishedExecutorTable}
-
-
-
-
- {if (hasDrivers)
-
-
-
Finished Drivers
- {finishedDriverTable}
-
-
+ {
+ if (finishedExecutors.nonEmpty) {
+
Finished Executors ({finishedExecutors.size})
++
+ finishedExecutorTable
+ }
}
-
;
- // scalastyle:on
+ {
+ if (finishedDrivers.nonEmpty) {
+
Finished Drivers ({finishedDrivers.size})
++
+ finishedDriverTable
+ }
+ }
+
+
;
UIUtils.basicSparkPage(content, "Spark Worker at %s:%s".format(
workerState.host, workerState.port))
}
@@ -122,6 +110,7 @@ private[spark] class WorkerPage(parent: WorkerWebUI) extends WebUIPage("") {
| {executor.execId} |
{executor.cores} |
+ {executor.state} |
{Utils.megabytesToString(executor.memory)}
|
diff --git a/core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala b/core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala
index b6ad9b6c3e168..443d1c587c3ee 100644
--- a/core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala
@@ -567,6 +567,28 @@ class PairRDDFunctions[K, V](self: RDD[(K, V)])
new FlatMappedValuesRDD(self, cleanF)
}
+ /**
+ * For each key k in `this` or `other1` or `other2` or `other3`,
+ * return a resulting RDD that contains a tuple with the list of values
+ * for that key in `this`, `other1`, `other2` and `other3`.
+ */
+ def cogroup[W1, W2, W3](other1: RDD[(K, W1)],
+ other2: RDD[(K, W2)],
+ other3: RDD[(K, W3)],
+ partitioner: Partitioner)
+ : RDD[(K, (Iterable[V], Iterable[W1], Iterable[W2], Iterable[W3]))] = {
+ if (partitioner.isInstanceOf[HashPartitioner] && keyClass.isArray) {
+ throw new SparkException("Default partitioner cannot partition array keys.")
+ }
+ val cg = new CoGroupedRDD[K](Seq(self, other1, other2, other3), partitioner)
+ cg.mapValues { case Seq(vs, w1s, w2s, w3s) =>
+ (vs.asInstanceOf[Seq[V]],
+ w1s.asInstanceOf[Seq[W1]],
+ w2s.asInstanceOf[Seq[W2]],
+ w3s.asInstanceOf[Seq[W3]])
+ }
+ }
+
/**
* For each key k in `this` or `other`, return a resulting RDD that contains a tuple with the
* list of values for that key in `this` as well as `other`.
@@ -599,6 +621,16 @@ class PairRDDFunctions[K, V](self: RDD[(K, V)])
}
}
+ /**
+ * For each key k in `this` or `other1` or `other2` or `other3`,
+ * return a resulting RDD that contains a tuple with the list of values
+ * for that key in `this`, `other1`, `other2` and `other3`.
+ */
+ def cogroup[W1, W2, W3](other1: RDD[(K, W1)], other2: RDD[(K, W2)], other3: RDD[(K, W3)])
+ : RDD[(K, (Iterable[V], Iterable[W1], Iterable[W2], Iterable[W3]))] = {
+ cogroup(other1, other2, other3, defaultPartitioner(self, other1, other2, other3))
+ }
+
/**
* For each key k in `this` or `other`, return a resulting RDD that contains a tuple with the
* list of values for that key in `this` as well as `other`.
@@ -633,6 +665,19 @@ class PairRDDFunctions[K, V](self: RDD[(K, V)])
cogroup(other1, other2, new HashPartitioner(numPartitions))
}
+ /**
+ * For each key k in `this` or `other1` or `other2` or `other3`,
+ * return a resulting RDD that contains a tuple with the list of values
+ * for that key in `this`, `other1`, `other2` and `other3`.
+ */
+ def cogroup[W1, W2, W3](other1: RDD[(K, W1)],
+ other2: RDD[(K, W2)],
+ other3: RDD[(K, W3)],
+ numPartitions: Int)
+ : RDD[(K, (Iterable[V], Iterable[W1], Iterable[W2], Iterable[W3]))] = {
+ cogroup(other1, other2, other3, new HashPartitioner(numPartitions))
+ }
+
/** Alias for cogroup. */
def groupWith[W](other: RDD[(K, W)]): RDD[(K, (Iterable[V], Iterable[W]))] = {
cogroup(other, defaultPartitioner(self, other))
@@ -644,6 +689,12 @@ class PairRDDFunctions[K, V](self: RDD[(K, V)])
cogroup(other1, other2, defaultPartitioner(self, other1, other2))
}
+ /** Alias for cogroup. */
+ def groupWith[W1, W2, W3](other1: RDD[(K, W1)], other2: RDD[(K, W2)], other3: RDD[(K, W3)])
+ : RDD[(K, (Iterable[V], Iterable[W1], Iterable[W2], Iterable[W3]))] = {
+ cogroup(other1, other2, other3, defaultPartitioner(self, other1, other2, other3))
+ }
+
/**
* Return an RDD with the pairs from `this` whose keys are not in `other`.
*
@@ -787,8 +838,7 @@ class PairRDDFunctions[K, V](self: RDD[(K, V)])
val outfmt = job.getOutputFormatClass
val jobFormat = outfmt.newInstance
- if (self.conf.getBoolean("spark.hadoop.validateOutputSpecs", true) &&
- jobFormat.isInstanceOf[NewFileOutputFormat[_, _]]) {
+ if (self.conf.getBoolean("spark.hadoop.validateOutputSpecs", true)) {
// FileOutputFormat ignores the filesystem parameter
jobFormat.checkOutputSpecs(job)
}
@@ -854,8 +904,7 @@ class PairRDDFunctions[K, V](self: RDD[(K, V)])
logDebug("Saving as hadoop file of type (" + keyClass.getSimpleName + ", " +
valueClass.getSimpleName + ")")
- if (self.conf.getBoolean("spark.hadoop.validateOutputSpecs", true) &&
- outputFormatInstance.isInstanceOf[FileOutputFormat[_, _]]) {
+ if (self.conf.getBoolean("spark.hadoop.validateOutputSpecs", true)) {
// FileOutputFormat ignores the filesystem parameter
val ignoredFs = FileSystem.get(conf)
conf.getOutputFormat.checkOutputSpecs(ignoredFs, conf)
diff --git a/core/src/main/scala/org/apache/spark/rdd/ParallelCollectionRDD.scala b/core/src/main/scala/org/apache/spark/rdd/ParallelCollectionRDD.scala
index 2425929fc73c5..66c71bf7e8bb5 100644
--- a/core/src/main/scala/org/apache/spark/rdd/ParallelCollectionRDD.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/ParallelCollectionRDD.scala
@@ -117,6 +117,15 @@ private object ParallelCollectionRDD {
if (numSlices < 1) {
throw new IllegalArgumentException("Positive number of slices required")
}
+ // Sequences need to be sliced at the same set of index positions for operations
+ // like RDD.zip() to behave as expected
+ def positions(length: Long, numSlices: Int): Iterator[(Int, Int)] = {
+ (0 until numSlices).iterator.map(i => {
+ val start = ((i * length) / numSlices).toInt
+ val end = (((i + 1) * length) / numSlices).toInt
+ (start, end)
+ })
+ }
seq match {
case r: Range.Inclusive => {
val sign = if (r.step < 0) {
@@ -128,18 +137,17 @@ private object ParallelCollectionRDD {
r.start, r.end + sign, r.step).asInstanceOf[Seq[T]], numSlices)
}
case r: Range => {
- (0 until numSlices).map(i => {
- val start = ((i * r.length.toLong) / numSlices).toInt
- val end = (((i + 1) * r.length.toLong) / numSlices).toInt
- new Range(r.start + start * r.step, r.start + end * r.step, r.step)
- }).asInstanceOf[Seq[Seq[T]]]
+ positions(r.length, numSlices).map({
+ case (start, end) =>
+ new Range(r.start + start * r.step, r.start + end * r.step, r.step)
+ }).toSeq.asInstanceOf[Seq[Seq[T]]]
}
case nr: NumericRange[_] => {
// For ranges of Long, Double, BigInteger, etc
val slices = new ArrayBuffer[Seq[T]](numSlices)
- val sliceSize = (nr.size + numSlices - 1) / numSlices // Round up to catch everything
var r = nr
- for (i <- 0 until numSlices) {
+ for ((start, end) <- positions(nr.length, numSlices)) {
+ val sliceSize = end - start
slices += r.take(sliceSize).asInstanceOf[Seq[T]]
r = r.drop(sliceSize)
}
@@ -147,11 +155,10 @@ private object ParallelCollectionRDD {
}
case _ => {
val array = seq.toArray // To prevent O(n^2) operations for List etc
- (0 until numSlices).map(i => {
- val start = ((i * array.length.toLong) / numSlices).toInt
- val end = (((i + 1) * array.length.toLong) / numSlices).toInt
- array.slice(start, end).toSeq
- })
+ positions(array.length, numSlices).map({
+ case (start, end) =>
+ array.slice(start, end).toSeq
+ }).toSeq
}
}
}
diff --git a/core/src/main/scala/org/apache/spark/rdd/RDD.scala b/core/src/main/scala/org/apache/spark/rdd/RDD.scala
index 446f369c9ea16..cebfd109d825f 100644
--- a/core/src/main/scala/org/apache/spark/rdd/RDD.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/RDD.scala
@@ -40,7 +40,7 @@ import org.apache.spark.partial.CountEvaluator
import org.apache.spark.partial.GroupedCountEvaluator
import org.apache.spark.partial.PartialResult
import org.apache.spark.storage.StorageLevel
-import org.apache.spark.util.{BoundedPriorityQueue, Utils}
+import org.apache.spark.util.{BoundedPriorityQueue, CallSite, Utils}
import org.apache.spark.util.collection.OpenHashMap
import org.apache.spark.util.random.{BernoulliSampler, PoissonSampler, SamplingUtils}
@@ -442,6 +442,18 @@ abstract class RDD[T: ClassTag](
*/
def ++(other: RDD[T]): RDD[T] = this.union(other)
+ /**
+ * Return this RDD sorted by the given key function.
+ */
+ def sortBy[K](
+ f: (T) => K,
+ ascending: Boolean = true,
+ numPartitions: Int = this.partitions.size)
+ (implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T] =
+ this.keyBy[K](f)
+ .sortByKey(ascending, numPartitions)
+ .values
+
/**
* Return the intersection of this RDD and another one. The output will not contain any duplicate
* elements, even if the input RDDs did.
@@ -1062,11 +1074,11 @@ abstract class RDD[T: ClassTag](
* Returns the top K (largest) elements from this RDD as defined by the specified
* implicit Ordering[T]. This does the opposite of [[takeOrdered]]. For example:
* {{{
- * sc.parallelize([10, 4, 2, 12, 3]).top(1)
- * // returns [12]
+ * sc.parallelize(Seq(10, 4, 2, 12, 3)).top(1)
+ * // returns Array(12)
*
- * sc.parallelize([2, 3, 4, 5, 6]).top(2)
- * // returns [6, 5]
+ * sc.parallelize(Seq(2, 3, 4, 5, 6)).top(2)
+ * // returns Array(6, 5)
* }}}
*
* @param num the number of top elements to return
@@ -1080,11 +1092,11 @@ abstract class RDD[T: ClassTag](
* implicit Ordering[T] and maintains the ordering. This does the opposite of [[top]].
* For example:
* {{{
- * sc.parallelize([10, 4, 2, 12, 3]).takeOrdered(1)
- * // returns [12]
+ * sc.parallelize(Seq(10, 4, 2, 12, 3)).takeOrdered(1)
+ * // returns Array(2)
*
- * sc.parallelize([2, 3, 4, 5, 6]).takeOrdered(2)
- * // returns [2, 3]
+ * sc.parallelize(Seq(2, 3, 4, 5, 6)).takeOrdered(2)
+ * // returns Array(2, 3)
* }}}
*
* @param num the number of top elements to return
@@ -1189,8 +1201,8 @@ abstract class RDD[T: ClassTag](
private var storageLevel: StorageLevel = StorageLevel.NONE
/** User code that created this RDD (e.g. `textFile`, `parallelize`). */
- @transient private[spark] val creationSiteInfo = Utils.getCallSiteInfo
- private[spark] def getCreationSite: String = Option(creationSiteInfo).getOrElse("").toString
+ @transient private[spark] val creationSite = Utils.getCallSite
+ private[spark] def getCreationSite: String = Option(creationSite).map(_.short).getOrElse("")
private[spark] def elementClassTag: ClassTag[T] = classTag[T]
diff --git a/core/src/main/scala/org/apache/spark/scheduler/ActiveJob.scala b/core/src/main/scala/org/apache/spark/scheduler/ActiveJob.scala
index 9257f48559c9e..b755d8fb15757 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/ActiveJob.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/ActiveJob.scala
@@ -20,6 +20,7 @@ package org.apache.spark.scheduler
import java.util.Properties
import org.apache.spark.TaskContext
+import org.apache.spark.util.CallSite
/**
* Tracks information about an active job in the DAGScheduler.
@@ -29,7 +30,7 @@ private[spark] class ActiveJob(
val finalStage: Stage,
val func: (TaskContext, Iterator[_]) => _,
val partitions: Array[Int],
- val callSite: String,
+ val callSite: CallSite,
val listener: JobListener,
val properties: Properties) {
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index 3c85b5a2ae776..b3ebaa547de0d 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -38,7 +38,7 @@ import org.apache.spark.executor.TaskMetrics
import org.apache.spark.partial.{ApproximateActionListener, ApproximateEvaluator, PartialResult}
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.{BlockId, BlockManager, BlockManagerMaster, RDDBlockId}
-import org.apache.spark.util.{SystemClock, Clock, Utils}
+import org.apache.spark.util.{CallSite, SystemClock, Clock, Utils}
/**
* The high-level scheduling layer that implements stage-oriented scheduling. It computes a DAG of
@@ -195,7 +195,9 @@ class DAGScheduler(
case Some(stage) => stage
case None =>
val stage =
- newOrUsedStage(shuffleDep.rdd, shuffleDep.rdd.partitions.size, shuffleDep, jobId)
+ newOrUsedStage(
+ shuffleDep.rdd, shuffleDep.rdd.partitions.size, shuffleDep, jobId,
+ shuffleDep.rdd.creationSite)
shuffleToMapStage(shuffleDep.shuffleId) = stage
stage
}
@@ -212,7 +214,7 @@ class DAGScheduler(
numTasks: Int,
shuffleDep: Option[ShuffleDependency[_, _, _]],
jobId: Int,
- callSite: Option[String] = None)
+ callSite: CallSite)
: Stage =
{
val id = nextStageId.getAndIncrement()
@@ -235,7 +237,7 @@ class DAGScheduler(
numTasks: Int,
shuffleDep: ShuffleDependency[_, _, _],
jobId: Int,
- callSite: Option[String] = None)
+ callSite: CallSite)
: Stage =
{
val stage = newStage(rdd, numTasks, Some(shuffleDep), jobId, callSite)
@@ -413,7 +415,7 @@ class DAGScheduler(
rdd: RDD[T],
func: (TaskContext, Iterator[T]) => U,
partitions: Seq[Int],
- callSite: String,
+ callSite: CallSite,
allowLocal: Boolean,
resultHandler: (Int, U) => Unit,
properties: Properties = null): JobWaiter[U] =
@@ -443,7 +445,7 @@ class DAGScheduler(
rdd: RDD[T],
func: (TaskContext, Iterator[T]) => U,
partitions: Seq[Int],
- callSite: String,
+ callSite: CallSite,
allowLocal: Boolean,
resultHandler: (Int, U) => Unit,
properties: Properties = null)
@@ -452,7 +454,7 @@ class DAGScheduler(
waiter.awaitResult() match {
case JobSucceeded => {}
case JobFailed(exception: Exception) =>
- logInfo("Failed to run " + callSite)
+ logInfo("Failed to run " + callSite.short)
throw exception
}
}
@@ -461,7 +463,7 @@ class DAGScheduler(
rdd: RDD[T],
func: (TaskContext, Iterator[T]) => U,
evaluator: ApproximateEvaluator[U, R],
- callSite: String,
+ callSite: CallSite,
timeout: Long,
properties: Properties = null)
: PartialResult[R] =
@@ -666,7 +668,7 @@ class DAGScheduler(
func: (TaskContext, Iterator[_]) => _,
partitions: Array[Int],
allowLocal: Boolean,
- callSite: String,
+ callSite: CallSite,
listener: JobListener,
properties: Properties = null)
{
@@ -674,7 +676,7 @@ class DAGScheduler(
try {
// New stage creation may throw an exception if, for example, jobs are run on a
// HadoopRDD whose underlying HDFS files have been deleted.
- finalStage = newStage(finalRDD, partitions.size, None, jobId, Some(callSite))
+ finalStage = newStage(finalRDD, partitions.size, None, jobId, callSite)
} catch {
case e: Exception =>
logWarning("Creating new stage failed due to exception - job: " + jobId, e)
@@ -685,7 +687,7 @@ class DAGScheduler(
val job = new ActiveJob(jobId, finalStage, func, partitions, callSite, listener, properties)
clearCacheLocs()
logInfo("Got job %s (%s) with %d output partitions (allowLocal=%s)".format(
- job.jobId, callSite, partitions.length, allowLocal))
+ job.jobId, callSite.short, partitions.length, allowLocal))
logInfo("Final stage: " + finalStage + "(" + finalStage.name + ")")
logInfo("Parents of final stage: " + finalStage.parents)
logInfo("Missing parents: " + getMissingParentStages(finalStage))
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala
index 23f57441b4b11..2b6f7e4205c32 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala
@@ -25,6 +25,7 @@ import scala.language.existentials
import org.apache.spark._
import org.apache.spark.executor.TaskMetrics
import org.apache.spark.rdd.RDD
+import org.apache.spark.util.CallSite
/**
* Types of events that can be handled by the DAGScheduler. The DAGScheduler uses an event queue
@@ -40,7 +41,7 @@ private[scheduler] case class JobSubmitted(
func: (TaskContext, Iterator[_]) => _,
partitions: Array[Int],
allowLocal: Boolean,
- callSite: String,
+ callSite: CallSite,
listener: JobListener,
properties: Properties = null)
extends DAGSchedulerEvent
diff --git a/core/src/main/scala/org/apache/spark/scheduler/Stage.scala b/core/src/main/scala/org/apache/spark/scheduler/Stage.scala
index 3bf9713f728c6..9a4be43ee219f 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/Stage.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/Stage.scala
@@ -20,6 +20,7 @@ package org.apache.spark.scheduler
import org.apache.spark._
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.BlockManagerId
+import org.apache.spark.util.CallSite
/**
* A stage is a set of independent tasks all computing the same function that need to run as part
@@ -35,6 +36,11 @@ import org.apache.spark.storage.BlockManagerId
* Each Stage also has a jobId, identifying the job that first submitted the stage. When FIFO
* scheduling is used, this allows Stages from earlier jobs to be computed first or recovered
* faster on failure.
+ *
+ * The callSite provides a location in user code which relates to the stage. For a shuffle map
+ * stage, the callSite gives the user code that created the RDD being shuffled. For a result
+ * stage, the callSite gives the user code that executes the associated action (e.g. count()).
+ *
*/
private[spark] class Stage(
val id: Int,
@@ -43,7 +49,7 @@ private[spark] class Stage(
val shuffleDep: Option[ShuffleDependency[_, _, _]], // Output shuffle if stage is a map stage
val parents: List[Stage],
val jobId: Int,
- callSite: Option[String])
+ val callSite: CallSite)
extends Logging {
val isShuffleMap = shuffleDep.isDefined
@@ -100,7 +106,8 @@ private[spark] class Stage(
id
}
- val name = callSite.getOrElse(rdd.getCreationSite)
+ val name = callSite.short
+ val details = callSite.long
override def toString = "Stage " + id
diff --git a/core/src/main/scala/org/apache/spark/scheduler/StageInfo.scala b/core/src/main/scala/org/apache/spark/scheduler/StageInfo.scala
index b42e231e11f91..7644e3f351b3c 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/StageInfo.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/StageInfo.scala
@@ -25,7 +25,12 @@ import org.apache.spark.storage.RDDInfo
* Stores information about a stage to pass from the scheduler to SparkListeners.
*/
@DeveloperApi
-class StageInfo(val stageId: Int, val name: String, val numTasks: Int, val rddInfos: Seq[RDDInfo]) {
+class StageInfo(
+ val stageId: Int,
+ val name: String,
+ val numTasks: Int,
+ val rddInfos: Seq[RDDInfo],
+ val details: String) {
/** When this stage was submitted from the DAGScheduler to a TaskScheduler. */
var submissionTime: Option[Long] = None
/** Time when all tasks in the stage completed or when the stage was cancelled. */
@@ -52,6 +57,6 @@ private[spark] object StageInfo {
def fromStage(stage: Stage): StageInfo = {
val ancestorRddInfos = stage.rdd.getNarrowAncestors.map(RDDInfo.fromRdd)
val rddInfos = Seq(RDDInfo.fromRdd(stage.rdd)) ++ ancestorRddInfos
- new StageInfo(stage.id, stage.name, stage.numTasks, rddInfos)
+ new StageInfo(stage.id, stage.name, stage.numTasks, rddInfos, stage.details)
}
}
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
index f52bc7075104b..d2f7baf928b62 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
@@ -363,6 +363,15 @@ private[spark] class BlockManager(
val info = blockInfo.get(blockId).orNull
if (info != null) {
info.synchronized {
+ // Double check to make sure the block is still there. There is a small chance that the
+ // block has been removed by removeBlock (which also synchronizes on the blockInfo object).
+ // Note that this only checks metadata tracking. If user intentionally deleted the block
+ // on disk or from off heap storage without using removeBlock, this conditional check will
+ // still pass but eventually we will get an exception because we can't find the block.
+ if (blockInfo.get(blockId).isEmpty) {
+ logWarning(s"Block $blockId had been removed")
+ return None
+ }
// If another thread is writing the block, wait for it to become ready.
if (!info.waitForReady()) {
diff --git a/core/src/main/scala/org/apache/spark/storage/RDDInfo.scala b/core/src/main/scala/org/apache/spark/storage/RDDInfo.scala
index 023fd6e4d8baa..5a72e216872a6 100644
--- a/core/src/main/scala/org/apache/spark/storage/RDDInfo.scala
+++ b/core/src/main/scala/org/apache/spark/storage/RDDInfo.scala
@@ -26,7 +26,7 @@ class RDDInfo(
val id: Int,
val name: String,
val numPartitions: Int,
- val storageLevel: StorageLevel)
+ var storageLevel: StorageLevel)
extends Ordered[RDDInfo] {
var numCachedPartitions = 0
@@ -36,8 +36,8 @@ class RDDInfo(
override def toString = {
import Utils.bytesToString
- ("RDD \"%s\" (%d) Storage: %s; CachedPartitions: %d; TotalPartitions: %d; MemorySize: %s; " +
- "TachyonSize: %s; DiskSize: %s").format(
+ ("RDD \"%s\" (%d) StorageLevel: %s; CachedPartitions: %d; TotalPartitions: %d; " +
+ "MemorySize: %s; TachyonSize: %s; DiskSize: %s").format(
name, id, storageLevel.toString, numCachedPartitions, numPartitions,
bytesToString(memSize), bytesToString(tachyonSize), bytesToString(diskSize))
}
diff --git a/core/src/main/scala/org/apache/spark/storage/StorageStatusListener.scala b/core/src/main/scala/org/apache/spark/storage/StorageStatusListener.scala
index a6e6627d54e01..c694fc8c347ec 100644
--- a/core/src/main/scala/org/apache/spark/storage/StorageStatusListener.scala
+++ b/core/src/main/scala/org/apache/spark/storage/StorageStatusListener.scala
@@ -37,7 +37,11 @@ class StorageStatusListener extends SparkListener {
val filteredStatus = storageStatusList.find(_.blockManagerId.executorId == execId)
filteredStatus.foreach { storageStatus =>
updatedBlocks.foreach { case (blockId, updatedStatus) =>
- storageStatus.blocks(blockId) = updatedStatus
+ if (updatedStatus.storageLevel == StorageLevel.NONE) {
+ storageStatus.blocks.remove(blockId)
+ } else {
+ storageStatus.blocks(blockId) = updatedStatus
+ }
}
}
}
diff --git a/core/src/main/scala/org/apache/spark/storage/StorageUtils.scala b/core/src/main/scala/org/apache/spark/storage/StorageUtils.scala
index 6f3252a2f6d31..f3bde1df45c79 100644
--- a/core/src/main/scala/org/apache/spark/storage/StorageUtils.scala
+++ b/core/src/main/scala/org/apache/spark/storage/StorageUtils.scala
@@ -89,10 +89,13 @@ private[spark] object StorageUtils {
// Add up memory, disk and Tachyon sizes
val persistedBlocks =
blocks.filter { status => status.memSize + status.diskSize + status.tachyonSize > 0 }
+ val _storageLevel =
+ if (persistedBlocks.length > 0) persistedBlocks(0).storageLevel else StorageLevel.NONE
val memSize = persistedBlocks.map(_.memSize).reduceOption(_ + _).getOrElse(0L)
val diskSize = persistedBlocks.map(_.diskSize).reduceOption(_ + _).getOrElse(0L)
val tachyonSize = persistedBlocks.map(_.tachyonSize).reduceOption(_ + _).getOrElse(0L)
rddInfoMap.get(rddId).map { rddInfo =>
+ rddInfo.storageLevel = _storageLevel
rddInfo.numCachedPartitions = persistedBlocks.length
rddInfo.memSize = memSize
rddInfo.diskSize = diskSize
diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
index 153434a2032be..a3f824a4e1f57 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
@@ -91,9 +91,17 @@ private[ui] class StageTableBase(
{s.name}
+ val details = if (s.details.nonEmpty) (
+
+ +show details
+
+ {s.details}
+ )
+
listener.stageIdToDescription.get(s.stageId)
.map(d => {d}
{nameLink} {killLink}
)
- .getOrElse( {killLink}{nameLink}
)
+ .getOrElse({killLink} {nameLink} {details}
)
}
protected def stageRow(s: StageInfo): Seq[Node] = {
diff --git a/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala b/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala
index 09825087bb048..7cecbfe62a382 100644
--- a/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala
+++ b/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala
@@ -184,6 +184,7 @@ private[spark] object JsonProtocol {
("Stage Name" -> stageInfo.name) ~
("Number of Tasks" -> stageInfo.numTasks) ~
("RDD Info" -> rddInfo) ~
+ ("Details" -> stageInfo.details) ~
("Submission Time" -> submissionTime) ~
("Completion Time" -> completionTime) ~
("Failure Reason" -> failureReason) ~
@@ -469,12 +470,13 @@ private[spark] object JsonProtocol {
val stageName = (json \ "Stage Name").extract[String]
val numTasks = (json \ "Number of Tasks").extract[Int]
val rddInfos = (json \ "RDD Info").extract[List[JValue]].map(rddInfoFromJson)
+ val details = (json \ "Details").extractOpt[String].getOrElse("")
val submissionTime = Utils.jsonOption(json \ "Submission Time").map(_.extract[Long])
val completionTime = Utils.jsonOption(json \ "Completion Time").map(_.extract[Long])
val failureReason = Utils.jsonOption(json \ "Failure Reason").map(_.extract[String])
val emittedTaskSizeWarning = (json \ "Emitted Task Size Warning").extract[Boolean]
- val stageInfo = new StageInfo(stageId, stageName, numTasks, rddInfos)
+ val stageInfo = new StageInfo(stageId, stageName, numTasks, rddInfos, details)
stageInfo.submissionTime = submissionTime
stageInfo.completionTime = completionTime
stageInfo.failureReason = failureReason
diff --git a/core/src/main/scala/org/apache/spark/util/MetadataCleaner.scala b/core/src/main/scala/org/apache/spark/util/MetadataCleaner.scala
index 7ebed5105b9fd..2889e171f627e 100644
--- a/core/src/main/scala/org/apache/spark/util/MetadataCleaner.scala
+++ b/core/src/main/scala/org/apache/spark/util/MetadataCleaner.scala
@@ -91,8 +91,13 @@ private[spark] object MetadataCleaner {
conf.set(MetadataCleanerType.systemProperty(cleanerType), delay.toString)
}
+ /**
+ * Set the default delay time (in seconds).
+ * @param conf SparkConf instance
+ * @param delay default delay time to set
+ * @param resetAll whether to reset all to default
+ */
def setDelaySeconds(conf: SparkConf, delay: Int, resetAll: Boolean = true) {
- // override for all ?
conf.set("spark.cleaner.ttl", delay.toString)
if (resetAll) {
for (cleanerType <- MetadataCleanerType.values) {
diff --git a/core/src/main/scala/org/apache/spark/util/Utils.scala b/core/src/main/scala/org/apache/spark/util/Utils.scala
index 4ce28bb0cf059..a2454e120a8ab 100644
--- a/core/src/main/scala/org/apache/spark/util/Utils.scala
+++ b/core/src/main/scala/org/apache/spark/util/Utils.scala
@@ -43,6 +43,9 @@ import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.executor.ExecutorUncaughtExceptionHandler
import org.apache.spark.serializer.{DeserializationStream, SerializationStream, SerializerInstance}
+/** CallSite represents a place in user code. It can have a short and a long form. */
+private[spark] case class CallSite(val short: String, val long: String)
+
/**
* Various utility methods used by Spark.
*/
@@ -799,21 +802,12 @@ private[spark] object Utils extends Logging {
*/
private val SPARK_CLASS_REGEX = """^org\.apache\.spark(\.api\.java)?(\.util)?(\.rdd)?\.[A-Z]""".r
- private[spark] class CallSiteInfo(val lastSparkMethod: String, val firstUserFile: String,
- val firstUserLine: Int, val firstUserClass: String) {
-
- /** Returns a printable version of the call site info suitable for logs. */
- override def toString = {
- "%s at %s:%s".format(lastSparkMethod, firstUserFile, firstUserLine)
- }
- }
-
/**
* When called inside a class in the spark package, returns the name of the user code class
* (outside the spark package) that called into Spark, as well as which Spark method they called.
* This is used, for example, to tell users where in their code each RDD got created.
*/
- def getCallSiteInfo: CallSiteInfo = {
+ def getCallSite: CallSite = {
val trace = Thread.currentThread.getStackTrace()
.filterNot(_.getMethodName.contains("getStackTrace"))
@@ -824,11 +818,11 @@ private[spark] object Utils extends Logging {
var lastSparkMethod = ""
var firstUserFile = ""
var firstUserLine = 0
- var finished = false
- var firstUserClass = ""
+ var insideSpark = true
+ var callStack = new ArrayBuffer[String]() :+ ""
for (el <- trace) {
- if (!finished) {
+ if (insideSpark) {
if (SPARK_CLASS_REGEX.findFirstIn(el.getClassName).isDefined) {
lastSparkMethod = if (el.getMethodName == "") {
// Spark method is a constructor; get its class name
@@ -836,15 +830,21 @@ private[spark] object Utils extends Logging {
} else {
el.getMethodName
}
+ callStack(0) = el.toString // Put last Spark method on top of the stack trace.
} else {
firstUserLine = el.getLineNumber
firstUserFile = el.getFileName
- firstUserClass = el.getClassName
- finished = true
+ callStack += el.toString
+ insideSpark = false
}
+ } else {
+ callStack += el.toString
}
}
- new CallSiteInfo(lastSparkMethod, firstUserFile, firstUserLine, firstUserClass)
+ val callStackDepth = System.getProperty("spark.callstack.depth", "20").toInt
+ CallSite(
+ short = "%s at %s:%s".format(lastSparkMethod, firstUserFile, firstUserLine),
+ long = callStack.take(callStackDepth).mkString("\n"))
}
/** Return a string containing part of a file from byte 'start' to 'end'. */
diff --git a/core/src/test/java/org/apache/spark/JavaAPISuite.java b/core/src/test/java/org/apache/spark/JavaAPISuite.java
index ef41bfb88de9d..761f2d6a77d33 100644
--- a/core/src/test/java/org/apache/spark/JavaAPISuite.java
+++ b/core/src/test/java/org/apache/spark/JavaAPISuite.java
@@ -21,6 +21,9 @@
import java.util.*;
import scala.Tuple2;
+import scala.Tuple3;
+import scala.Tuple4;
+
import com.google.common.collect.Iterables;
import com.google.common.collect.Iterators;
@@ -180,6 +183,39 @@ public void sortByKey() {
Assert.assertEquals(new Tuple2(3, 2), sortedPairs.get(2));
}
+ @Test
+ public void sortBy() {
+ List> pairs = new ArrayList>();
+ pairs.add(new Tuple2(0, 4));
+ pairs.add(new Tuple2(3, 2));
+ pairs.add(new Tuple2(-1, 1));
+
+ JavaRDD> rdd = sc.parallelize(pairs);
+
+ // compare on first value
+ JavaRDD> sortedRDD = rdd.sortBy(new Function, Integer>() {
+ public Integer call(Tuple2 t) throws Exception {
+ return t._1();
+ }
+ }, true, 2);
+
+ Assert.assertEquals(new Tuple2(-1, 1), sortedRDD.first());
+ List> sortedPairs = sortedRDD.collect();
+ Assert.assertEquals(new Tuple2(0, 4), sortedPairs.get(1));
+ Assert.assertEquals(new Tuple2(3, 2), sortedPairs.get(2));
+
+ // compare on second value
+ sortedRDD = rdd.sortBy(new Function, Integer>() {
+ public Integer call(Tuple2 t) throws Exception {
+ return t._2();
+ }
+ }, true, 2);
+ Assert.assertEquals(new Tuple2(-1, 1), sortedRDD.first());
+ sortedPairs = sortedRDD.collect();
+ Assert.assertEquals(new Tuple2(3, 2), sortedPairs.get(1));
+ Assert.assertEquals(new Tuple2(0, 4), sortedPairs.get(2));
+ }
+
@Test
public void foreach() {
final Accumulator accum = sc.accumulator(0);
@@ -271,6 +307,66 @@ public void cogroup() {
cogrouped.collect();
}
+ @SuppressWarnings("unchecked")
+ @Test
+ public void cogroup3() {
+ JavaPairRDD categories = sc.parallelizePairs(Arrays.asList(
+ new Tuple2("Apples", "Fruit"),
+ new Tuple2("Oranges", "Fruit"),
+ new Tuple2("Oranges", "Citrus")
+ ));
+ JavaPairRDD prices = sc.parallelizePairs(Arrays.asList(
+ new Tuple2("Oranges", 2),
+ new Tuple2("Apples", 3)
+ ));
+ JavaPairRDD quantities = sc.parallelizePairs(Arrays.asList(
+ new Tuple2("Oranges", 21),
+ new Tuple2("Apples", 42)
+ ));
+
+ JavaPairRDD, Iterable, Iterable>> cogrouped =
+ categories.cogroup(prices, quantities);
+ Assert.assertEquals("[Fruit, Citrus]",
+ Iterables.toString(cogrouped.lookup("Oranges").get(0)._1()));
+ Assert.assertEquals("[2]", Iterables.toString(cogrouped.lookup("Oranges").get(0)._2()));
+ Assert.assertEquals("[42]", Iterables.toString(cogrouped.lookup("Apples").get(0)._3()));
+
+
+ cogrouped.collect();
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void cogroup4() {
+ JavaPairRDD categories = sc.parallelizePairs(Arrays.asList(
+ new Tuple2("Apples", "Fruit"),
+ new Tuple2("Oranges", "Fruit"),
+ new Tuple2("Oranges", "Citrus")
+ ));
+ JavaPairRDD prices = sc.parallelizePairs(Arrays.asList(
+ new Tuple2("Oranges", 2),
+ new Tuple2("Apples", 3)
+ ));
+ JavaPairRDD quantities = sc.parallelizePairs(Arrays.asList(
+ new Tuple2("Oranges", 21),
+ new Tuple2("Apples", 42)
+ ));
+ JavaPairRDD countries = sc.parallelizePairs(Arrays.asList(
+ new Tuple2("Oranges", "BR"),
+ new Tuple2("Apples", "US")
+ ));
+
+ JavaPairRDD, Iterable, Iterable, Iterable>> cogrouped =
+ categories.cogroup(prices, quantities, countries);
+ Assert.assertEquals("[Fruit, Citrus]",
+ Iterables.toString(cogrouped.lookup("Oranges").get(0)._1()));
+ Assert.assertEquals("[2]", Iterables.toString(cogrouped.lookup("Oranges").get(0)._2()));
+ Assert.assertEquals("[42]", Iterables.toString(cogrouped.lookup("Apples").get(0)._3()));
+ Assert.assertEquals("[BR]", Iterables.toString(cogrouped.lookup("Oranges").get(0)._4()));
+
+ cogrouped.collect();
+ }
+
@SuppressWarnings("unchecked")
@Test
public void leftOuterJoin() {
diff --git a/core/src/test/scala/org/apache/spark/SparkContextInfoSuite.scala b/core/src/test/scala/org/apache/spark/SparkContextInfoSuite.scala
index cd3887dcc7371..1fde4badda949 100644
--- a/core/src/test/scala/org/apache/spark/SparkContextInfoSuite.scala
+++ b/core/src/test/scala/org/apache/spark/SparkContextInfoSuite.scala
@@ -70,7 +70,7 @@ package object testPackage extends Assertions {
def runCallSiteTest(sc: SparkContext) {
val rdd = sc.makeRDD(Array(1, 2, 3, 4), 2)
val rddCreationSite = rdd.getCreationSite
- val curCallSite = sc.getCallSite() // note: 2 lines after definition of "rdd"
+ val curCallSite = sc.getCallSite().short // note: 2 lines after definition of "rdd"
val rddCreationLine = rddCreationSite match {
case CALL_SITE_REGEX(func, file, line) => {
diff --git a/core/src/test/scala/org/apache/spark/rdd/PairRDDFunctionsSuite.scala b/core/src/test/scala/org/apache/spark/rdd/PairRDDFunctionsSuite.scala
index 0b9004448a63e..447e38ec9dbd0 100644
--- a/core/src/test/scala/org/apache/spark/rdd/PairRDDFunctionsSuite.scala
+++ b/core/src/test/scala/org/apache/spark/rdd/PairRDDFunctionsSuite.scala
@@ -249,6 +249,39 @@ class PairRDDFunctionsSuite extends FunSuite with SharedSparkContext {
))
}
+ test("groupWith3") {
+ val rdd1 = sc.parallelize(Array((1, 1), (1, 2), (2, 1), (3, 1)))
+ val rdd2 = sc.parallelize(Array((1, 'x'), (2, 'y'), (2, 'z'), (4, 'w')))
+ val rdd3 = sc.parallelize(Array((1, 'a'), (3, 'b'), (4, 'c'), (4, 'd')))
+ val joined = rdd1.groupWith(rdd2, rdd3).collect()
+ assert(joined.size === 4)
+ val joinedSet = joined.map(x => (x._1,
+ (x._2._1.toList, x._2._2.toList, x._2._3.toList))).toSet
+ assert(joinedSet === Set(
+ (1, (List(1, 2), List('x'), List('a'))),
+ (2, (List(1), List('y', 'z'), List())),
+ (3, (List(1), List(), List('b'))),
+ (4, (List(), List('w'), List('c', 'd')))
+ ))
+ }
+
+ test("groupWith4") {
+ val rdd1 = sc.parallelize(Array((1, 1), (1, 2), (2, 1), (3, 1)))
+ val rdd2 = sc.parallelize(Array((1, 'x'), (2, 'y'), (2, 'z'), (4, 'w')))
+ val rdd3 = sc.parallelize(Array((1, 'a'), (3, 'b'), (4, 'c'), (4, 'd')))
+ val rdd4 = sc.parallelize(Array((2, '@')))
+ val joined = rdd1.groupWith(rdd2, rdd3, rdd4).collect()
+ assert(joined.size === 4)
+ val joinedSet = joined.map(x => (x._1,
+ (x._2._1.toList, x._2._2.toList, x._2._3.toList, x._2._4.toList))).toSet
+ assert(joinedSet === Set(
+ (1, (List(1, 2), List('x'), List('a'), List())),
+ (2, (List(1), List('y', 'z'), List(), List('@'))),
+ (3, (List(1), List(), List('b'), List())),
+ (4, (List(), List('w'), List('c', 'd'), List()))
+ ))
+ }
+
test("zero-partition RDD") {
val emptyDir = Files.createTempDir()
emptyDir.deleteOnExit()
diff --git a/core/src/test/scala/org/apache/spark/rdd/ParallelCollectionSplitSuite.scala b/core/src/test/scala/org/apache/spark/rdd/ParallelCollectionSplitSuite.scala
index 4df36558b6d4b..1b112f1a41ca9 100644
--- a/core/src/test/scala/org/apache/spark/rdd/ParallelCollectionSplitSuite.scala
+++ b/core/src/test/scala/org/apache/spark/rdd/ParallelCollectionSplitSuite.scala
@@ -111,6 +111,24 @@ class ParallelCollectionSplitSuite extends FunSuite with Checkers {
assert(slices.forall(_.isInstanceOf[Range]))
}
+ test("identical slice sizes between Range and NumericRange") {
+ val r = ParallelCollectionRDD.slice(1 to 7, 4)
+ val nr = ParallelCollectionRDD.slice(1L to 7L, 4)
+ assert(r.size === 4)
+ for (i <- 0 until r.size) {
+ assert(r(i).size === nr(i).size)
+ }
+ }
+
+ test("identical slice sizes between List and NumericRange") {
+ val r = ParallelCollectionRDD.slice(List(1, 2), 4)
+ val nr = ParallelCollectionRDD.slice(1L to 2L, 4)
+ assert(r.size === 4)
+ for (i <- 0 until r.size) {
+ assert(r(i).size === nr(i).size)
+ }
+ }
+
test("large ranges don't overflow") {
val N = 100 * 1000 * 1000
val data = 0 until N
diff --git a/core/src/test/scala/org/apache/spark/rdd/RDDSuite.scala b/core/src/test/scala/org/apache/spark/rdd/RDDSuite.scala
index e94a1e76d410c..0e5625b7645d5 100644
--- a/core/src/test/scala/org/apache/spark/rdd/RDDSuite.scala
+++ b/core/src/test/scala/org/apache/spark/rdd/RDDSuite.scala
@@ -26,6 +26,8 @@ import org.apache.spark._
import org.apache.spark.SparkContext._
import org.apache.spark.util.Utils
+import org.apache.spark.rdd.RDDSuiteUtils._
+
class RDDSuite extends FunSuite with SharedSparkContext {
test("basic operations") {
@@ -585,14 +587,63 @@ class RDDSuite extends FunSuite with SharedSparkContext {
}
}
+ test("sortByKey") {
+ val data = sc.parallelize(Seq("5|50|A","4|60|C", "6|40|B"))
+
+ val col1 = Array("4|60|C", "5|50|A", "6|40|B")
+ val col2 = Array("6|40|B", "5|50|A", "4|60|C")
+ val col3 = Array("5|50|A", "6|40|B", "4|60|C")
+
+ assert(data.sortBy(_.split("\\|")(0)).collect() === col1)
+ assert(data.sortBy(_.split("\\|")(1)).collect() === col2)
+ assert(data.sortBy(_.split("\\|")(2)).collect() === col3)
+ }
+
+ test("sortByKey ascending parameter") {
+ val data = sc.parallelize(Seq("5|50|A","4|60|C", "6|40|B"))
+
+ val asc = Array("4|60|C", "5|50|A", "6|40|B")
+ val desc = Array("6|40|B", "5|50|A", "4|60|C")
+
+ assert(data.sortBy(_.split("\\|")(0), true).collect() === asc)
+ assert(data.sortBy(_.split("\\|")(0), false).collect() === desc)
+ }
+
+ test("sortByKey with explicit ordering") {
+ val data = sc.parallelize(Seq("Bob|Smith|50",
+ "Jane|Smith|40",
+ "Thomas|Williams|30",
+ "Karen|Williams|60"))
+
+ val ageOrdered = Array("Thomas|Williams|30",
+ "Jane|Smith|40",
+ "Bob|Smith|50",
+ "Karen|Williams|60")
+
+ // last name, then first name
+ val nameOrdered = Array("Bob|Smith|50",
+ "Jane|Smith|40",
+ "Karen|Williams|60",
+ "Thomas|Williams|30")
+
+ val parse = (s: String) => {
+ val split = s.split("\\|")
+ Person(split(0), split(1), split(2).toInt)
+ }
+
+ import scala.reflect.classTag
+ assert(data.sortBy(parse, true, 2)(AgeOrdering, classTag[Person]).collect() === ageOrdered)
+ assert(data.sortBy(parse, true, 2)(NameOrdering, classTag[Person]).collect() === nameOrdered)
+ }
+
test("intersection") {
val all = sc.parallelize(1 to 10)
val evens = sc.parallelize(2 to 10 by 2)
val intersection = Array(2, 4, 6, 8, 10)
// intersection is commutative
- assert(all.intersection(evens).collect.sorted === intersection)
- assert(evens.intersection(all).collect.sorted === intersection)
+ assert(all.intersection(evens).collect().sorted === intersection)
+ assert(evens.intersection(all).collect().sorted === intersection)
}
test("intersection strips duplicates in an input") {
@@ -600,8 +651,8 @@ class RDDSuite extends FunSuite with SharedSparkContext {
val b = sc.parallelize(Seq(1,1,2,3))
val intersection = Array(1,2,3)
- assert(a.intersection(b).collect.sorted === intersection)
- assert(b.intersection(a).collect.sorted === intersection)
+ assert(a.intersection(b).collect().sorted === intersection)
+ assert(b.intersection(a).collect().sorted === intersection)
}
test("zipWithIndex") {
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/api/java/JavaHiveSuite.scala b/core/src/test/scala/org/apache/spark/rdd/RDDSuiteUtils.scala
similarity index 51%
rename from sql/hive/src/test/scala/org/apache/spark/sql/hive/api/java/JavaHiveSuite.scala
rename to core/src/test/scala/org/apache/spark/rdd/RDDSuiteUtils.scala
index 9c5d7c81f7c09..4762fc17855ce 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/api/java/JavaHiveSuite.scala
+++ b/core/src/test/scala/org/apache/spark/rdd/RDDSuiteUtils.scala
@@ -15,27 +15,17 @@
* limitations under the License.
*/
-package org.apache.spark.sql.hive.api.java
+package org.apache.spark.rdd
-import org.scalatest.FunSuite
+object RDDSuiteUtils {
+ case class Person(first: String, last: String, age: Int)
-import org.apache.spark.api.java.JavaSparkContext
-import org.apache.spark.sql.test.TestSQLContext
-import org.apache.spark.sql.hive.test.TestHive
-
-// Implicits
-import scala.collection.JavaConversions._
-
-class JavaHiveSQLSuite extends FunSuite {
- ignore("SELECT * FROM src") {
- val javaCtx = new JavaSparkContext(TestSQLContext.sparkContext)
- // There is a little trickery here to avoid instantiating two HiveContexts in the same JVM
- val javaSqlCtx = new JavaHiveContext(javaCtx) {
- override val sqlContext = TestHive
- }
+ object AgeOrdering extends Ordering[Person] {
+ def compare(a:Person, b:Person) = a.age compare b.age
+ }
- assert(
- javaSqlCtx.hql("SELECT * FROM src").collect().map(_.getInt(0)) ===
- TestHive.sql("SELECT * FROM src").collect().map(_.getInt(0)).toSeq)
+ object NameOrdering extends Ordering[Person] {
+ def compare(a:Person, b:Person) =
+ implicitly[Ordering[Tuple2[String,String]]].compare((a.last, a.first), (b.last, b.first))
}
}
diff --git a/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
index 7506d56d7e26d..45368328297d3 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
@@ -29,6 +29,7 @@ import org.apache.spark._
import org.apache.spark.rdd.RDD
import org.apache.spark.scheduler.SchedulingMode.SchedulingMode
import org.apache.spark.storage.{BlockId, BlockManagerId, BlockManagerMaster}
+import org.apache.spark.util.CallSite
class BuggyDAGEventProcessActor extends Actor {
val state = 0
@@ -211,7 +212,7 @@ class DAGSchedulerSuite extends TestKit(ActorSystem("DAGSchedulerSuite")) with F
allowLocal: Boolean = false,
listener: JobListener = jobListener): Int = {
val jobId = scheduler.nextJobId.getAndIncrement()
- runEvent(JobSubmitted(jobId, rdd, func, partitions, allowLocal, null, listener))
+ runEvent(JobSubmitted(jobId, rdd, func, partitions, allowLocal, CallSite("", ""), listener))
jobId
}
@@ -251,7 +252,7 @@ class DAGSchedulerSuite extends TestKit(ActorSystem("DAGSchedulerSuite")) with F
override def toString = "DAGSchedulerSuite Local RDD"
}
val jobId = scheduler.nextJobId.getAndIncrement()
- runEvent(JobSubmitted(jobId, rdd, jobComputeFunc, Array(0), true, null, jobListener))
+ runEvent(JobSubmitted(jobId, rdd, jobComputeFunc, Array(0), true, CallSite("", ""), jobListener))
assert(results === Map(0 -> 42))
assertDataStructuresEmpty
}
@@ -265,7 +266,7 @@ class DAGSchedulerSuite extends TestKit(ActorSystem("DAGSchedulerSuite")) with F
override def toString = "DAGSchedulerSuite Local RDD"
}
val jobId = scheduler.nextJobId.getAndIncrement()
- runEvent(JobSubmitted(jobId, rdd, jobComputeFunc, Array(0), true, null, jobListener))
+ runEvent(JobSubmitted(jobId, rdd, jobComputeFunc, Array(0), true, CallSite("", ""), jobListener))
assert(results.size == 0)
assertDataStructuresEmpty
}
diff --git a/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
index be506e0287a16..abd7b22310f1a 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
@@ -239,11 +239,14 @@ class SparkListenerSuite extends FunSuite with LocalSparkContext with Matchers
checkNonZeroAvg(
taskInfoMetrics.map(_._2.executorDeserializeTime),
stageInfo + " executorDeserializeTime")
+
+ /* Test is disabled (SEE SPARK-2208)
if (stageInfo.rddInfos.exists(_.name == d4.name)) {
checkNonZeroAvg(
taskInfoMetrics.map(_._2.shuffleReadMetrics.get.fetchWaitTime),
stageInfo + " fetchWaitTime")
}
+ */
taskInfoMetrics.foreach { case (taskInfo, taskMetrics) =>
taskMetrics.resultSize should be > (0l)
diff --git a/core/src/test/scala/org/apache/spark/ui/jobs/JobProgressListenerSuite.scala b/core/src/test/scala/org/apache/spark/ui/jobs/JobProgressListenerSuite.scala
index 91b4c7b0dd962..c3a14f48de38e 100644
--- a/core/src/test/scala/org/apache/spark/ui/jobs/JobProgressListenerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/ui/jobs/JobProgressListenerSuite.scala
@@ -32,12 +32,12 @@ class JobProgressListenerSuite extends FunSuite with LocalSparkContext with Matc
val listener = new JobProgressListener(conf)
def createStageStartEvent(stageId: Int) = {
- val stageInfo = new StageInfo(stageId, stageId.toString, 0, null)
+ val stageInfo = new StageInfo(stageId, stageId.toString, 0, null, "")
SparkListenerStageSubmitted(stageInfo)
}
def createStageEndEvent(stageId: Int) = {
- val stageInfo = new StageInfo(stageId, stageId.toString, 0, null)
+ val stageInfo = new StageInfo(stageId, stageId.toString, 0, null, "")
SparkListenerStageCompleted(stageInfo)
}
diff --git a/core/src/test/scala/org/apache/spark/util/FileAppenderSuite.scala b/core/src/test/scala/org/apache/spark/util/FileAppenderSuite.scala
index 53d7f5c6072e6..02e228945bbd9 100644
--- a/core/src/test/scala/org/apache/spark/util/FileAppenderSuite.scala
+++ b/core/src/test/scala/org/apache/spark/util/FileAppenderSuite.scala
@@ -120,7 +120,7 @@ class FileAppenderSuite extends FunSuite with BeforeAndAfter with Logging {
// on SparkConf settings.
def testAppenderSelection[ExpectedAppender: ClassTag, ExpectedRollingPolicy](
- properties: Seq[(String, String)], expectedRollingPolicyParam: Long = -1): FileAppender = {
+ properties: Seq[(String, String)], expectedRollingPolicyParam: Long = -1): Unit = {
// Set spark conf properties
val conf = new SparkConf
@@ -129,8 +129,9 @@ class FileAppenderSuite extends FunSuite with BeforeAndAfter with Logging {
}
// Create and test file appender
- val inputStream = new PipedInputStream(new PipedOutputStream())
- val appender = FileAppender(inputStream, new File("stdout"), conf)
+ val testOutputStream = new PipedOutputStream()
+ val testInputStream = new PipedInputStream(testOutputStream)
+ val appender = FileAppender(testInputStream, testFile, conf)
assert(appender.isInstanceOf[ExpectedAppender])
assert(appender.getClass.getSimpleName ===
classTag[ExpectedAppender].runtimeClass.getSimpleName)
@@ -144,7 +145,8 @@ class FileAppenderSuite extends FunSuite with BeforeAndAfter with Logging {
}
assert(policyParam === expectedRollingPolicyParam)
}
- appender
+ testOutputStream.close()
+ appender.awaitTermination()
}
import RollingFileAppender._
diff --git a/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala b/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala
index 3031015256ec9..f72389b6b323f 100644
--- a/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala
+++ b/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala
@@ -117,6 +117,17 @@ class JsonProtocolSuite extends FunSuite {
testBlockId(StreamBlockId(1, 2L))
}
+ test("Backward compatibility") {
+ // StageInfo.details was added after 1.0.0.
+ val info = makeStageInfo(1, 2, 3, 4L, 5L)
+ assert(info.details.nonEmpty)
+ val newJson = JsonProtocol.stageInfoToJson(info)
+ val oldJson = newJson.removeField { case (field, _) => field == "Details" }
+ val newInfo = JsonProtocol.stageInfoFromJson(oldJson)
+ assert(info.name === newInfo.name)
+ assert("" === newInfo.details)
+ }
+
/** -------------------------- *
| Helper test running methods |
@@ -235,6 +246,7 @@ class JsonProtocolSuite extends FunSuite {
(0 until info1.rddInfos.size).foreach { i =>
assertEquals(info1.rddInfos(i), info2.rddInfos(i))
}
+ assert(info1.details === info2.details)
}
private def assertEquals(info1: RDDInfo, info2: RDDInfo) {
@@ -438,7 +450,7 @@ class JsonProtocolSuite extends FunSuite {
private def makeStageInfo(a: Int, b: Int, c: Int, d: Long, e: Long) = {
val rddInfos = (1 to a % 5).map { i => makeRddInfo(a % i, b % i, c % i, d % i, e % i) }
- new StageInfo(a, "greetings", b, rddInfos)
+ new StageInfo(a, "greetings", b, rddInfos, "details")
}
private def makeTaskInfo(a: Long, b: Int, c: Long) = {
diff --git a/docs/mllib-optimization.md b/docs/mllib-optimization.md
index 97e8f4e9661b6..ae9ede58e8e60 100644
--- a/docs/mllib-optimization.md
+++ b/docs/mllib-optimization.md
@@ -147,9 +147,9 @@ are developed, see the
linear methods
section for example.
-The SGD method
-[GradientDescent.runMiniBatchSGD](api/scala/index.html#org.apache.spark.mllib.optimization.GradientDescent)
-has the following parameters:
+The SGD class
+[GradientDescent](api/scala/index.html#org.apache.spark.mllib.optimization.GradientDescent)
+sets the following parameters:
* `Gradient` is a class that computes the stochastic gradient of the function
being optimized, i.e., with respect to a single training example, at the
@@ -171,7 +171,7 @@ each iteration, to compute the gradient direction.
Available algorithms for gradient descent:
-* [GradientDescent.runMiniBatchSGD](api/scala/index.html#org.apache.spark.mllib.optimization.GradientDescent)
+* [GradientDescent](api/scala/index.html#org.apache.spark.mllib.optimization.GradientDescent)
### L-BFGS
L-BFGS is currently only a low-level optimization primitive in `MLlib`. If you want to use L-BFGS in various
diff --git a/docs/programming-guide.md b/docs/programming-guide.md
index 79784682bfd1b..65d75b85efda6 100644
--- a/docs/programming-guide.md
+++ b/docs/programming-guide.md
@@ -377,13 +377,15 @@ Some notes on reading files with Spark:
* The `textFile` method also takes an optional second argument for controlling the number of slices of the file. By default, Spark creates one slice for each block of the file (blocks being 64MB by default in HDFS), but you can also ask for a higher number of slices by passing a larger value. Note that you cannot have fewer slices than blocks.
-Apart from reading files as a collection of lines,
-`SparkContext.wholeTextFiles` lets you read a directory containing multiple small text files, and returns each of them as (filename, content) pairs. This is in contrast with `textFile`, which would return one record per line in each file.
+Apart from text files, Spark's Python API also supports several other data formats:
-### SequenceFile and Hadoop InputFormats
+* `SparkContext.wholeTextFiles` lets you read a directory containing multiple small text files, and returns each of them as (filename, content) pairs. This is in contrast with `textFile`, which would return one record per line in each file.
+
+* `RDD.saveAsPickleFile` and `SparkContext.pickleFile` support saving an RDD in a simple format consisting of pickled Python objects. Batching is used on pickle serialization, with default batch size 10.
-In addition to reading text files, PySpark supports reading ```SequenceFile```
-and any arbitrary ```InputFormat```.
+* Details on reading `SequenceFile` and arbitrary Hadoop `InputFormat` are given below.
+
+### SequenceFile and Hadoop InputFormats
**Note** this feature is currently marked ```Experimental``` and is intended for advanced users. It may be replaced in future with read/write support based on SparkSQL, in which case SparkSQL is the preferred approach.
@@ -760,6 +762,11 @@ val counts = pairs.reduceByKey((a, b) => a + b)
We could also use `counts.sortByKey()`, for example, to sort the pairs alphabetically, and finally
`counts.collect()` to bring them back to the driver program as an array of objects.
+**Note:** when using custom objects as the key in key-value pair operations, you must be sure that a
+custom `equals()` method is accompanied with a matching `hashCode()` method. For full details, see
+the contract outlined in the [Object.hashCode()
+documentation](http://docs.oracle.com/javase/7/docs/api/java/lang/Object.html#hashCode()).
+
@@ -792,6 +799,10 @@ JavaPairRDD counts = pairs.reduceByKey((a, b) -> a + b);
We could also use `counts.sortByKey()`, for example, to sort the pairs alphabetically, and finally
`counts.collect()` to bring them back to the driver program as an array of objects.
+**Note:** when using custom objects as the key in key-value pair operations, you must be sure that a
+custom `equals()` method is accompanied with a matching `hashCode()` method. For full details, see
+the contract outlined in the [Object.hashCode()
+documentation](http://docs.oracle.com/javase/7/docs/api/java/lang/Object.html#hashCode()).
@@ -888,7 +899,7 @@ for details.
| reduceByKey(func, [numTasks]) |
- When called on a dataset of (K, V) pairs, returns a dataset of (K, V) pairs where the values for each key are aggregated using the given reduce function. Like in groupByKey, the number of reduce tasks is configurable through an optional second argument. |
+ When called on a dataset of (K, V) pairs, returns a dataset of (K, V) pairs where the values for each key are aggregated using the given reduce function func, which must be of type (V,V) => V. Like in groupByKey, the number of reduce tasks is configurable through an optional second argument. |
| aggregateByKey(zeroValue)(seqOp, combOp, [numTasks]) |
@@ -1056,7 +1067,10 @@ storage levels is:
Store RDD in serialized format in Tachyon.
Compared to MEMORY_ONLY_SER, OFF_HEAP reduces garbage collection overhead and allows executors
to be smaller and to share a pool of memory, making it attractive in environments with
- large heaps or multiple concurrent applications.
+ large heaps or multiple concurrent applications. Furthermore, as the RDDs reside in Tachyon,
+ the crash of an executor does not lead to losing the in-memory cache. In this mode, the memory
+ in Tachyon is discardable. Thus, Tachyon does not attempt to reconstruct a block that it evicts
+ from memory.
|
diff --git a/docs/running-on-yarn.md b/docs/running-on-yarn.md
index af1788f2aa151..fecd8f2cc2d48 100644
--- a/docs/running-on-yarn.md
+++ b/docs/running-on-yarn.md
@@ -67,6 +67,34 @@ Most of the configs are the same for Spark on YARN as for other deployment modes
The address of the Spark history server (i.e. host.com:18080). The address should not contain a scheme (http://). Defaults to not being set since the history server is an optional service. This address is given to the YARN ResourceManager when the Spark application finishes to link the application from the ResourceManager UI to the Spark history server UI.
+
+ spark.yarn.dist.archives |
+ (none) |
+
+ Comma separated list of archives to be extracted into the working directory of each executor.
+ |
+
+
+ spark.yarn.dist.files |
+ (none) |
+
+ Comma-separated list of files to be placed in the working directory of each executor.
+ |
+ |
+
+ spark.yarn.executor.memoryOverhead |
+ 384 |
+
+ The amount of off heap memory (in megabytes) to be allocated per executor. This is memory that accounts for things like VM overheads, interned strings, other native overheads, etc.
+ |
+
+
+ spark.yarn.driver.memoryOverhead |
+ 384 |
+
+ The amount of off heap memory (in megabytes) to be allocated per driver. This is memory that accounts for things like VM overheads, interned strings, other native overheads, etc.
+ |
+
By default, Spark on YARN will use a Spark jar installed locally, but the Spark JAR can also be in a world-readable location on HDFS. This allows YARN to cache it on nodes so that it doesn't need to be distributed each time an application runs. To point to a JAR on HDFS, `export SPARK_JAR=hdfs:///some/path`.
diff --git a/docs/sql-programming-guide.md b/docs/sql-programming-guide.md
index 4623bb4247d77..522c83884ef42 100644
--- a/docs/sql-programming-guide.md
+++ b/docs/sql-programming-guide.md
@@ -17,20 +17,20 @@ Spark. At the core of this component is a new type of RDD,
[Row](api/scala/index.html#org.apache.spark.sql.catalyst.expressions.Row) objects along with
a schema that describes the data types of each column in the row. A SchemaRDD is similar to a table
in a traditional relational database. A SchemaRDD can be created from an existing RDD, [Parquet](http://parquet.io)
-file, or by running HiveQL against data stored in [Apache Hive](http://hive.apache.org/).
+file, a JSON dataset, or by running HiveQL against data stored in [Apache Hive](http://hive.apache.org/).
All of the examples on this page use sample data included in the Spark distribution and can be run in the `spark-shell`.
-Spark SQL allows relational queries expressed in SQL, HiveQL, or Scala to be executed using
+Spark SQL allows relational queries expressed in SQL or HiveQL to be executed using
Spark. At the core of this component is a new type of RDD,
[JavaSchemaRDD](api/scala/index.html#org.apache.spark.sql.api.java.JavaSchemaRDD). JavaSchemaRDDs are composed
[Row](api/scala/index.html#org.apache.spark.sql.api.java.Row) objects along with
a schema that describes the data types of each column in the row. A JavaSchemaRDD is similar to a table
in a traditional relational database. A JavaSchemaRDD can be created from an existing RDD, [Parquet](http://parquet.io)
-file, or by running HiveQL against data stored in [Apache Hive](http://hive.apache.org/).
+file, a JSON dataset, or by running HiveQL against data stored in [Apache Hive](http://hive.apache.org/).
@@ -41,7 +41,7 @@ Spark. At the core of this component is a new type of RDD,
[Row](api/python/pyspark.sql.Row-class.html) objects along with
a schema that describes the data types of each column in the row. A SchemaRDD is similar to a table
in a traditional relational database. A SchemaRDD can be created from an existing RDD, [Parquet](http://parquet.io)
-file, or by running HiveQL against data stored in [Apache Hive](http://hive.apache.org/).
+file, a JSON dataset, or by running HiveQL against data stored in [Apache Hive](http://hive.apache.org/).
All of the examples on this page use sample data included in the Spark distribution and can be run in the `pyspark` shell.
@@ -64,8 +64,8 @@ descendants. To create a basic SQLContext, all you need is a SparkContext.
val sc: SparkContext // An existing SparkContext.
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
-// Importing the SQL context gives access to all the public SQL functions and implicit conversions.
-import sqlContext._
+// createSchemaRDD is used to implicitly convert an RDD to a SchemaRDD.
+import sqlContext.createSchemaRDD
{% endhighlight %}
{% highlight scala %}
-val sqlContext = new org.apache.spark.sql.SQLContext(sc)
-import sqlContext._
+// sqlContext from the previous example is used in this example.
+// createSchemaRDD is used to implicitly convert an RDD to a SchemaRDD.
+import sqlContext.createSchemaRDD
val people: RDD[Person] = ... // An RDD of case class objects, from the previous example.
-// The RDD is implicitly converted to a SchemaRDD, allowing it to be stored using Parquet.
+// The RDD is implicitly converted to a SchemaRDD by createSchemaRDD, allowing it to be stored using Parquet.
people.saveAsParquetFile("people.parquet")
// Read in the parquet file created above. Parquet files are self-describing so the schema is preserved.
-// The result of loading a Parquet file is also a JavaSchemaRDD.
+// The result of loading a Parquet file is also a SchemaRDD.
val parquetFile = sqlContext.parquetFile("people.parquet")
//Parquet files can also be registered as tables and then used in SQL statements.
parquetFile.registerAsTable("parquetFile")
-val teenagers = sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19")
-teenagers.collect().foreach(println)
+val teenagers = sqlContext.sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19")
+teenagers.map(t => "Name: " + t(0)).collect().foreach(println)
{% endhighlight %}
@@ -275,6 +302,7 @@ teenagers.collect().foreach(println)
{% highlight java %}
+// sqlContext from the previous example is used in this example.
JavaSchemaRDD schemaPeople = ... // The JavaSchemaRDD from the previous example.
@@ -283,13 +311,16 @@ schemaPeople.saveAsParquetFile("people.parquet");
// Read in the Parquet file created above. Parquet files are self-describing so the schema is preserved.
// The result of loading a parquet file is also a JavaSchemaRDD.
-JavaSchemaRDD parquetFile = sqlCtx.parquetFile("people.parquet");
+JavaSchemaRDD parquetFile = sqlContext.parquetFile("people.parquet");
//Parquet files can also be registered as tables and then used in SQL statements.
parquetFile.registerAsTable("parquetFile");
-JavaSchemaRDD teenagers = sqlCtx.sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19");
-
-
+JavaSchemaRDD teenagers = sqlContext.sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19");
+List teenagerNames = teenagers.map(new Function() {
+ public String call(Row row) {
+ return "Name: " + row.getString(0);
+ }
+}).collect();
{% endhighlight %}
@@ -297,50 +328,149 @@ JavaSchemaRDD teenagers = sqlCtx.sql("SELECT name FROM parquetFile WHERE age >=
{% highlight python %}
+# sqlContext from the previous example is used in this example.
-peopleTable # The SchemaRDD from the previous example.
+schemaPeople # The SchemaRDD from the previous example.
# SchemaRDDs can be saved as Parquet files, maintaining the schema information.
-peopleTable.saveAsParquetFile("people.parquet")
+schemaPeople.saveAsParquetFile("people.parquet")
# Read in the Parquet file created above. Parquet files are self-describing so the schema is preserved.
# The result of loading a parquet file is also a SchemaRDD.
-parquetFile = sqlCtx.parquetFile("people.parquet")
+parquetFile = sqlContext.parquetFile("people.parquet")
# Parquet files can also be registered as tables and then used in SQL statements.
parquetFile.registerAsTable("parquetFile");
-teenagers = sqlCtx.sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19")
-
+teenagers = sqlContext.sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19")
+teenNames = teenagers.map(lambda p: "Name: " + p.name)
+for teenName in teenNames.collect():
+ print teenName
{% endhighlight %}
-## Writing Language-Integrated Relational Queries
+## JSON Datasets
+
-**Language-Integrated queries are currently only supported in Scala.**
+
+Spark SQL can automatically infer the schema of a JSON dataset and load it as a SchemaRDD.
+This conversion can be done using one of two methods in a SQLContext:
-Spark SQL also supports a domain specific language for writing queries. Once again,
-using the data from the above examples:
+* `jsonFile` - loads data from a directory of JSON files where each line of the files is a JSON object.
+* `jsonRdd` - loads data from an existing RDD where each element of the RDD is a string containing a JSON object.
{% highlight scala %}
+// sc is an existing SparkContext.
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
-import sqlContext._
-val people: RDD[Person] = ... // An RDD of case class objects, from the first example.
-// The following is the same as 'SELECT name FROM people WHERE age >= 10 AND age <= 19'
-val teenagers = people.where('age >= 10).where('age <= 19).select('name)
+// A JSON dataset is pointed to by path.
+// The path can be either a single text file or a directory storing text files.
+val path = "examples/src/main/resources/people.json"
+// Create a SchemaRDD from the file(s) pointed to by path
+val people = sqlContext.jsonFile(path)
+
+// The inferred schema can be visualized using the printSchema() method.
+people.printSchema()
+// root
+// |-- age: IntegerType
+// |-- name: StringType
+
+// Register this SchemaRDD as a table.
+people.registerAsTable("people")
+
+// SQL statements can be run by using the sql methods provided by sqlContext.
+val teenagers = sqlContext.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19")
+
+// Alternatively, a SchemaRDD can be created for a JSON dataset represented by
+// an RDD[String] storing one JSON object per string.
+val anotherPeopleRDD = sc.parallelize(
+ """{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}""" :: Nil)
+val anotherPeople = sqlContext.jsonRDD(anotherPeopleRDD)
{% endhighlight %}
-The DSL uses Scala symbols to represent columns in the underlying table, which are identifiers
-prefixed with a tick (`'`). Implicit conversions turn these symbols into expressions that are
-evaluated by the SQL execution engine. A full list of the functions supported can be found in the
-[ScalaDoc](api/scala/index.html#org.apache.spark.sql.SchemaRDD).
+
-
+
+Spark SQL can automatically infer the schema of a JSON dataset and load it as a JavaSchemaRDD.
+This conversion can be done using one of two methods in a JavaSQLContext :
-# Hive Support
+* `jsonFile` - loads data from a directory of JSON files where each line of the files is a JSON object.
+* `jsonRdd` - loads data from an existing RDD where each element of the RDD is a string containing a JSON object.
+
+{% highlight java %}
+// sc is an existing JavaSparkContext.
+JavaSQLContext sqlContext = new org.apache.spark.sql.api.java.JavaSQLContext(sc);
+
+// A JSON dataset is pointed to by path.
+// The path can be either a single text file or a directory storing text files.
+String path = "examples/src/main/resources/people.json";
+// Create a JavaSchemaRDD from the file(s) pointed to by path
+JavaSchemaRDD people = sqlContext.jsonFile(path);
+
+// The inferred schema can be visualized using the printSchema() method.
+people.printSchema();
+// root
+// |-- age: IntegerType
+// |-- name: StringType
+
+// Register this JavaSchemaRDD as a table.
+people.registerAsTable("people");
+
+// SQL statements can be run by using the sql methods provided by sqlContext.
+JavaSchemaRDD teenagers = sqlContext.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19");
+
+// Alternatively, a JavaSchemaRDD can be created for a JSON dataset represented by
+// an RDD[String] storing one JSON object per string.
+List jsonData = Arrays.asList(
+ "{\"name\":\"Yin\",\"address\":{\"city\":\"Columbus\",\"state\":\"Ohio\"}}");
+JavaRDD anotherPeopleRDD = sc.parallelize(jsonData);
+JavaSchemaRDD anotherPeople = sqlContext.jsonRDD(anotherPeopleRDD);
+{% endhighlight %}
+
+
+
+Spark SQL can automatically infer the schema of a JSON dataset and load it as a SchemaRDD.
+This conversion can be done using one of two methods in a SQLContext:
+
+* `jsonFile` - loads data from a directory of JSON files where each line of the files is a JSON object.
+* `jsonRdd` - loads data from an existing RDD where each element of the RDD is a string containing a JSON object.
+
+{% highlight python %}
+# sc is an existing SparkContext.
+from pyspark.sql import SQLContext
+sqlContext = SQLContext(sc)
+
+# A JSON dataset is pointed to by path.
+# The path can be either a single text file or a directory storing text files.
+path = "examples/src/main/resources/people.json"
+# Create a SchemaRDD from the file(s) pointed to by path
+people = sqlContext.jsonFile(path)
+
+# The inferred schema can be visualized using the printSchema() method.
+people.printSchema()
+# root
+# |-- age: IntegerType
+# |-- name: StringType
+
+# Register this SchemaRDD as a table.
+people.registerAsTable("people")
+
+# SQL statements can be run by using the sql methods provided by sqlContext.
+teenagers = sqlContext.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19")
+
+# Alternatively, a SchemaRDD can be created for a JSON dataset represented by
+# an RDD[String] storing one JSON object per string.
+anotherPeopleRDD = sc.parallelize([
+ '{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}'])
+anotherPeople = sqlContext.jsonRDD(anotherPeopleRDD)
+{% endhighlight %}
+
+
+