Scala - 在多个字段上使用 DF 过滤器

Question

我已经通过 stackoverflow 搜索了几天，但我只是没有找到以下问题的答案。我对 scala 编码真的很陌生，所以这可能是一个非常基本的问题。任何帮助将不胜感激。

我遇到的问题（出现错误）与最后一段代码有关。
我正在尝试从数据框中获取过滤后的记录子集，其中所有过滤后的记录都缺少一个或多个指定字段中的数据。

我在 Eclipse 中使用 Scala IDE Build 4.7.0。
我使用的 pom.xml 文件有 spark-core_2.11，版本 2.0.0

谢谢。
杰西

val source_path = args(0)
val source_file = args(1)

val vFile = sc.textFile(source_path + "/" + source_file)

val vSchema = StructType(
            StructField("FIELD_1",LongType,false)::
            StructField("FIELD_2",LongType,false)::
            StructField("FIELD_3",StringType,true)::
            StructField("FIELD_4",StringType,false)::
            StructField("FIELD_ADD_1",StringType,false)::
            StructField("FIELD_ADD_2",StringType,false)::
            StructField("FIELD_ADD_3",StringType,false)::
            StructField("FIELD_ADD_4",StringType,false)::
            StructField("FIELD_5",StringType,false)::
            StructField("FIELD_6",StringType,false)::
            StructField("FIELD_7",StringType,false)::
            StructField("FIELD_8",StringType,false)::
            Nil)

// val vRow = vFile.map(x=>x.split((char)30, -1)).map(x=> Row(
val vRow = vFile.map(x=>x.split("", -1)).map(x=> Row(
                            x(1).toLong,
                            x(2).toLong,
                            x(3).toString.trim(),
                            x(4).toString.trim(),
                            x(5).toString.trim(),
                            x(6).toString.trim(),
                            x(7).toString.trim(),
                            x(8).toString.trim(),
                            x(9).toString.trim(),
                            x(10).toString.trim(),
                            x(11).toString.trim(),
                            x(12).toString.trim()
                        ))

val dfData = sqlContext.createDataFrame(vRow.distinct(),vSchema)

val dfBlankRecords = dfData.filter(x => (
                    x.trim(col("FIELD_ADD_1")) == "" ||
                    x.trim(col("FIELD_ADD_2")) == "" ||
                    x.trim(col("FIELD_ADD_3")) == "" ||
                    x.trim(col("FIELD_ADD_4")) == ""
                ))

Answer 1

spark.read.* 函数将数据直接读入 Dataset/Dataframe API，避免（在某种程度上）需要架构定义，并完全使用 RDD API。

val source_path = args(0)
val source_file = args(1)

val dfData = spark.read.textFile(source_path + "/" + source_file)
  .flatMap(l => {
    val a = l.split('\u001e'.toString, -1).map(_.trim())
    val f1 = a(0).toLong
    val f2 = a(1).toLong
    val Array(f3, f4, fa1, fa2, fa3, fa4, f5, f6, f7, f8) = a.slice(2,12)

    if (fa1 == "" ||
        fa2 == "" ||
        fa3 == "" ||
        fa4 == "") {
      Some(f1, f2, f3, f4, fa1, fa2, fa3, fa4, f5, f6, f7, f8)
    } else {
      None
    }
  }).toDF("FIELD_1", "FIELD_2", "FIELD_3", "FIELD_4",
          "FIELD_ADD_1", "FIELD_ADD_2", "FIELD_ADD_3", "FIELD_ADD_4",
          "FIELD_5", "FIELD_6", "FIELD_7", "FIELD_8")

我认为这会产生你想要的结果。我相信比我更好的人可以更好地优化它，并且使用更简洁的代码。

请注意，数组索引为零，如果您有意选择特定字段，则必须对其进行调整。我也不确定'\u001e'（十六进制值 30）是否是您拆分字符串所需的合适值。