如何在spark sql中传递动态条件列表

Question

我有一个用于星火sql查询的筛选条件列表，其形式为

Map<String, List<String>> conditions

假设初始数据集是

Dataset<Row> dataSet

我想使用地图中的每一个条目，比如(StringA，ListA)，(StringB，ListB).以此类推，在过滤器中，每个条目将是一个或条件

(col("my_col1").equals("StringA").and(col("my_col2").isInCollection(ListA)))

OR

(col("my_col1").equals("StringB").and(col("my_col2").isInCollection(ListB)))

OR

(col("my_col1").equals("StringC").and(col("my_col2").isInCollection(ListC)))
....

dataset是一个具有(my_col1，my_col2)分区键的Cassandra表。

我预计地图上不会有超过5-6个条目

Answer 1

使用isin

d.filter(col("value").isin(desiredThings: _*))

但是如果你真的想要foldLeft，你必须提供基本条件

d.filter(desiredThings.foldLeft(lit(false))(
(acc, x) => (acc || col("value") === (x)))
) 

或者，若要与筛选器一起使用或在何处使用，可以使用以下方法生成SQL表达式：

val filterExpr = desiredThings.map( v => s"value = $v").mkString(" or ")

然后把它当作

d.filter(filterExpr).show
// or
d.where(filterExpr).show

//+-----+
//|value|
//+-----+
//|    1|
//|    3|
//+-----+