如何在 Spark 中使用 Scala 从数据框中的单元格中提取表格信息

Question

我需要从数据框行中的单元格中提取制表符分隔的数据

我尝试使用拆分方法，但它不起作用

val df = spark.sql("select _time, _raw, host from logs")

    val extractedData = df.filter(
      $"host" === "ausflscgap01.us.dell.com" ||
      $"host" ==="ausflscgap02.us.dell.com" ||
      $"host" === "ausplscgap01.us.dell.com" ||
      $"host" === "ausplscgap02.us.dell.com")
    .withColumn("splitted", split($"_raw", "\t"))
      .select($"splitted".getItem(5)
        .alias("pctIdle"))
      .show()

行中的实际数据：

CPU    pctUser    pctNice  pctSystem  pctIowait    pctIdle
all       9.55       0.00      36.18       1.51      52.76
0        10.00       0.00      37.00       4.00      49.00
1         9.00       0.00      34.00       0.00      57.00

我只需要为“所有”行提取 pctIdle 列 预期输出

pctIdle
52.76

Answer 1

如果我对您的理解正确，并且您在 logs 表中的文本字段 _raw 中有“行中的实际数据”，比你需要这样的东西：

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

val extractPctIdle = udf{(raw: String) =>
raw
  .split("\n")
  .map(_.split("\t"))
  .find(_(0) == "all")
  .map(_(5))
  .getOrElse("unknown")
}

val extractedData = df.filter(
      $"host" === "ausflscgap01.us.dell.com" ||
        $"host" ==="ausflscgap02.us.dell.com" ||
        $"host" === "ausplscgap01.us.dell.com" ||
        $"host" === "ausplscgap02.us.dell.com")
      .withColumn("pctIdle", extractPctIdle($"_raw"))
      .show()

即您可以通过自定义 udf 解析您的 _raw 字段。 这是最简单的版本，但最好在 _raw 字段格式错误的情况下进行一些错误处理。

案例是这样建模的：

case class R(host: String, _raw: String)

val df = Seq(
      R("ausflscgap02.us.dell.com", "CPU\tpctUser\tpctNice\tpctSystem\tpctIowait\tpctIdle\nall\t9.55\t0.00\t36.18\t1.51\t52.76\n0\t10.00\t0.00\t37.00\t4.00\t49.00\n1\t9.00\t0.00\t34.00\t0.00\t57.00"),
      R("ausplscgap01.us.dell.com", "CPU\tpctUser\tpctNice\tpctSystem\tpctIowait\tpctIdle\nall\t9.55\t0.00\t36.18\t1.51\t52.76\n0\t10.00\t0.00\t37.00\t4.00\t49.00\n1\t9.00\t0.00\t34.00\t0.00\t57.00")
    ).toDF()

编辑

如果您需要 _raw 中的多列数据：

case class RawInfo(pctUser: String, pctIdle: String)

val extractRawInfo = udf{(raw: String) =>
      val all = raw
        .split("\n")
        .map(_.split("\t"))
        .find(_(0) == "all")

      def getValue(pos: Int) = all.map(_(pos)).getOrElse("unknown")

      RawInfo(
        pctUser = getValue(1),
        pctIdle = getValue(5))
    }

    df.filter($"host".isin("ausflscgap01.us.dell.com", "ausflscgap02.us.dell.com", "ausplscgap01.us.dell.com", "ausplscgap02.us.dell.com"))
      .withColumn("info", extractRawInfo($"_raw"))
      .select("host", "info.pctUser", "info.pctIdle")
      .show()

备注：可能只从 udf 返回 Array[string] 并稍后检索特定列（如 $"info"(0).as("pctUser")），但我更喜欢上面显示的类型化解决方案。