星火：如何使用mapPartition并创建每个分区/关闭连接 [英] Spark : How to use mapPartition and create/close connection per partition

问题描述

所以，我想对我的火花数据框中指定的工作，他们写信给DB和创造末另一个数据框。它看起来是这样的：

So, I want to do certain operations on my spark DataFrame, write them to DB and create another DataFrame at the end. It looks like this :

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] ，但在这里它是单位 。我知道这是可能的forEachPartition，但我喜欢做的映射也。这样做分开将是一个开销（额外的火花的工作）。怎么办？

This gives me an error as mapPartitions expects return type of Iterator[NotInferedR], but here it is Unit. I know this is possible with forEachPartition, but I'd like to do the mapping also. Doing it separate would be an overhead (extra spark job). What to do?

谢谢！

推荐答案

在匿名函数执行的最后一个前pression必须返回值：

The last expression in the anonymous function implementation must be the return value:

import sqlContext.implicits._

val newDF = myDF.mapPartitions(
  iterator => {
    val conn = new DbConnection
    val result = iterator.map(/* the same... */)
    conn.writeTheBatchToDB()
    conn.close()
    result
  }
).toDF()

这篇关于星火：如何使用mapPartition并创建每个分区/关闭连接的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！