调用 scala 方法，将数据帧的每一行作为输入传递答案

【问题标题】：calling a scala method passing each row of a dataframe as input调用 scala 方法，将数据帧的每一行作为输入传递
【发布时间】：2019-01-04 12:09:53
【问题描述】：

我有一个包含两列的数据框，已创建导入 .txt 文件。

示例文件内容::

Sankar Biswas, Played{"94"}
Puja "Kumari" Jha, Didnot
Man Women, null
null,Gay Gentleman
null,null

创建了一个导入上述文件的数据框::

val a = sc.textFile("file:////Users/sankar.biswas/Desktop/hello.txt")

case class Table(contentName: String, VersionDetails: String)

val b = a.map(_.split(",")).map(p => Table(p(0).trim,p(1).trim)).toDF

现在我定义了一个函数，可以这样说::

  def getFormattedName(contentName : String, VersionDetails:String): Option[String] = {
    Option(contentName+titleVersionDesc)
  }

现在我需要做的是获取数据帧的每一行并调用 getFormattedName 方法，传递数据帧每一行的 2 个参数。

我尝试过这样的方法和许多其他方法，但没有成功::

val a = b.map((m,n) => getFormattedContentName(m,n))

期待您对我的任何建议。提前致谢。

【问题讨论】：

如果你打算执行像map & filter 这样的高阶函数，我建议你改用Dataset。另外，我会使用 DataFrameReader API 来读取您的 csv，正如康斯坦丁已经建议的那样 - 顺便说一句，您可以从 case 类 派生 schema，并且您可以将DataFrame 转换为Dataset[T]，其中T 是一个案例类，因为您可以像元组一样对您的案例类进行模式匹配。如果你愿意，你也可以使用元组。

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

【解决方案1】：

我认为你有一个结构化的模式，它可以用一个数据框来表示。 Dataframe 支持读取 csv 输入。

import org.apache.spark.sql.types._
val customSchema = StructType(Array(StructField("contentName", StringType, true),StructField("titleVersionDesc", StringType, true)))

val df = spark.read.schema(customSchema).csv("input.csv")

要在数据集上调用自定义方法，您可以创建一个 UDF（用户定义函数）。

def getFormattedName(contentName : String, titleVersionDesc:String): Option[String] = {
    Option(contentName+titleVersionDesc)
  }

val get_formatted_name = udf(getFormattedName _)

df.select(get_formatted_name($"contentName", $"titleVersionDesc"))

【讨论】：

【解决方案2】：

试试

val a = b.map(row => getFormattedContentName(row(0),row(1)))

请记住，数据框的行是它们自己的类型，而不是元组或其他东西，您需要使用正确的方法来引用它们的元素。

【讨论】：

apply 中的 apply 方法返回 Any，您必须将值转换为之后才能将它们传递给该方法。我会改用row.getAs[String](0)。