-1

我阅读了以下博客，发现 API 非常有用。

https://databricks.com/blog/2017/02/23/working-complex-data-formats-structured-streaming-apache-spark-2-1.html

在博客中，有很多数据选择示例。就像使用输入

{
  "a": {
     "b": 1
  }
}

应用Scala: events.select("a.b")，输出将是

{
  "b": 1
}

但是博客中没有提到数据类型转换。说我有以下输入：

{
  "timestampInSec": "1514917353",
  "ip": "123.39.76.112",
  "money": "USD256",
  "countInString": "6"
}

预期的输出是：

{
  "timestamp": "2018-01-02 11:22:33",
  "ip_long": 2066173040,
  "currency": "USD",
  "money_amount": 256,
  "count": 6
}

有一些转换未包含在org.apache.spark.sql.functions._：

  • 时间戳在秒，是字符串类型
  • 将IP转换为长
  • 拆分USD256为两列并将其中一列转换为数字
  • 将字符串转换为数字

另一件事是错误处理和默认值。如果输入无效，例如：

{
  "timestampInSec": "N/A",
  "money": "999",
  "countInString": "Number-Six"
}

预计输出可以是

{
  "timestamp": "1970-01-01 00:00:00",
  "ip_long": 0,
  "currency": "NA",
  "money_amount": 999,
  "count": -1
}
  • 输入timestampInSec不是数字。预计使用 0 并创建一个时间戳字符串作为返回值
  • ip输入中缺少。预计使用默认值 0。
  • money字段不完整。它有金额但错过了货币。预计将NA用作默认货币并正确翻译money_amount
  • countInString不是一个数字。预计使用-1( not 0 ) 作为默认值。

这些需求并不常见，需要一些自定义的业务逻辑代码。

我确实检查了一些功能，例如to_timestamp. 有一些代码生成的东西，似乎不太容易添加新功能。是否有一些关于编写自定义转换函数的指南/文档？有没有简单的方法来满足要求？

4

1 回答 1

2

对所有人：

import org.apache.spark.sql.functions._

  • 时间戳在秒，是字符串类型

    val timestamp = coalesce(
   $"timestampInSec".cast("long").cast("timestamp"), 
   lit(0).cast("timestamp")
).alias("timestamp")

  • 将 USD256 拆分为两列并将其中一列转换为数字

    val currencyPattern = "^([A-Z]+)?([0-9]+)$"

val currency = (trim(regexp_extract($"money", currencyPattern, 1)) match {
  case c => when(length(c) === 0, "NA").otherwise(c)
}).alias("currency")
val amount = regexp_extract($"money", currencyPattern, 2)
  .cast("decimal(38, 0)").alias("money_amount")

  • 将字符串转换为数字

    val count = coalesce($"countInString".cast("long"), lit(-1)).alias("count")

  • 将IP转换为长

    val ipPattern = "^([0-9]{1,3})\\.([0-9]{1,3})\\.([0-9]{1,3})\\.([0-9]{1,3})"
val ip_long = coalesce(Seq((1, 24), (2, 16), (3, 8), (4, 0)).map {
  case (group, numBits) => shiftLeft(
    regexp_extract($"ip", ipPattern, group).cast("long"),
    numBits
  )
}.reduce(_ + _), lit(0)).alias("ip_long")

结果

val df = Seq(
  ("1514917353", "123.39.76.112", "USD256", "6"),
  ("N/A", null, "999", null)
).toDF("timestampInSec", "ip", "money", "countInString")

df.select(timestamp, currency, amount, count, ip_long).show
// +-------------------+--------+------------+-----+----------+
// |          timestamp|currency|money_amount|count|   ip_long|
// +-------------------+--------+------------+-----+----------+
// |2018-01-02 18:22:33|     USD|         256|    6|2066173040|
// |1970-01-01 00:00:00|      NA|         999|   -1|         0|
// +-------------------+--------+------------+-----+----------+
于 2018-02-09T18:22:36.427 回答