SparkSQL 并在 Java 中的 DataFrame 上爆炸

Posted 2023-04-15

【中文标题】SparkSQL 并在 Java 中的 DataFrame 上爆炸

【英文标题】：SparkSQL and explode on DataFrame in Java

【发布时间】：2015-08-06 15:03:19

【问题描述】：

有没有一种简单的方法可以在 SparkSQL DataFrame 的数组列上使用 explode？在 Scala 中相对简单，但是在 Java 中这个功能似乎不可用（如 javadoc 中所述）。

一个选项是在查询中使用SQLContext.sql(...) 和explode 函数，但我正在寻找更好、更简洁的方法。 DataFrames 是从 parquet 文件中加载的。

【参考方案1】：

我以这种方式解决了这个问题：假设您有一个数组列，其中包含名为“职位”的职位描述，每个人都有“全名”。

然后你从初始模式中得到：

root
|-- fullName: string (nullable = true)
|-- positions: array (nullable = true)
    |    |-- element: struct (containsNull = true)
    |    |    |-- companyName: string (nullable = true)
    |    |    |-- title: string (nullable = true)
...

到架构：

root
 |-- personName: string (nullable = true)
 |-- companyName: string (nullable = true)
 |-- positionTitle: string (nullable = true)

通过做：

    DataFrame personPositions = persons.select(persons.col("fullName").as("personName"),
          org.apache.spark.sql.functions.explode(persons.col("positions")).as("pos"));

    DataFrame test = personPositions.select(personPositions.col("personName"),
    personPositions.col("pos").getField("companyName").as("companyName"), personPositions.col("pos").getField("title").as("positionTitle"));

【讨论】：

在 java 8 和 Spark 2.4.7 中，explode(..) 方法返回一个 Column 而不是一个 DataFrame。你能指定你的版本，explode 给你一个数据框吗？

【参考方案2】：

似乎可以使用org.apache.spark.sql.functions.explode(Column col) 和DataFrame.withColumn(String colName, Column col) 的组合来用它的分解版本替换列。