小编vai*_*ira的帖子

我是spark和python的新手.任何帮助赞赏.

我有一个UDF并创建了一个带有美国zipcd,纬度和经度的火花数据帧

UDF:

import math
def distance(origin, destination):
lat1, lon1 = origin
lat2, lon2 = destination
radius = 6371 # km
dlat = math.radians(lat2-lat1)
dlon = math.radians(lon2-lon1)
a = math.sin(dlat/2) * math.sin(dlat/2) + math.cos(math.radians(lat1)) \
    * math.cos(math.radians(lat2)) * math.sin(dlon/2) * math.sin(dlon/2)
c = 2 * math.atan2(math.sqrt(a), math.sqrt(1-a))
d = radius * c
return d

示例UDF输出:

distance((101,121),(-121,-212)) 

15447.812243421227

数据帧:

zip=spark.read.option("sep", ",").csv('wasb://hdiazurepoc@dsazurepoc.blob.core.windows.net/main/zip.txt')
zip1=zip.select(zip._c0,zip._c1.cast("Double"),zip._c2.cast("Double"))

zip1数据示例:

zip1.first()        

行(_c0 = u'00601',_ c1 = 18.180555,_c2 = -66.749961)

现在我试图将经度和经度从df zip1传递到udf距离,但我得到的错误就像"需要浮动".我相信udf没有从df字段获取数据,而是将df列作为常量值读取; 因此我得到了错误.

z=zip1.select(distance((zip1._c1,100.23),(zip1._c2,-99.21)))

回溯(最近一次调用最后一次):
文件"",第1行,在
文件"",第5行,距离
TypeError:需要一个浮点数

请告诉我将df字段传递给udf的正确方法.