Shared singleton for using with Spark
まだ誰も着手していません。
評価
- 難易度
- 5/5
- 見積もり時間
- 1週間以上
- 初心者へのやさしさ
- 25/100
- issue の種類
- 機能追加
- 明瞭さ
- 説明が足りない
- 活発さ
- 停滞
- 技術スタック
- scala
調査の方向性
issue の Spark の例から始め、特に Module、Runner、Database を対象に、シリアライズによってタスク間でデータベースインスタンスがどのように複製されるかを追跡します。完了には、Spark タスク間でデータベースクライアントを共有するための、決定済みでサポートされたパターンまたは Feature があり、その動作と使用方法が文書化されていることが必要です。
索引モデルが issue の本文から書いたものです。
説明
When using Spark with external resources like a database, a somehow common pattern is to make the database client shared between tasks so the connection pool is shared. Otherwise, with a large number of tasks/threads, the database connections are exhausted and will lead to issues when scaling.
This rises some complications when using such an object, as it must implement some kind of singleton shared between threads that receive serialized objects.
Any idea on how to do this with MacWire? Any pattern that can be used?
A simple example:
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.scalatest.funspec.AnyFunSpec
import org.scalatest.matchers.must.Matchers.{be, convertToAnyMustWrapper}
class ModuleWithSparkSpec extends AnyFunSpec {
it("runs module with spark") {
val parallelism = 4
val module = new Module {
override lazy val connectionString: String = ""
override lazy val sparkConf: SparkConf = new SparkConf().setAppName("Test").setMaster(s"local[$parallelism]")
}
module.run(parallelism * 3) must be(parallelism * 3) // prints 4 thread ids and 4 different hash codes for 3 times
}
}
class Runner(val sparkConf: SparkConf, val database: Database) extends Serializable {
def run(count: Int): Long = {
val database = this.database
val sparkConf = this.sparkConf
SparkSession
.builder()
.config(sparkConf)
.getOrCreate()
.sparkContext
.parallelize(0 until count)
.map { n => database.insert(n) }
.count()
}
}
trait Module extends Serializable {
def run(count: Int): Long = runner.run(count)
import com.softwaremill.macwire._
protected lazy val connectionString: String = ""
protected lazy val sparkConf: SparkConf = new SparkConf().setAppName("").setMaster("")
protected lazy val database: Database = wire[Database] // this will be serialized and duplicated 4 times
protected lazy val runner: Runner = wire[Runner]
}
class Database(connectionString: String) extends Serializable with AutoCloseable {
def insert(n: Int): Unit = {
println(s"Insert $n on thread id = ${Thread.currentThread().getId}, instance hash code = ${hashCode()}")
}
override def close(): Unit = {}
}
So the idea would be to have something instead of wire, or beside, that would make it use a single instance. I was thinking to implement a shared singleton Scope that picks the instance from a concurrent collection, would this be the best way to do it?
protected lazy val database: Database = sharedSingleton(wire[Database])
- 主要言語
- Scala
- スター
- 1.3k
- フォーク
- 77
- 平均マージ
- 9分
- マージ済み PR(30日)
- 4
コントリビューションガイド
このリポジトリのコントリビューションガイドは索引されていません
はじめの一歩
- issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
- 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
- リポジトリをフォークし、ブランチを切って変更します。
- issue 番号を参照したプルリクエストを送ります。
softwaremill/macwire のほかの issue
-
難易度 4/5 3〜5日 初心者へのやさしさ 35/100
softwaremill/macwire#412 · コメント 3 件 ·
-
難易度 4/5 3〜5日 初心者へのやさしさ 25/100
softwaremill/macwire#391 · コメント 1 件 ·
-
難易度 5/5 1週間以上 初心者へのやさしさ 20/100
softwaremill/macwire#301 · コメント 2 件 ·
-
難易度 5/5 1週間以上 初心者へのやさしさ 20/100
softwaremill/macwire#281 · コメント 5 件 ·
-
難易度 4/5 3〜5日 初心者へのやさしさ 35/100
softwaremill/macwire#278 · コメント 3 件 ·
softwaremill/macwire の issue をすべて見る
似ている issue
-
難易度 1/5 1時間未満 初心者へのやさしさ 70/100
-
難易度 2/5 1〜3時間 初心者へのやさしさ 62/100
ergoplatform/ergodocs#614 ·
-
area:ci enhancement requires-triage
難易度 2/5 1〜3時間 初心者へのやさしさ 84/100
apache/datafusion-comet#6078 ·
-
enhancement
難易度 2/5 1〜3時間 初心者へのやさしさ 68/100
-
bug delta-datalayout flink
難易度 2/5 1〜3時間 初心者へのやさしさ 86/100