讓我先告訴大家我對Spark很新 .

我需要在表中處理大量記錄，當它通過電子郵件分組時大約是100萬 . 我需要根據 individual email 和 update the database based on the logical calculation 的數據集執行多個邏輯計算

我的代碼結構大致相似

Initial Data Load ...

import sparkSession.implicits._ var tableData = sparkSession.read.jdbc(,, connectionProperties).select(“email”) . where()

//Data Frame with Records with grouping on email count greater than one

var recordsGroupedBy = tableData.groupBy(“email”) . count() . withColumnRenamed(“count”，“recordcount”) . filter(“recordcount> 1”) . toDF()

現在是使用processDataAgainstEmail()方法對電子郵件進行分組后的處理

recordsGroupedBy.collect().foreach(x=>processDataAgainstEmail(x.getAs("email"),sparkSession))

在這里，我看到foreach沒有并行執行 . 我需要并行調用processDataAgainstEmail(，)方法 . 但是，如果我嘗試通過做并行化

嗨，我可以通過調用獲得一個列表

val emailList =dataFrameWithGroupedByMultipleRecords.select("email").rdd.map(r => r(0).asInstanceOf[String]).collect().toList

var rdd = sc.parallelize(emailList )

rdd.foreach(x => processDataAgainstEmail(x.getAs("email"),sparkSession))

這不受支持，因為我在使用parallelize時無法傳遞sparkSession .

任何人都可以幫助我，因為在processDataAgainstEmail(，)中將執行與數據庫插入和更新相關的多個操作，并且還需要執行spark數據幀和spark sql操作 .

To summerize I need to invoke parallelly processDataAgainstEmail(,) with sparksession

如果不是所有可能通過spark會話，該方法將無法在數據庫上執行任何操作 . 我不確定什么是替代方式，因為電子郵件的并行性必須適用于我的方案 .

總結

以上是生活随笔為你收集整理的java调用spark_spark中的并行方法调用以及传递方法中使用spark会话的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。