用逗号分割 Dataset<Row> 上的字符串列并获取新的 Dataset<Row>答案

【问题标题】：Split String Column on the Dataset<Row> with comma and get new Dataset<Row>用逗号分割 Dataset<Row> 上的字符串列并获取新的 Dataset<Row>
【发布时间】：2016-09-02 02:03:36
【问题描述】：

我正在使用 Spark(2.0) 开发 Spark SQL，并使用 Java API 读取 CSV。

在 CSV 文件中有一个双引号，逗号分隔的列。例如："Express Air,Delivery Truck"

读取CSV并返回Dataset的代码：

Dataset<Row> df = spark.read()
                .format("com.databricks.spark.csv")
                .option("inferSchema", "true")
                .option("header", "true")
                .load(filename)

结果：

+-----+--------------+--------------------------+
|Year |       State  |                Ship Mode |...
+-----+--------------+--------------------------+
|2012 |New York      |Express Air,Delivery Truck|...
|2013 |Nevada        |Delivery Truck            |...
|2013 |North Carolina|Regular Air,Delivery Truck|...
+-----+--------------+--------------------------+

但是，我想将Shop Mode 拆分为Mode1 和Mode2 列并作为数据集返回。

+-----+--------------+--------------+---------------+
|Year |       State  |     Mode1    |         Mode2 |...
+-----+--------------+--------------+---------------+
|2012 |New York      |Express Air   |Delivery Truck |...
|2013 |Nevada        |Delivery Truck|null           |...
|2013 |North Carolina|Regular Air   |Delivery Truck |...
+-----+--------------+--------------+---------------+

我有什么方法可以使用 Java Spark 做到这一点？

我尝试使用 MapFunction，但 call() 方法没有返回 Row。 Ship Mode 将是动态的，即 CSV 可能包含一种或两种船舶模式。

谢谢。

【问题讨论】：

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

【解决方案1】：

您可以使用 selectExpr，这是一种接受 SQL 表达式的 select 变体，如下所示：

df.selectExpr("Year","State","split(Ship Mode, ',')[0] as Mode1","split(Ship Mode, ',')[1] as Mode2");

结果是行的数据集。

【讨论】：

当逗号分隔值的数量未知时，您能否提出另一种解决方案？它可能从 col1、col2、.... 到 col40 不等。最高为 40。

【解决方案2】：

我们可以：

定义一个用户定义函数 (UDF) 以只执行一次拆分操作
使用 select 表达式将拆分的列映射到两个新列

例如：

import org.apache.spark.sql.functions._
import org.apache.spark.sql.{Column, Row}

val splitter = udf((str: String) => {
  val splitted = str.split(",").lift
  Array(splitted(0), splitted(1))
})

val dfShipMode = df.select($"year",$"state", splitter($"shipMode") as "modes")
                   .select($"year", $"state", $"modes"(0) as "mode1", $"modes"(1) as "mode2")

【讨论】：