scala 对Spark`Dataset`/`Dataframe`有条件地应用`filter`/`where`

sshcrbum  于 2022-11-09  发布在  Scala
关注(0)|答案(2)|浏览(193)

大家好,我有一个函数,可以从一些S3位置加载数据集并返回感兴趣的数据

private def filterBrowseIndex(spark: SparkSession, s3BrowseIndex: String, mids: Seq[String] = Seq(), indices: Seq[String] = Seq()): Dataset[BrowseIndex] = {
import spark.implicits._

spark
  .sparkContext.textFile(s3BrowseIndex)
  // split text dataset
  .map(line => line.split("\\s+"))
  // get types for attributes
  .map(BrowseIndex.strAttributesToBrowseIndex)
  // cast it to a dataset (requires implicit conversions)
  .toDS()
  // pick rows for the given marketplaces
  .where($"mid".isin(mids: _*))
  // pick rows for the given indices
  .where($"index".isin(indices: _*))

}
如果有人提供mids = Seq()indices = Seq(),此实现将过滤掉所有内容。另一方面,我希望语义是“只有当mids不为空时才应用这个WHERE子句”(与indices相同),这样,如果函数的用户提供空序列,则不会发生过滤。
有没有一种很好的实用方法来做到这一点?

crcmnpdw

crcmnpdw1#

如果你不介意略显复杂的逻辑,拉斐尔·罗斯的答案对于应用过滤器这一特定问题来说是一个很好的选择。适用于任何条件转换的一般解决方案(不只是过滤并且不只是在决策分支之一上什么都不做)是使用transform,例如,

spark
  .sparkContext.textFile(s3BrowseIndex)
  // split text dataset
  .map(line => line.split("\\s+"))
  // get types for attributes
  .map(BrowseIndex.strAttributesToBrowseIndex)
  // cast it to a dataset (requires implicit conversions)
  .toDS()
  .transform { ds =>
    // pick rows for the given marketplaces
    if (mids.isEmpty) ds
    else ds.where($"mid".isin(mids: _*))
  }
  .transform { ds =>
    // pick rows for the given indices
    if (indices.isEmpty) ds
    else ds.where($"index".isin(indices: _*))
  }

如果您使用的是稳定类型的数据集(或 Dataframe ,即Dataset[Row]),则transform可能非常有用,因为您可以构建转换函数序列,然后应用它们:

transformations.foldLeft(ds)(_ transform _)

在许多情况下,这种方法有助于代码重用和可测试性。

5f0d552i

5f0d552i2#

您可以使用短路评估,这应该仅在所提供的Seq不为空时应用过滤器:

import org.apache.spark.sql.functions.lit

spark
    .sparkContext.textFile(s3BrowseIndex)
    // split text dataset
    .map(line => line.split("\\s+"))
    // get types for attributes
    .map(BrowseIndex.strAttributesToBrowseIndex)
    // cast it to a dataset (requires implicit conversions)
    .toDS()
    // pick rows for the given marketplaces
    .where(lit(mids.isEmpty) or $"mid".isin(mids: _*))
    // pick rows for the given indices
    .where(lit(indices.isEmpty) or $"index".isin(indices: _*))

相关问题