将 Dataframe 内容传递到 sql 存储过程答案

【问题标题】：passing Dataframe contents into sql stored procedure将 Dataframe 内容传递到 sql 存储过程
【发布时间】：2017-08-08 14:45:40
【问题描述】：

我正在尝试将数据框中的内容传递到我的 sql 存储过程中。我使用 map 函数遍历数据帧内容并将它们发送到数据库中。尝试执行此操作时出错。

我收到一个名为 No Encoder found for Any 的错误 - 字段（类：“java.lang.Object”，名称：“_1”） - 根类：“scala.Tuple2”

谁能帮我纠正这个问题。

下面是我的代码

 val savedDataFrame = dataFrame.map(m => sendDataFrameToDB(m.get(0), m.get(1), m.get(2), m.get(3)))
   savedDataFrame.collect()

 def sendDataFrameToDB(firstName : String, lastName : String, address : String, age : Long) = {
var jdbcConnection: java.sql.Connection = null

try {
  val jdbcTemplate = new JDBCTemplate()
  jdbcTemplate.getConfiguration()
  jdbcConnection = jdbcTemplate.getConnection

  if (jdbcConnection != null) {
    val statement = "{call insert_user_details (?,?,?,?)}"

    val callableStatement = jdbcConnection.prepareCall(statement)

    callableStatement.setString(1, firstName)
    callableStatement.setString(2, lastName)
    callableStatement.setString(3, address)

    callableStatement.setLong(4, age)

    callableStatement.executeUpdate
  }
} catch {
  case e: SQLException => logger.error(e.getMessage)
} 
}

【问题讨论】：

尝试 foreach 而不是 map
否不能使用foreach。 map 提供比 foreach 更好的性能
然后为地图提供编码器doc
@AnkushSingh 你能提一下应该如何写地图。真的很有帮助
你用 udf 函数试过了吗？

标签： scala apache-spark apache-spark-sql spark-dataframe

【解决方案1】：

passing Dataframe contents into sql stored procedure

dataFrame.map(m => sendDataFrameRDBMS(f.getAs("firstname").toString, f.getAs("lastname").toString, f.getAs("address").toString, f.getAs("年龄").toString.toLong))

m.get(0) 属于 Any 类型，根据您的示例，它不能直接传递给 String 类型的 firstName 。 Datframe 与 RDD 不同。 “DataFrame 是组织成命名列的 Dataset。它在概念上相当于关系数据库中的表或 R/Python 中的数据框，但在底层进行了更丰富的优化”link

当你制作数据框时，会制作诸如

之类的列

val dataFrame = dataSet.toDF("firstname", "lastName", "address", "age")

然后您可以访问数据框中的元素，如下所示并传递给您的任何方法

dataFrame.map(m => sendDataFrameRDBMS(f.getAs("firstname").toString, f.getAs("lastname").toString, f.getAs("address").toString, f.getAs("age").toString.toLong))

【讨论】：