如何使用 Java 在 Spark DataFrame 中应用地图功能？答案

【问题标题】：How to apply map function in Spark DataFrame using Java?如何使用 Java 在 Spark DataFrame 中应用地图功能？
【发布时间】：2019-03-21 11:49:45
【问题描述】：

我正在尝试使用 Java 在 Spark 中的 DataFrame 上使用 map 函数。我正在关注文档中说的

map(scala.Function1 f, scala.reflect.ClassTag 证据$4) 通过将函数应用于此 DataFrame 的所有行，返回一个新的 RDD。

在 map 中使用 Function1 时，我需要实现所有功能。我见过一些与此相关的questions，但提供的解决方案将DataFrame 转换为RDD。如何在DataFrame 中使用map 函数而不将其转换为RDD 还有什么是map 的第二个参数，即scala.reflect.ClassTag<R> evidence$4

我正在使用 Java 7 和 Spark 1.6。

【问题讨论】：

map 函数将返回一个 RDD，如文档所述......无论如何，是什么阻止您获得 Spark 2 或至少 Java 8？
是的。 map函数无论如何都会返回RDD，但是如果不能直接使用它为什么在DataFrame中提供了一个map函数呢？实际上我正处于学习阶段，我们不知道客户端是使用 spark 1.7 还是 spark 2 。所以我们必须同时解决。

标签： java apache-spark apache-spark-sql

【解决方案1】：

我知道您的问题是关于 Java 7 和 Spark 1.6，但在 Spark 2（显然还有 Java 8）中，您可以将 map 函数作为类的一部分，因此您无需操作 Java lambda。

调用看起来像：

Dataset<String> dfMap = df.map(
    new CountyFipsExtractorUsingMap(),
    Encoders.STRING());
dfMap.show(5);

这个类看起来像：

  /**
   * Returns a substring of the values in the id2 column.
   * 
   * @author jgp
   */
  private final class CountyFipsExtractorUsingMap
      implements MapFunction<Row, String> {
    private static final long serialVersionUID = 26547L;

    @Override
    public String call(Row r) throws Exception {
      String s = r.getAs("id2").toString().substring(2);
      return s;
    }
  }

您可以在this example on GitHub找到更多详细信息。

【讨论】：

【解决方案2】：

我认为map 不是在DataFrame 上使用的正确方法。也许你应该看看例子in the API

他们展示了如何在DataFrames 上操作

【讨论】：

我在 DataFrame 中使用了groupBy,agg,orderBy 操作。那么哪些是我们需要在 DataFrame 上避免的操作，我没有看到任何关于它的文档。无论如何感谢您的重播。

【解决方案3】：

您可以直接使用数据集，无需将读取的数据转换为RDD，不必要的资源消耗。

dataset.map(mapfuncton{...}, 编码器);这应该可以满足您的需求。

【讨论】：

【解决方案4】：

因为您没有给出任何具体问题，所以在DataFrame 中有一些常见的替代map，例如select、selectExpr、withColumn。如果 spark sql 内置函数不能满足您的任务，您可以使用 UTF。

【讨论】：