[SPARK-34309][BUILD][CORE][SQL][K8S]Use Caffeine instead of Guava Cache #31517
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,14 @@ | ||
| ================================================================================================ | ||
| Loading Cache | ||
| ================================================================================================ | ||
|
|
||
| OpenJDK 64-Bit Server VM 11.0.8+10-LTS on Mac OS X 10.15.7 | ||
| Intel(R) Core(TM) i5-7360U CPU @ 2.30GHz | ||
| Loading Cache: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative | ||
| -------------------------------------------------------------------------------------------------------------------------- | ||
| Guava Cache 5 6 0 15.2 65.6 1.0X | ||
| Caffeine 2 2 1 46.9 21.3 3.1X | ||
| CaffeinatedGuava with Guava CacheLoader 2 2 0 46.0 21.8 3.0X | ||
| CaffeinatedGuava with Caffeine CacheLoader 2 2 0 46.1 21.7 3.0X | ||
|
|
||
|
|
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,14 @@ | ||
| ================================================================================================ | ||
| Loading Cache | ||
| ================================================================================================ | ||
|
|
||
| OpenJDK 64-Bit Server VM 1.8.0_232-b18 on Mac OS X 10.15.7 | ||
| Intel(R) Core(TM) i5-7360U CPU @ 2.30GHz | ||
| Loading Cache: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative | ||
| -------------------------------------------------------------------------------------------------------------------------- | ||
| Guava Cache 5 6 2 15.2 65.7 1.0X | ||
| Caffeine 2 2 1 41.2 24.3 2.7X | ||
| CaffeinatedGuava with Guava CacheLoader 2 2 0 44.0 22.7 2.9X | ||
| CaffeinatedGuava with Caffeine CacheLoader 2 2 0 44.2 22.6 2.9X | ||
|
|
||
|
|
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -25,7 +25,9 @@ import javax.servlet.http.{HttpServletRequest, HttpServletResponse} | |
| import scala.collection.JavaConverters._ | ||
|
|
||
| import com.codahale.metrics.{Counter, MetricRegistry, Timer} | ||
| import com.github.benmanes.caffeine.cache.{Caffeine, RemovalCause, RemovalListener} | ||
| import com.github.benmanes.caffeine.guava.CaffeinatedGuava | ||
| import com.google.common.cache.{CacheLoader, LoadingCache} | ||
| import com.google.common.util.concurrent.UncheckedExecutionException | ||
| import org.eclipse.jetty.servlet.FilterHolder | ||
|
|
||
|
|
@@ -62,21 +64,27 @@ private[history] class ApplicationCache( | |
|
|
||
| /** | ||
| * Removal event notifies the provider to detach the UI. | ||
| * @param key removal key | ||
| * @param value removal value | ||
| */ | ||
| override def onRemoval(key: CacheKey, value: CacheEntry, | ||
| cause: RemovalCause): Unit = { | ||
|
Contributor
There was a problem hiding this comment. Maybe mention RemovalCause in the scaladoc?
Member
There was a problem hiding this comment. nit: move this to the previous line?
Contributor
Author
There was a problem hiding this comment. ok ~ It doesn't look exceeds 100 characters ~ |
||
| metrics.evictionCount.inc() | ||
| logDebug(s"Evicting entry $key") | ||
| operations.detachSparkUI(key.appId, key.attemptId, value.loadedUI.ui) | ||
| } | ||
| } | ||
|
|
||
| private val appCache: LoadingCache[CacheKey, CacheEntry] = { | ||
| val builder = Caffeine.newBuilder() | ||
| .maximumSize(retainedApplications) | ||
| .removalListener(removalListener) | ||
| // SPARK-34309: Use custom Executor to compatible with | ||
| // the data eviction behavior of Guava cache | ||
| .executor((command: Runnable) => command.run()) | ||
|
Contributor
There was a problem hiding this comment. Is this going to run in the same thread? Is that what the old behaviour would have been?
Contributor
Author
There was a problem hiding this comment. Yes, it's a compromise for compatibility with old behaviour at present. In the future, I will optimize this behaviour through other pr. |
||
| // Wrapping as CaffeinatedGuava to be compatible with | ||
| // the exception behavior of Guava cache | ||
| CaffeinatedGuava.build(builder, appLoader) | ||
|
Contributor
There was a problem hiding this comment. Is there a reason to preserve behavior of guava cache here ? There was a problem hiding this comment. My uninformed review is that you can drop the guava adapters. The catch in
Contributor
There was a problem hiding this comment. Thanks @ben-manes !
Contributor
Author
There was a problem hiding this comment. I think we can remove the
For example, there are some differences in the processing of error, when the
For example, |
||
| } | ||
|
|
||
| /** | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| /* | ||
| * 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.spark | ||
|
|
||
| import scala.util.Random | ||
|
|
||
| import com.github.benmanes.caffeine.cache.{CacheLoader => CaffeineCacheLoader, Caffeine} | ||
| import com.github.benmanes.caffeine.guava.CaffeinatedGuava | ||
| import com.google.common.cache.{CacheBuilder, CacheLoader, LoadingCache} | ||
|
|
||
| import org.apache.spark.benchmark.{Benchmark, BenchmarkBase} | ||
|
|
||
| /** | ||
| * Benchmark for Guava Cache vs Caffeine. | ||
| * To run this benchmark: | ||
| * {{{ | ||
| * 1. without sbt: | ||
| * bin/spark-submit --class <this class> --jars <spark core test jar> | ||
| * 2. build/sbt "core/test:runMain <this class>" | ||
| * 3. generate result: | ||
| * SPARK_GENERATE_BENCHMARK_FILES=1 build/sbt "core/test:runMain <this class>" | ||
| * Results will be written to "benchmarks/KryoBenchmark-results.txt". | ||
|
|
||
| * }}} | ||
| */ | ||
| object LocalCacheBenchmark extends BenchmarkBase { | ||
|
|
||
| override def runBenchmarkSuite(mainArgs: Array[String]): Unit = { | ||
| runBenchmark("Loading Cache") { | ||
| val size = 10000 | ||
| val parallelism = 8 | ||
| val guavaCacheConcurrencyLevel = 8 | ||
| val dataset = (1 to parallelism) | ||
| .map(_ => Random.shuffle(List.range(0, size))) | ||
| .map(list => list.map(i => TestData(i))) | ||
|
Comment on lines
+50
to
+52
There was a problem hiding this comment. This distribution is uniformly distributed with only single key overlaps. This means that there are not hot and cold entries, e.g. random eviction has an optimal hit rate. In reality, some entries will be used much more often and follows a power law curve. That is fairly generous distribution for a cache like guava, which uses coarse locking of multiple hash tables. That way the access distribution matches the hash distribution, so ideally spread across all of the locks. In reality, while the hash distribution will be uniform the access distribution is not so a lock holding hot entries will be used much more frequently. In Caffeine's benchmarks, it uses a scrambled Zipfian distribution (YCSB's generator). That would show an even larger speedup. More just an fyi that your benchmarks are conservative and you may see a larger gain. Of course, if the caches are not a bottleneck you might not see any benefit except if the eviction policy improves the hit rates in your workloads.
Contributor
Author
There was a problem hiding this comment. Thank you for your advice. I think we should avoid introducing more dependencies, so I'll try to implement this data generator in spark code. There was a problem hiding this comment. I think your code is fine as is. Maybe just document the simplification? I mostly wanted to let you know since writing a good benchmark is hard, not that you should change it. Your code served its purpose, and you might not get much more out of improving it. |
||
| val guavaCacheLoader = new CacheLoader[TestData, TestData]() { | ||
| override def load(id: TestData): TestData = { | ||
| id | ||
| } | ||
| } | ||
| val caffeineCacheLoader = new CaffeineCacheLoader[TestData, TestData]() { | ||
| override def load(id: TestData): TestData = { | ||
| id | ||
| } | ||
| } | ||
|
|
||
| val benchmark = new Benchmark("Loading Cache", size * parallelism, 3, output = output) | ||
| benchmark.addCase("Guava Cache") { _ => | ||
| val cache = CacheBuilder.newBuilder() | ||
| .concurrencyLevel(guavaCacheConcurrencyLevel).build[TestData, TestData](guavaCacheLoader) | ||
| dataset.par.foreach(dataList => dataList.foreach(key => cache.get(key))) | ||
| cache.cleanUp() | ||
| } | ||
|
|
||
| benchmark.addCase("Caffeine") { _ => | ||
| val cache = Caffeine.newBuilder().build[TestData, TestData](caffeineCacheLoader) | ||
| dataset.par.foreach(dataList => dataList.foreach(key => cache.get(key))) | ||
| cache.cleanUp() | ||
| } | ||
|
|
||
| benchmark.addCase("CaffeinatedGuava with Guava CacheLoader") { _ => | ||
| val cache: LoadingCache[TestData, TestData] = | ||
| CaffeinatedGuava.build(Caffeine.newBuilder(), guavaCacheLoader) | ||
| dataset.par.foreach(dataList => dataList.foreach(key => cache.get(key))) | ||
| cache.cleanUp() | ||
| } | ||
|
|
||
| benchmark.addCase("CaffeinatedGuava with Caffeine CacheLoader") { _ => | ||
| val cache: LoadingCache[TestData, TestData] = | ||
| CaffeinatedGuava.build(Caffeine.newBuilder(), caffeineCacheLoader) | ||
| dataset.par.foreach(dataList => dataList.foreach(key => cache.get(key))) | ||
| cache.cleanUp() | ||
| } | ||
|
|
||
| benchmark.run() | ||
| } | ||
| } | ||
|
|
||
| case class TestData(content: Int) | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Could we avoid this since it depends on Guava and guava conflicts can be a pain to resolve.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is a good idea, but there are some interface differences between Caffeine and Guava Cache, and spark just uses these interfaces. For example,
SessionCatalogusesV get(K key, Callable<? extends V> valueLoader)method only define in Guava Cache APIUh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Could you use Caffeine's
V get(K key, Function<? extends K, ? extends V> mappingFunction)instead? That should be equivalent but be friendly for method references.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, I'm trying to get rid of the dependence on
com.github.ben-manes.caffeine:guavain this wayThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@holdenk @ben-manes All the dependencies related to
com.github.ben-manes.caffeine: guavahave been removed.If all test passed , I will try to upgrade caffeine from 2.9.0 to 2.9.1 in this pr