[SPARK-34374][SQL][DSTREAM] Use standard methods to extract keys or v…

…alues from a Map ### What changes were proposed in this pull request? Use standard methods to extract `keys` or `values` from a `Map`, it's semantically consistent and use the `DefaultKeySet` and `DefaultValuesIterable` instead of a manual loop. **Before** ``` map.map(_._1) map.map(_._2) ``` **After** ``` map.keys map.values ``` ### Why are the changes needed? Code Simpilefications. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Pass the Jenkins or GitHub Action Closes #31484 from LuciferYang/keys-and-values. Authored-by: yangjie01 <[email protected]> Signed-off-by: Sean Owen <[email protected]>
apache · Feb 8, 2021 · 777d51e · 777d51e
1 parent 3b26bc2
commit 777d51e
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 3 deletions.
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala
@@ -406,8 +406,7 @@ object PreprocessTableInsertion extends Rule[LogicalPlan] {
       catalogTable.get.tracksPartitionsInCatalog
     if (partitionsTrackedByCatalog && normalizedPartSpec.nonEmpty) {
       // empty partition column value
-      if (normalizedPartSpec.map(_._2)
-          .filter(_.isDefined).map(_.get).exists(v => v != null && v.isEmpty)) {
+      if (normalizedPartSpec.values.flatten.exists(v => v != null && v.isEmpty)) {
         val spec = normalizedPartSpec.map(p => p._1 + "=" + p._2).mkString("[", ", ", "]")
         throw new AnalysisException(
           s"Partition spec is invalid. The spec ($spec) contains an empty partition column value")

diff --git a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverSchedulingPolicy.scala b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverSchedulingPolicy.scala
@@ -128,7 +128,7 @@ private[streaming] class ReceiverSchedulingPolicy {
     }
 
     // Assign idle executors to receivers that have less executors
-    val idleExecutors = numReceiversOnExecutor.filter(_._2 == 0).map(_._1)
+    val idleExecutors = numReceiversOnExecutor.filter(_._2 == 0).keys
     for (executor <- idleExecutors) {
       // Assign an idle executor to the receiver that has least candidate executors.
       val leastScheduledExecutors = scheduledLocations.minBy(_.size)