You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sedona.apache.org by ji...@apache.org on 2023/09/29 07:50:12 UTC
[sedona] branch master updated: [SEDONA-405] Replace Metric with LongAccumulator to reduce memory overhead for spatial join (#1041)
This is an automated email from the ASF dual-hosted git repository.
jiayu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/sedona.git
The following commit(s) were added to refs/heads/master by this push:
new db0795302 [SEDONA-405] Replace Metric with LongAccumulator to reduce memory overhead for spatial join (#1041)
db0795302 is described below
commit db07953021619b4bb3878fd212d34000e87d6b6d
Author: Jia Yu <ji...@apache.org>
AuthorDate: Fri Sep 29 00:50:06 2023 -0700
[SEDONA-405] Replace Metric with LongAccumulator to reduce memory overhead for spatial join (#1041)
---
.../joinJudgement/DynamicIndexLookupJudgement.java | 10 ++--
.../sedona/core/joinJudgement/JudgementBase.java | 12 ++---
.../joinJudgement/LeftIndexLookupJudgement.java | 10 ++--
.../core/joinJudgement/NestedLoopJudgement.java | 10 ++--
.../joinJudgement/RightIndexLookupJudgement.java | 10 ++--
.../sedona/core/spatialOperator/JoinQuery.java | 10 ++--
.../org/apache/sedona/core/monitoring/Metric.scala | 53 ----------------------
.../apache/sedona/core/monitoring/Metrics.scala | 5 +-
.../sedona/core/monitoring/MetricSuite.scala | 52 ---------------------
9 files changed, 34 insertions(+), 138 deletions(-)
diff --git a/spark/common/src/main/java/org/apache/sedona/core/joinJudgement/DynamicIndexLookupJudgement.java b/spark/common/src/main/java/org/apache/sedona/core/joinJudgement/DynamicIndexLookupJudgement.java
index 7767c9699..45a2cbe3c 100644
--- a/spark/common/src/main/java/org/apache/sedona/core/joinJudgement/DynamicIndexLookupJudgement.java
+++ b/spark/common/src/main/java/org/apache/sedona/core/joinJudgement/DynamicIndexLookupJudgement.java
@@ -22,10 +22,10 @@ package org.apache.sedona.core.joinJudgement;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.sedona.core.enums.IndexType;
import org.apache.sedona.core.enums.JoinBuildSide;
-import org.apache.sedona.core.monitoring.Metric;
import org.apache.sedona.core.spatialOperator.SpatialPredicate;
import org.apache.sedona.core.utils.TimeUtils;
import org.apache.spark.api.java.function.FlatMapFunction2;
+import org.apache.spark.util.LongAccumulator;
import org.locationtech.jts.geom.Envelope;
import org.locationtech.jts.geom.Geometry;
import org.locationtech.jts.index.SpatialIndex;
@@ -50,10 +50,10 @@ public class DynamicIndexLookupJudgement<T extends Geometry, U extends Geometry>
SpatialPredicate spatialPredicate,
IndexType indexType,
JoinBuildSide joinBuildSide,
- Metric buildCount,
- Metric streamCount,
- Metric resultCount,
- Metric candidateCount)
+ LongAccumulator buildCount,
+ LongAccumulator streamCount,
+ LongAccumulator resultCount,
+ LongAccumulator candidateCount)
{
super(spatialPredicate, buildCount, streamCount, resultCount, candidateCount);
this.indexType = indexType;
diff --git a/spark/common/src/main/java/org/apache/sedona/core/joinJudgement/JudgementBase.java b/spark/common/src/main/java/org/apache/sedona/core/joinJudgement/JudgementBase.java
index 8b2d869fe..4e4a8a305 100644
--- a/spark/common/src/main/java/org/apache/sedona/core/joinJudgement/JudgementBase.java
+++ b/spark/common/src/main/java/org/apache/sedona/core/joinJudgement/JudgementBase.java
@@ -23,10 +23,10 @@ import org.apache.commons.lang3.tuple.Pair;
import org.apache.log4j.Level;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
-import org.apache.sedona.core.monitoring.Metric;
import org.apache.sedona.core.spatialOperator.SpatialPredicate;
import org.apache.sedona.core.spatialOperator.SpatialPredicateEvaluators;
import org.apache.spark.TaskContext;
+import org.apache.spark.util.LongAccumulator;
import org.locationtech.jts.geom.Geometry;
import org.locationtech.jts.index.SpatialIndex;
@@ -49,10 +49,10 @@ abstract class JudgementBase<T extends Geometry, U extends Geometry>
private final SpatialPredicate spatialPredicate;
private transient SpatialPredicateEvaluators.SpatialPredicateEvaluator evaluator;
- protected final Metric buildCount;
- protected final Metric streamCount;
- protected final Metric resultCount;
- protected final Metric candidateCount;
+ protected final LongAccumulator buildCount;
+ protected final LongAccumulator streamCount;
+ protected final LongAccumulator resultCount;
+ protected final LongAccumulator candidateCount;
private int shapeCnt;
@@ -69,7 +69,7 @@ abstract class JudgementBase<T extends Geometry, U extends Geometry>
* @param resultCount num of join results
* @param candidateCount num of candidate pairs to be refined by their real geometries
*/
- protected JudgementBase(SpatialPredicate spatialPredicate, Metric buildCount, Metric streamCount, Metric resultCount, Metric candidateCount)
+ protected JudgementBase(SpatialPredicate spatialPredicate, LongAccumulator buildCount, LongAccumulator streamCount, LongAccumulator resultCount, LongAccumulator candidateCount)
{
this.spatialPredicate = spatialPredicate;
this.buildCount = buildCount;
diff --git a/spark/common/src/main/java/org/apache/sedona/core/joinJudgement/LeftIndexLookupJudgement.java b/spark/common/src/main/java/org/apache/sedona/core/joinJudgement/LeftIndexLookupJudgement.java
index 1514571a4..09841af04 100644
--- a/spark/common/src/main/java/org/apache/sedona/core/joinJudgement/LeftIndexLookupJudgement.java
+++ b/spark/common/src/main/java/org/apache/sedona/core/joinJudgement/LeftIndexLookupJudgement.java
@@ -20,9 +20,9 @@
package org.apache.sedona.core.joinJudgement;
import org.apache.commons.lang3.tuple.Pair;
-import org.apache.sedona.core.monitoring.Metric;
import org.apache.sedona.core.spatialOperator.SpatialPredicate;
import org.apache.spark.api.java.function.FlatMapFunction2;
+import org.apache.spark.util.LongAccumulator;
import org.locationtech.jts.geom.Geometry;
import org.locationtech.jts.index.SpatialIndex;
@@ -39,10 +39,10 @@ public class LeftIndexLookupJudgement<T extends Geometry, U extends Geometry>
* @see JudgementBase
*/
public LeftIndexLookupJudgement(SpatialPredicate spatialPredicate,
- Metric buildCount,
- Metric streamCount,
- Metric resultCount,
- Metric candidateCount)
+ LongAccumulator buildCount,
+ LongAccumulator streamCount,
+ LongAccumulator resultCount,
+ LongAccumulator candidateCount)
{
super(spatialPredicate, buildCount, streamCount, resultCount, candidateCount);
}
diff --git a/spark/common/src/main/java/org/apache/sedona/core/joinJudgement/NestedLoopJudgement.java b/spark/common/src/main/java/org/apache/sedona/core/joinJudgement/NestedLoopJudgement.java
index 66138895b..f2103ca13 100644
--- a/spark/common/src/main/java/org/apache/sedona/core/joinJudgement/NestedLoopJudgement.java
+++ b/spark/common/src/main/java/org/apache/sedona/core/joinJudgement/NestedLoopJudgement.java
@@ -20,9 +20,9 @@
package org.apache.sedona.core.joinJudgement;
import org.apache.commons.lang3.tuple.Pair;
-import org.apache.sedona.core.monitoring.Metric;
import org.apache.sedona.core.spatialOperator.SpatialPredicate;
import org.apache.spark.api.java.function.FlatMapFunction2;
+import org.apache.spark.util.LongAccumulator;
import org.locationtech.jts.geom.Geometry;
import java.io.Serializable;
@@ -39,10 +39,10 @@ public class NestedLoopJudgement<T extends Geometry, U extends Geometry>
* @see JudgementBase
*/
public NestedLoopJudgement(SpatialPredicate spatialPredicate,
- Metric buildCount,
- Metric streamCount,
- Metric resultCount,
- Metric candidateCount)
+ LongAccumulator buildCount,
+ LongAccumulator streamCount,
+ LongAccumulator resultCount,
+ LongAccumulator candidateCount)
{
super(spatialPredicate, buildCount, streamCount, resultCount, candidateCount);
}
diff --git a/spark/common/src/main/java/org/apache/sedona/core/joinJudgement/RightIndexLookupJudgement.java b/spark/common/src/main/java/org/apache/sedona/core/joinJudgement/RightIndexLookupJudgement.java
index 9cce4dc62..e7edd4607 100644
--- a/spark/common/src/main/java/org/apache/sedona/core/joinJudgement/RightIndexLookupJudgement.java
+++ b/spark/common/src/main/java/org/apache/sedona/core/joinJudgement/RightIndexLookupJudgement.java
@@ -20,9 +20,9 @@
package org.apache.sedona.core.joinJudgement;
import org.apache.commons.lang3.tuple.Pair;
-import org.apache.sedona.core.monitoring.Metric;
import org.apache.sedona.core.spatialOperator.SpatialPredicate;
import org.apache.spark.api.java.function.FlatMapFunction2;
+import org.apache.spark.util.LongAccumulator;
import org.locationtech.jts.geom.Geometry;
import org.locationtech.jts.index.SpatialIndex;
@@ -39,10 +39,10 @@ public class RightIndexLookupJudgement<T extends Geometry, U extends Geometry>
* @see JudgementBase
*/
public RightIndexLookupJudgement(SpatialPredicate spatialPredicate,
- Metric buildCount,
- Metric streamCount,
- Metric resultCount,
- Metric candidateCount)
+ LongAccumulator buildCount,
+ LongAccumulator streamCount,
+ LongAccumulator resultCount,
+ LongAccumulator candidateCount)
{
super(spatialPredicate, buildCount, streamCount, resultCount, candidateCount);
}
diff --git a/spark/common/src/main/java/org/apache/sedona/core/spatialOperator/JoinQuery.java b/spark/common/src/main/java/org/apache/sedona/core/spatialOperator/JoinQuery.java
index 1088b8096..891e126b0 100644
--- a/spark/common/src/main/java/org/apache/sedona/core/spatialOperator/JoinQuery.java
+++ b/spark/common/src/main/java/org/apache/sedona/core/spatialOperator/JoinQuery.java
@@ -27,7 +27,6 @@ import org.apache.sedona.common.utils.GeomUtils;
import org.apache.sedona.core.enums.IndexType;
import org.apache.sedona.core.enums.JoinBuildSide;
import org.apache.sedona.core.joinJudgement.*;
-import org.apache.sedona.core.monitoring.Metric;
import org.apache.sedona.core.monitoring.Metrics;
import org.apache.sedona.core.spatialPartitioning.SpatialPartitioner;
import org.apache.sedona.core.spatialRDD.CircleRDD;
@@ -38,6 +37,7 @@ import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
+import org.apache.spark.util.LongAccumulator;
import org.locationtech.jts.geom.Geometry;
import scala.Tuple2;
@@ -535,10 +535,10 @@ public class JoinQuery
verifyPartitioningMatch(leftRDD, rightRDD);
SparkContext sparkContext = leftRDD.spatialPartitionedRDD.context();
- Metric buildCount = Metrics.createMetric(sparkContext, "buildCount");
- Metric streamCount = Metrics.createMetric(sparkContext, "streamCount");
- Metric resultCount = Metrics.createMetric(sparkContext, "resultCount");
- Metric candidateCount = Metrics.createMetric(sparkContext, "candidateCount");
+ LongAccumulator buildCount = Metrics.createMetric(sparkContext, "buildCount");
+ LongAccumulator streamCount = Metrics.createMetric(sparkContext, "streamCount");
+ LongAccumulator resultCount = Metrics.createMetric(sparkContext, "resultCount");
+ LongAccumulator candidateCount = Metrics.createMetric(sparkContext, "candidateCount");
final SpatialPartitioner partitioner =
(SpatialPartitioner) rightRDD.spatialPartitionedRDD.partitioner().get();
diff --git a/spark/common/src/main/scala/org/apache/sedona/core/monitoring/Metric.scala b/spark/common/src/main/scala/org/apache/sedona/core/monitoring/Metric.scala
deleted file mode 100644
index 373073a0e..000000000
--- a/spark/common/src/main/scala/org/apache/sedona/core/monitoring/Metric.scala
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.sedona.core.monitoring
-
-import org.apache.spark.TaskContext
-import org.apache.spark.util.AccumulatorV2
-
-import scala.collection.mutable
-
-/**
- * An accumulator to collect custom per-task metrics into a map
- * keyed by partition ID processed by the task.
- */
-case class Metric(initialValue: Map[Int, Long] = Map()) extends AccumulatorV2[Long, Map[Int, Long]] {
- private var _counts: mutable.Map[Int, Long] = mutable.Map[Int, Long]() ++ initialValue
-
- override def isZero: Boolean = _counts.isEmpty
-
- override def copy(): AccumulatorV2[Long, Map[Int, Long]] = new Metric(_counts.toMap)
-
- override def reset(): Unit = _counts.clear()
-
- override def add(v: Long): Unit = add(TaskContext.getPartitionId, v)
-
- private def add(partitionId: Int, value: Long) = {
- _counts(partitionId) = value + _counts.getOrElse(partitionId, 0L)
- }
-
- override def merge(other: AccumulatorV2[Long, Map[Int, Long]]): Unit = {
- other.asInstanceOf[Metric]._counts.foreach {
- case (partitionId, value) => add(partitionId, value)
- }
- }
-
- override def value: Map[Int, Long] = _counts.toMap
-}
diff --git a/spark/common/src/main/scala/org/apache/sedona/core/monitoring/Metrics.scala b/spark/common/src/main/scala/org/apache/sedona/core/monitoring/Metrics.scala
index b4359f2ab..ee295d222 100644
--- a/spark/common/src/main/scala/org/apache/sedona/core/monitoring/Metrics.scala
+++ b/spark/common/src/main/scala/org/apache/sedona/core/monitoring/Metrics.scala
@@ -20,10 +20,11 @@
package org.apache.sedona.core.monitoring
import org.apache.spark.SparkContext
+import org.apache.spark.util.LongAccumulator
object Metrics {
- def createMetric(sc: SparkContext, name: String): Metric = {
- val acc = new Metric()
+ def createMetric(sc: SparkContext, name: String): LongAccumulator = {
+ val acc = new LongAccumulator()
sc.register(acc, "sedona.spatialjoin." + name)
acc
}
diff --git a/spark/common/src/test/scala/org/apache/sedona/core/monitoring/MetricSuite.scala b/spark/common/src/test/scala/org/apache/sedona/core/monitoring/MetricSuite.scala
deleted file mode 100644
index 5e589ec16..000000000
--- a/spark/common/src/test/scala/org/apache/sedona/core/monitoring/MetricSuite.scala
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.sedona.core.monitoring
-
-import org.apache.sedona.sql.TestBaseScala
-import org.apache.spark.TaskContext
-
-class MetricSuite extends TestBaseScala {
-
- it("simple count") {
-
- val data = 1 to 100
- val rdd = sc.parallelize(data, 5)
-
- val metric = Metrics.createMetric(sc, "count")
-
- // Count elements in each partition. For partitions with even IDs count each element twice.
- rdd.map { x =>
- if (TaskContext.getPartitionId() % 2 == 0)
- metric.add(2)
- else
- metric.add(1)
- x
- }.collect
-
- // Verify metric
- val expectedPartitionIds = 0 to 4
- assert(expectedPartitionIds == metric.value.keys.toSeq.sorted)
-
- for ((partitionId, count) <- metric.value) {
- val expectedCount = if (partitionId % 2 == 0) 40 else 20
- assert(count == expectedCount)
- }
- }
-}