尝试在 Spark DataFrame 上使用地图

Question

我最近开始尝试使用 Spark 和 Java。我最初使用RDD 浏览了著名的WordCountexample，一切都按预期进行。现在我正在尝试实现我自己的示例，但使用的是 DataFrames 而不是 RDD。

所以我正在从文件中读取数据集

DataFrame df = sqlContext.read()
        .format("com.databricks.spark.csv")
        .option("inferSchema", "true")
        .option("delimiter", ";")
        .option("header", "true")
        .load(inputFilePath);

然后我尝试选择一个特定的列并像这样对每一行应用一个简单的转换

df = df.select("start")
        .map(text -> text + "asd");

但是编译发现第二行有问题，我不完全理解（开始列推断为string类型）。

在接口 scala.Function1 中找到多个非覆盖抽象方法

为什么我的 lambda 函数被视为 Scala 函数，错误消息的实际含义是什么？

Answer 1

如果您在数据帧上使用select函数，您会得到一个数据帧。然后在Rowdatatype 上应用一个函数，而不是行的值。之后您应该首先获取该值，因此您应该执行以下操作：

df.select("start").map(el->el.getString(0)+"asd")

但是你会得到一个 RDD 作为返回值而不是一个 DF

Answer 2

我使用 concat 来实现这个

df.withColumn( concat(col('start'), lit('asd'))

由于您两次映射相同的文本，我不确定您是否还希望替换字符串的第一部分？但如果你是，我会这样做：

df.withColumn('start', concat(
                      when(col('start') == 'text', lit('new'))
                      .otherwise(col('start))
                     , lit('asd')
                     )

此解决方案在使用大数据时可扩展，因为它连接两列而不是迭代值。