將引數傳遞給Scala中的執行緒

我正在學習 Scala，作為練習，我正在將一些 python (PySpark) 代碼轉換為 Scala (spark/Scala) 代碼。一切都很順利，直到我開始處理 scala 執行緒。那么，您現在如何將以下代碼重新撰寫到 scala 中？

先感謝您！

def load_tables(table_name, spark):
    source_path = f"s3://data/tables/{table_name}"
    table = spark.read.format("csv").load(source_path)
    table.createOrReplaceTempView(table_name)

def read_initial_tables(spark):
    threads  = []
    tables = ["table1", "table2", "table3"]
    for table in tables:
        t = threading.Thread(target=load_tables, args=(table, spark))
        threads.append(t)
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()

uj5u.com熱心網友回復：

...將引數傳遞到執行緒中...

Scala 使用 Java 標準庫，在 Java 中啟動執行緒與在 Python 中啟動執行緒有點不同。主要區別在于，在 Python 中，您可以target為執行緒的頂層選擇任何（即任何函式或可呼叫物件），并且您可以傳入任何args您喜歡的內容。但是當你啟動一個 Java 執行緒時，頂層函式必須是一個名為run()that 的無引數方法，它屬于一個implements java.lang.Runnable.

您的 Python 執行緒的頂級函式是load_tables(table, spark). 因此，您的 Scala 程式中需要一個執行緒，其頂級函式run()是呼叫 load_tables(table, spark).

我實際上并不了解 Scala，但也許此網頁上的示例將引導您朝著正確的方向前進：https ://alvinalexander.com/scala/how-to-create-java-thread-runnable-in-scala/

基本上，我認為您所要做的就是效仿他的榜樣，并將您的load_tables(table, spark)電話放在他的榜樣所說的地方，“您的自定義行為在這里。”

uj5u.com熱心網友回復：

所羅門是對的。我無法更好地描述它。利用 Scala 通過 Java 提供的語法糖，您的 Python 代碼不再使用 Scala：

  def load_tables(table_name: String, spark: SparkSession): Runnable = () => {
    val source_path = s"s3://data/tables/$table_name"
    val table = spark.read.format("csv").load(source_path)
    table.createOrReplaceTempView(table_name)
  }

  def read_initial_tables(spark: SparkSession): Unit = {
    val tables = List("table1", "table2", "table3")
    val threads = for {
      table <- tables
    } yield new Thread(load_tables(table, spark))
    for (thread <- threads)
      thread.start()
    for (thread <- threads)
      thread.join()
  }

你可能會問run()，所羅門所說的方法在哪里。實際上，符號()后的空括號表示傳遞給方法的無引數引數串列，而方法體是符號后大括號之間的代碼塊。所以呼叫實際上會回傳一個新實體。=load_tablesrunrun=>load_tablesRunnable

這被稱為單一抽象方法，它只是一種語法糖，給人的印象load_tables看起來像在 Python 中一樣可呼叫，但實際上并非如此。只有它的回傳型別是，因為它回傳一個Runnable物件。這個簡短的版本是可以實作的，因為Runnable它是一個功能介面。

我不是 Spark 方面的專家，所以我不確定這是否是使用 Spark 在 Scala 中編碼的慣用方式，但這是一個很好的起點。

uj5u.com熱心網友回復：

也許不是你真正想要的，但它可能很有趣。Scala 有一些非常方便的東西，可以通過以下方法并行化集合.par：

val parallelizedList = List(1, 2, 3, 4).par
parallelizedList.foreach(i => println(i)) // this is executed in parallel, not sequentially
// output:
// 2
// 4
// 1
// 3

因此，您可以將此語法與 spark 一起使用來并行讀取多個表：

def loadTable(tableName: String, spark: SparkSession): Unit = {
    val sourcePath = f"s3://data/tables/$tableName"
    val table = spark.read.format("csv").load(sourcePath)
    table.createOrReplaceTempView(tableName)
}

val tableNames = List("table1", "table2", "table3")
tableNames.par.foreach(name => loadTable(name, spark))

編輯

• 如果您使用 Scala 2.12，則可以使用并行集合。它們已在 2.13 中移至自己的模塊：scala/scala-parallel-collection
  • libraryDependencies = "org.scala-lang.modules" %% "scala-parallel-collections" % "1.0.0"
  • import scala.collection.parallel.CollectionConverters._
• 如果您對并行集合執行的操作修改了相同的資料，請務必小心。這可能導致不確定的行為（請參閱下面的@Alin Gabriel Arhip 的評論）。
• 顯然，并沒有真正鼓勵將并行集合與 Spark 一起使用（另請參閱下面的@Alin Gabriel Arhip 的評論），但到目前為止我從未遇到過任何問題（盡管我通常只將它們用于我知道的非常簡單的處理不會使用所有可用資源）

標籤：多线程 斯卡拉 阿帕奇火花

上一篇：從fetchDiagnosisForUniqueCodes方法回傳Future[List[DiagnosisCode]]

下一篇：在Scala3中為不透明型別派生型別類實體