在不使用UDF的情况下添加多列中的多列

Question

我想使用addressParser功能从给定的表结构中解析地址列，以获取数字，街道，城市和国家。

样本输入：

地址	地址
add001	“ 384，East Avenue Street，纽约，美国
add002	” 123，Maccolm Street，MacColm Street，Copenhagen，Denmark，Denmark，“

示例代码供参考：

object ParseAddress extends App {

  val spark = SparkSession.builder().master("local[*]").appName("ParseAddress ").getOrCreate()

  import spark.implicits._

  case class AddressRawData(addressId: String, address: String)

  case class AddressData(
                          addressId: String,
                          address: String,
                          number: Option[Int],
                          road: Option[String],
                          city: Option[String],
                          country: Option[String]
                        )

 def addressParser(unparsedAddress: Seq[AddressData]): Seq[AddressData] = {
    unparsedAddress.map(address => {
      val split = address.address.split(", ")

      address.copy(
        number = Some(split(0).toInt),
        road = Some(split(1)),
        city = Some(split(2)),
        country = Some(split(3))
      )
    }
    )
  }

  val addressDS: Dataset[AddressRawData] = addressDF.as[AddressRawData]

}

预期输出：

地址ID	地址	号	公路	城市	，
ADD001	“ 384，East Avenue Street，纽约，美国	384	East Avenue Street	New York New York	USA
add002	” 123“ 123”，Maccolm Street，Copenhagen，Denmark，	Denmark	Street	Copenhagen	Denmark

Maccolm 123我将地址转换为输入以解析列数据。非常感谢解决此问题的某种形式的帮助。

Answer 1

我认为重要的是，最好设计自己的功能，以获取单个输入并返回单个输出（在您的情况下），如果您有一个集合或行数据集，则可以将每一行映射到此功能，这就是为什么制作所有这些多态性功能（地图，flatmap，fold等）的原因，对吗？因此，您可以实现一种方法，该方法接收addressRawdata并返回addressData：

def singleAddressParser(unparsedAddress: AddressRawData): AddressData = {
  val split = unparsedAddress.address.split(", ")
  AddressData(
    addressId = unparsedAddress.addressId,
    address = unparsedAddress.address,
    number = Some(split(0).toInt),
    road = Some(split(1)),
    city = Some(split(2)),
    country = Some(split(3))
  )
}

然后将每个原始数据映射到此函数：

import org.apache.spark.sql.Dataset

val addressDS: Dataset[AddressData] = 
  addressDF.as[AddressRawData].map(singleAddressParser)

这是预期的输出：

scala> addressDS.show(false)
+---------+-----------------------------------------+------+------------------+----------+-------+
|addressId|address                                  |number|road              |city      |country|
+---------+-----------------------------------------+------+------------------+----------+-------+
|ADD001   |384, East Avenue Street, New York, USA   |384   |East Avenue Street|New York  |USA    |
|ADD002   |123, Malccolm Street, Copenhagen, Denmark|123   |Malccolm Street   |Copenhagen|Denmark|
+---------+-----------------------------------------+------+------------------+----------+-------+