You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2018/05/18 07:33:42 UTC
spark git commit: [SPARK-24277][SQL] Code clean up in SQL module:
HadoopMapReduceCommitProtocol
Repository: spark
Updated Branches:
refs/heads/master d4a0895c6 -> 7b2dca5b1
[SPARK-24277][SQL] Code clean up in SQL module: HadoopMapReduceCommitProtocol
## What changes were proposed in this pull request?
In HadoopMapReduceCommitProtocol and FileFormatWriter, there are unnecessary settings in hadoop configuration.
Also clean up some code in SQL module.
## How was this patch tested?
Unit test
Author: Gengliang Wang <ge...@databricks.com>
Closes #21329 from gengliangwang/codeCleanWrite.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/7b2dca5b
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/7b2dca5b
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/7b2dca5b
Branch: refs/heads/master
Commit: 7b2dca5b12164b787ec4e8e7e9f92c60a3f9563e
Parents: d4a0895
Author: Gengliang Wang <ge...@databricks.com>
Authored: Fri May 18 15:32:29 2018 +0800
Committer: Wenchen Fan <we...@databricks.com>
Committed: Fri May 18 15:32:29 2018 +0800
----------------------------------------------------------------------
.../internal/io/HadoopMapReduceCommitProtocol.scala | 15 +++------------
.../execution/datasources/orc/OrcColumnVector.java | 6 +-----
.../parquet/VectorizedRleValuesReader.java | 4 ++--
.../scala/org/apache/spark/sql/api/r/SQLUtils.scala | 2 +-
.../apache/spark/sql/execution/command/views.scala | 10 ++++------
.../sql/execution/datasources/FileFormatWriter.scala | 11 +++++------
.../sql/execution/ui/SQLAppStatusListener.scala | 2 +-
7 files changed, 17 insertions(+), 33 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/7b2dca5b/core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala b/core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala
index 3e60c50..163511b 100644
--- a/core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala
+++ b/core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala
@@ -145,18 +145,9 @@ class HadoopMapReduceCommitProtocol(
}
override def setupJob(jobContext: JobContext): Unit = {
- // Setup IDs
- val jobId = SparkHadoopWriterUtils.createJobID(new Date, 0)
- val taskId = new TaskID(jobId, TaskType.MAP, 0)
- val taskAttemptId = new TaskAttemptID(taskId, 0)
-
- // Set up the configuration object
- jobContext.getConfiguration.set("mapreduce.job.id", jobId.toString)
- jobContext.getConfiguration.set("mapreduce.task.id", taskAttemptId.getTaskID.toString)
- jobContext.getConfiguration.set("mapreduce.task.attempt.id", taskAttemptId.toString)
- jobContext.getConfiguration.setBoolean("mapreduce.task.ismap", true)
- jobContext.getConfiguration.setInt("mapreduce.task.partition", 0)
-
+ // Create a dummy [[TaskAttemptContextImpl]] with configuration to get [[OutputCommitter]]
+ // instance on Spark driver. Note that the job/task/attampt id doesn't matter here.
+ val taskAttemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
val taskAttemptContext = new TaskAttemptContextImpl(jobContext.getConfiguration, taskAttemptId)
committer = setupCommitter(taskAttemptContext)
committer.setupJob(jobContext)
http://git-wip-us.apache.org/repos/asf/spark/blob/7b2dca5b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcColumnVector.java
----------------------------------------------------------------------
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcColumnVector.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcColumnVector.java
index 12f4d65..fcf73e8 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcColumnVector.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcColumnVector.java
@@ -47,11 +47,7 @@ public class OrcColumnVector extends org.apache.spark.sql.vectorized.ColumnVecto
OrcColumnVector(DataType type, ColumnVector vector) {
super(type);
- if (type instanceof TimestampType) {
- isTimestamp = true;
- } else {
- isTimestamp = false;
- }
+ isTimestamp = type instanceof TimestampType;
baseData = vector;
if (vector instanceof LongColumnVector) {
http://git-wip-us.apache.org/repos/asf/spark/blob/7b2dca5b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
----------------------------------------------------------------------
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
index fe3d31a..de0d65a 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
@@ -571,7 +571,7 @@ public final class VectorizedRleValuesReader extends ValuesReader
int ch3 = in.read();
int ch2 = in.read();
int ch1 = in.read();
- return ((ch1 << 24) + (ch2 << 16) + (ch3 << 8) + (ch4 << 0));
+ return ((ch1 << 24) + (ch2 << 16) + (ch3 << 8) + (ch4));
}
/**
@@ -592,7 +592,7 @@ public final class VectorizedRleValuesReader extends ValuesReader
int ch3 = in.read();
int ch2 = in.read();
int ch1 = in.read();
- return (ch1 << 16) + (ch2 << 8) + (ch3 << 0);
+ return (ch1 << 16) + (ch2 << 8) + (ch3);
}
case 4: {
return readIntLittleEndian();
http://git-wip-us.apache.org/repos/asf/spark/blob/7b2dca5b/sql/core/src/main/scala/org/apache/spark/sql/api/r/SQLUtils.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/api/r/SQLUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/api/r/SQLUtils.scala
index af20764..265a84b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/api/r/SQLUtils.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/api/r/SQLUtils.scala
@@ -188,7 +188,7 @@ private[sql] object SQLUtils extends Logging {
dataType match {
case 's' =>
// Read StructType for DataFrame
- val fields = SerDe.readList(dis, jvmObjectTracker = null).asInstanceOf[Array[Object]]
+ val fields = SerDe.readList(dis, jvmObjectTracker = null)
Row.fromSeq(fields)
case _ => null
}
http://git-wip-us.apache.org/repos/asf/spark/blob/7b2dca5b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
index 5172f32..6373584 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala
@@ -410,12 +410,10 @@ object ViewHelper {
}
// Detect cyclic references from subqueries.
- plan.expressions.foreach { expr =>
- expr match {
- case s: SubqueryExpression =>
- checkCyclicViewReference(s.plan, path, viewIdent)
- case _ => // Do nothing.
- }
+ plan.expressions.foreach {
+ case s: SubqueryExpression =>
+ checkCyclicViewReference(s.plan, path, viewIdent)
+ case _ => // Do nothing.
}
}
}
http://git-wip-us.apache.org/repos/asf/spark/blob/7b2dca5b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
index 401597f..681bb1d 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
@@ -244,18 +244,17 @@ object FileFormatWriter extends Logging {
iterator: Iterator[InternalRow]): WriteTaskResult = {
val jobId = SparkHadoopWriterUtils.createJobID(new Date, sparkStageId)
- val taskId = new TaskID(jobId, TaskType.MAP, sparkPartitionId)
- val taskAttemptId = new TaskAttemptID(taskId, sparkAttemptNumber)
// Set up the attempt context required to use in the output committer.
val taskAttemptContext: TaskAttemptContext = {
+ val taskId = new TaskID(jobId, TaskType.MAP, sparkPartitionId)
+ val taskAttemptId = new TaskAttemptID(taskId, sparkAttemptNumber)
// Set up the configuration object
val hadoopConf = description.serializableHadoopConf.value
hadoopConf.set("mapreduce.job.id", jobId.toString)
hadoopConf.set("mapreduce.task.id", taskAttemptId.getTaskID.toString)
hadoopConf.set("mapreduce.task.attempt.id", taskAttemptId.toString)
hadoopConf.setBoolean("mapreduce.task.ismap", true)
- hadoopConf.setInt("mapreduce.task.partition", 0)
new TaskAttemptContextImpl(hadoopConf, taskAttemptId)
}
@@ -378,7 +377,7 @@ object FileFormatWriter extends Logging {
dataSchema = description.dataColumns.toStructType,
context = taskAttemptContext)
- statsTrackers.map(_.newFile(currentPath))
+ statsTrackers.foreach(_.newFile(currentPath))
}
override def execute(iter: Iterator[InternalRow]): ExecutedWriteSummary = {
@@ -429,10 +428,10 @@ object FileFormatWriter extends Logging {
committer: FileCommitProtocol) extends ExecuteWriteTask {
/** Flag saying whether or not the data to be written out is partitioned. */
- val isPartitioned = desc.partitionColumns.nonEmpty
+ private val isPartitioned = desc.partitionColumns.nonEmpty
/** Flag saying whether or not the data to be written out is bucketed. */
- val isBucketed = desc.bucketIdExpression.isDefined
+ private val isBucketed = desc.bucketIdExpression.isDefined
assert(isPartitioned || isBucketed,
s"""DynamicPartitionWriteTask should be used for writing out data that's either
http://git-wip-us.apache.org/repos/asf/spark/blob/7b2dca5b/sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListener.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListener.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListener.scala
index d254af4..2c4d0bc 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListener.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListener.scala
@@ -170,7 +170,7 @@ class SQLAppStatusListener(
.filter { case (id, _) => metricIds.contains(id) }
.groupBy(_._1)
.map { case (id, values) =>
- id -> SQLMetrics.stringValue(metricTypes(id), values.map(_._2).toSeq)
+ id -> SQLMetrics.stringValue(metricTypes(id), values.map(_._2))
}
// Check the execution again for whether the aggregated metrics data has been calculated.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org