Я пытаюсь динамически сформировать условие фильтра из структуры dict в Python, и это очень простое условие, которое выдает ошибку ниже:
Final constructed filter condition: Column Type of final_condition: PySparkValueError: [CANNOT_CONVERT_COLUMN_INTO_BOOL] Cannot convert column into bool: please use '&' for 'and', '|' for 'or', '~' for 'not' when building DataFrame boolean expressions.
У меня есть данные и столбец, как показано ниже:
data = [("John", 1930), ("Doe", 1931), ("Jane", 1940)]
columns = ["Name", "Code"]
Я пытаюсь имитировать фильтр ниже, используя динамический код:
df = spark.createDataFrame(data, schema=columns)
df.filter(F.col("Code").isin(1930, 1931)).show()
мой код такой:
{
"conditions": [
{"column": "Code", "operator": "IN", "values": [1930, 1931]}
]
}
и ниже мой код:
def construct_filter_condition(conditions):
"""
Construct filter condition from a list of conditions.
:param conditions: List of conditions
:return: PySpark Column expression
"""
combined_conditions = []
for condition in conditions:
column = condition["column"]
operator = condition["operator"].lower()
values = condition.get("values")
value = condition.get("value")
if operator == "in":
new_condition = F.col(column).isin(values)
elif operator == "eq":
new_condition = F.col(column) == value
elif operator == "ne":
new_condition = F.col(column) != value
elif operator == "lt":
new_condition = F.col(column) < value
elif operator == "le":
new_condition = F.col(column) value
elif operator == "ge":
new_condition = F.col(column) >= value
elif operator == "not_in":
new_condition = ~F.col(column).isin(values)
else:
continue # Skip invalid operators
combined_conditions.append(new_condition)
# Combine all conditions using 'AND' logic
if combined_conditions:
final_condition = combined_conditions[0]
for cond in combined_conditions[1:]:
final_condition = final_condition & cond
return final_condition
else:
return None
Пример условия:
flt_conditions = [
{
"conditions": [
{"column": "CompanyCode", "operator": "IN", "values": [1930, 1931]}
]
}
]
Создание и применение фильтра
final_condition = construct_filter_condition(flt_conditions[0]["conditions"])
if final_condition:
print(f"Constructed filter condition: {final_condition}")
filtered_df = df.filter(final_condition)
print(f"DataFrame filter syntax: df.filter({final_condition})")
filtered_df.show()
else:
print("No valid filter conditions provided.")
Подробнее здесь: https://stackoverflow.com/questions/788 ... y-operator
