如何在没有爆炸的情况下绘制星火阵列的地图？

Question

我的情况是，我有一个数组列，我想要过滤。请考虑以下几点：

+------------------------------------------------------+
|                                                column|
+------------------------------------------------------+
|[prefix1-whatever, prefix2-whatever, prefix4-whatever]|
|[prefix1-whatever, prefix2-whatever, prefix3-whatever]|
|[prefix1-whatever, prefix2-whatever, prefix5-whatever]|
|[prefix1-whatever, prefix2-whatever, prefix3-whatever]|
+------------------------------------------------------+

我只想过滤包含前缀-4，前缀-5，前缀-6，前缀-7，.因此，在这里使用“或”语句是不可伸缩的。

当然，我可以：

val prefixesList = List("prefix-4", "prefix-5", "prefix-6", "prefix-7")

df
.withColumn("prefix", explode($"column"))
.withColumn("prefix", split($"prefix", "\\-").getItem(0))
.withColumn("filterColumn", $"prefix".inInCollection(prefixesList))

但这涉及到爆炸，我想避免。我现在的计划是从prefixesList中定义一个数组列，然后使用array_intersect对其进行过滤--但是，为了使其正常工作，我必须去掉-whatever部分(显然，每个条目都不同)。如果这是一个Scala数组，我可以很容易地对它做一个映射。但是，作为火花阵列，我不知道这是否可能。

DR I有一个包含数组列的数据。我试图在没有爆炸的情况下操纵它并过滤它(因为，如果我真的爆炸了，我将不得不稍后操纵它来逆转爆炸，我想避免它)。

我能在不爆炸的情况下做到这一点吗？如果是这样的话，是怎么做的？

Answer 1

不确定我是否正确地理解了您的问题:您希望保留不包含prefixesList中任何前缀的所有行

如果是这样，您可以编写自己的过滤器函数：

def filterPrefixes (row: Row) : Boolean = {
  for( s <- row.getSeq[String](0)) {
    for( p <- Seq("prefix4", "prefix5", "prefix6", "prefix7")) {
      if( s.startsWith(p) ) {
        return false
      }
    }
  }
  return true
}

然后将其用作筛选器调用的参数：

df.filter(filterPrefixes _)
  .show(false)

版画

+------------------------------------------------------+
|column                                                |
+------------------------------------------------------+
|[prefix1-whatever, prefix2-whatever, prefix3-whatever]|
|[prefix1-whatever, prefix2-whatever, prefix3-whatever]|
+------------------------------------------------------+

Answer 2

将Dataframe转换为Dataset[Array[String]]并将这些数组映射为整体元素相对来说很简单。基本思想是，您可以轻松地遍历数组列表，而不必将整个数据集夷为平地。

val df = Seq(Seq("prefix1-whatever", "prefix2-whatever", "prefix4-whatever"),
             Seq("prefix1-whatever", "prefix2-whatever", "prefix3-whatever"),
             Seq("prefix1-whatever", "prefix2-whatever", "prefix5-whatever"),
             Seq("prefix1-whatever", "prefix2-whatever", "prefix3-whatever")
).toDF("column")

val pl = List("prefix4", "prefix5", "prefix6", "prefix7")

val df2 = df.as[Array[String]].map(a => {
    a.flatMap(s => {
        val start = s.split("-")(0)
        if(pl.contains(start)) {
            Some(s)
        } else {
            None
        }
    })
}).toDF("column")

df2.show(false)

上述守则的结果是：

+------------------+
|column            |
+------------------+
|[prefix4-whatever]|
|[]                |
|[prefix5-whatever]|
|[]                |
+------------------+

我不完全确定这将如何将性能明智地与实际扁平和重组数据集进行比较。这样做错过了任何催化剂优化，但避免了许多不必要的数据洗牌。

我更正了前缀列表中的一个小问题，因为“前缀-N”与数据的模式不匹配。

Answer 3

您可以使用SQL实现它。如果希望只保留包含任何值prefix-4, prefix-5, prefix-6, prefix-7的行，则可以使用arrays_overlap函数。否则，如果希望保留包含所有值的行，可以尝试array_intersect，然后检查其大小是否等于计算值。

 val df = Seq(
  Seq("prefix1-a", "prefix2-b", "prefix3-c", "prefix4-d"),
  Seq("prefix4-e", "prefix5-f", "prefix6-g", "prefix7-h", "prefix8-i"),
  Seq("prefix6-a", "prefix7-b", "prefix8-c", "prefix9-d"),
  Seq("prefix8-d", "prefix9-e", "prefix10-c", "prefix12-a")
).toDF("arr")


val schema = StructType(Seq(
  StructField("arr", ArrayType.apply(StringType)),
  StructField("arr2", ArrayType.apply(StringType))
))
val encoder = RowEncoder(schema)

val df2 = df.map(s =>
  (s.getSeq[String](0).toArray, s.getSeq[String](0).map(s => s.substring(0, s.indexOf("-"))).toArray)
).map(s => RowFactory.create(s._1, s._2))(encoder)


val prefixesList = Array("prefix4", "prefix5", "prefix6", "prefix7")
val prefixesListSize = prefixesList.size
val prefixesListCol = lit(prefixesList)

df2.select('arr,'arr2,
  arrays_overlap('arr2,prefixesListCol).as("OR"),
  (size(array_intersect('arr2,prefixesListCol)) === prefixesListSize).as("AND")
).show(false)

它会给你：

+-------------------------------------------------------+---------------------------------------------+-----+-----+
|arr                                                    |arr2                                         |OR   |AND  |
+-------------------------------------------------------+---------------------------------------------+-----+-----+
|[prefix1-a, prefix2-b, prefix3-c, prefix4-d]           |[prefix1, prefix2, prefix3, prefix4]         |true |false|
|[prefix4-e, prefix5-f, prefix6-g, prefix7-h, prefix8-i]|[prefix4, prefix5, prefix6, prefix7, prefix8]|true |true |
|[prefix6-a, prefix7-b, prefix8-c, prefix9-d]           |[prefix6, prefix7, prefix8, prefix9]         |true |false|
|[prefix8-d, prefix9-e, prefix10-c, prefix12-a]         |[prefix8, prefix9, prefix10, prefix12]       |false|false|
+-------------------------------------------------------+---------------------------------------------+-----+-----+

最后，您可以使用：

df2.filter(size(array_intersect('arr2,prefixesListCol)) === prefixesListSize).show(false)

你会得到以下结果：

+-------------------------------------------------------+---------------------------------------------+
|arr                                                    |arr2                                         |
+-------------------------------------------------------+---------------------------------------------+
|[prefix4-e, prefix5-f, prefix6-g, prefix7-h, prefix8-i]|[prefix4, prefix5, prefix6, prefix7, prefix8]|
+-------------------------------------------------------+---------------------------------------------+