scala – Spark：如何使用mapPartition并为每个分区创建/关闭连

发布时间：2020-12-16 09:25:02 所属栏目：安全来源：网络整理

导读：所以,我想对我的spark DataFrame进行某些操作,将它们写入DB并在最后创建另一个DataFrame.它看起来像这样： import sqlContext.implicits._val newDF = myDF.mapPartitions( iterator = { val conn = new DbConnection iterator.map( row = { addRowToBatch(r

所以,我想对我的spark DataFrame进行某些操作,将它们写入DB并在最后创建另一个DataFrame.它看起来像这样：

import sqlContext.implicits._

val newDF = myDF.mapPartitions(
  iterator => {
    val conn = new DbConnection
    iterator.map(
       row => {
         addRowToBatch(row)
         convertRowToObject(row)
     })
    conn.writeTheBatchToDB()
    conn.close()
  })
  .toDF()

这给了我一个错误,因为mapPartitions期望Iterator [NotInferedR]的返回类型,但这里是Unit.我知道这可以用forEachPartition,但我也想做映射.分开进行将是一个开销(额外的火花工作).该怎么办？

谢谢！

解决方法

匿名函数实现中的最后一个表达式必须是返回值：

import sqlContext.implicits._

val newDF = myDF.mapPartitions(
  iterator => {
    val conn = new DbConnection
    // using toList to force eager computation - make it happen now when connection is open
    val result = iterator.map(/* the same... */).toList
    conn.writeTheBatchToDB()
    conn.close()
    result.iterator
  }
).toDF()

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!