如何使动态查询过滤器在pyspark中运行？

debugcn 发表于 Dev

用户名

我需要准备一个解决方案以创建一个参数化解决方案以运行不同的过滤器。例如：我当前正在使用下面的查询将过滤器应用于数据框，但是

input_df.filter("not is_deleted and status == 'Active' and brand in ('abc', 'def')")

需要更改此方法以通过配置构建此查询：

  filters_to_apply = {'table_name' : 'input_df',
    'rule_1' : 'not is_deleted',
    'rule_2' : 'status == "Active"'
    'rule_3' : 'brand in ("abc", "def")'
                              }

filters_to_apply['table_name'].filter(' and '.join([(filters_to_apply[key]) for key in filters_to_apply.keys() if 'rule' in key]))

我收到以下错误消息：AttributeError：'str'对象没有属性'filter'

请您指教

boechat107

首先，了解为什么会出现此错误很重要。

filters_to_apply定义字典的方式是将字符串键映射到字符串值。具有定义的变量input_df与带有字符的字符串无关"input_df"。对于Python，它们是完全不同的两件事。

据我所知，这里有两种选择：

您可以在input_df周围传递变量（假设它在某处定义）并直接在其上应用过滤器（您可以拥有一个函数，其中一个参数是一个DataFrame，因此它可以处理不同的DataFrame）。
使用定义的变量input_df，您可以注册一个临时视图并在以后检索相同的DataFrame。

我的建议

定义一个函数，其参数由DataFrame和规则序列组成。

函数定义示例：

from pyspark.sql import DataFrame
from typing import Iterable

def my_filter(df: DataFrame, conditions: Iterable[str]) -> DataFrame:
    return df.filter(" and ".join(conditions))

用法示例：

df = sparksession.createDataFrame(
    [(1, True, "Active"), (2, False, "Active"), (3, True, "Disabled")],
    ["id", "bool", "status"]
)
df.show()

my_filter(df, ["not bool", "status = 'Active'"]).show()

结果：

+---+-----+--------+
| id| bool|  status|
+---+-----+--------+
|  1| true|  Active|
|  2|false|  Active|
|  3| true|Disabled|
+---+-----+--------+

+---+-----+------+
| id| bool|status|
+---+-----+------+
|  2|false|Active|
+---+-----+------+

本文收集自互联网，转载请注明来源。

如有侵权，请联系[email protected] 删除。