过滤Spark SQL数据框以供距离

Question

我有一个带有列纬度和经度的火花SQL DataDrame，我试图通过计算到输入的距离来过滤掉以下阈值以下的行。我当前的代码看起来像。我正在使用geopy（great_circle）来计算LAT长对之间的距离。

from geopy.distance import great_circle
point = (10, 20)
threshold = 10
filtered_df = df.filter(great_circle(point, (df.lat, df.lon)) < threshold)

运行此代码时，我会收到以下错误

ValueError: Cannot convert column into bool: please use '&' for 'and', '|' for 'or', '~' for 'not' when building DataFrame boolean expressions. 

我对滤波器表达式的哪一部分错误感到困惑。

Answer 1

您不能在DataFrame上应用普通的Python函数。您必须使用udf：

from pyspark.sql.functions import udf
@udf("float")
def great_circle_udf(x, y):
    return great_circle(x, y).kilometers

并将其与列

应用

from pyspark.sql.functions import lit, struct
point = struct(lit(10), lit(20))
df.filter(great_circle_udf(point, struct(df.lat, df.lon)) < threshold))

Decorator语法将从2.2起可以正常工作，对于早期版本，您需要标准udf调用：

udf(great_circle, FloatType())