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

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

我的代碼結(jié)構(gòu)大致相似

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()

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

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

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

嗨，我可以通過調(diào)用獲得一個列表

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(，)中將執(zhí)行與數(shù)據(jù)庫插入和更新相關(guān)的多個操作，并且還需要執(zhí)行spark數(shù)據(jù)幀和spark sql操作 .

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

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

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。