<Row> 用逗号拆分数据集上的字符串列并获取新数据集<Row>

Question

我正在使用Spark SQL与Spark(2.0)和使用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()方法不返回行。Ship Mode将是动态的，即CSV可能包含一种或两种船舶模式。

谢谢。

Answer 1

您可以使用selectExpr，它是select的一个变体，接受SQL表达式，如下所示:

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

结果是Row.

Answer 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")