Я пытаюсь реализовать калькулятор haversine_distance в pyspark. Я повторно использую код Python, который я использовал раньше для той же цели, поэтому вот что я сделал: 1. Реализовать функцию для harvesine_distance как udf 2. В моем кадре данных используется это для вычисления расстояния между двумя точками широты/долготы 3. Когда я запускаю проверку значений, это кажется правильным 4. Но когда я попытался поместить некоторое предложение where в dist_km, я получаю сообщение об ошибке:
File "<stdin>", line 13, in haversine_distance
TypeError: a float is required
Примеры значений в кадре данных upd_join_final66_df:
+------------+------------+--------+---------+---------+
|delv_lat_upd|delv_lng_upd|LATITUDE|LONGITUDE| dist_km|
+------------+------------+--------+---------+---------+
| 33.7871| -84.382| 33.7602| -84.5398|14.893161|
| 33.7871| -84.382| 33.7602| -84.5398|14.893161|
| 33.7874| -84.3822| 33.7602| -84.5398|14.881769|
| 33.7874| -84.3822| 33.7602| -84.5398|14.881769|
| 33.7874| -84.3822| 33.7602| -84.5398|14.881769|
| 33.7874| -84.3822| 33.7602| -84.5398|14.881769|
| 33.7874| -84.3822| 33.7602| -84.5398|14.881769|
| 33.7872| -84.3822| 33.7602| -84.5398| 14.87728|
| 33.7872| -84.3822| 33.7602| -84.5398| 14.87728|
| 33.7872| -84.3822| 33.7602| -84.5398| 14.87728|
| 33.7872| -84.3822| 33.7602| -84.5398| 14.87728|
| 33.7872| -84.3822| 33.7602| -84.5398| 14.87728|
| 33.7871| -84.3821| 33.7602| -84.5398|14.884104|
| 33.7871| -84.3821| 33.7602| -84.5398|14.884104|
| 33.7869| -84.382| 33.7602| -84.5398|14.888724|
| 33.7871| -84.3825| 33.7602| -84.5398|14.847878|
| 33.7869| -84.3818| 33.7602| -84.5398|14.906844|
| 33.7869| -84.3818| 33.7602| -84.5398|14.906844|
| 33.7869| -84.3818| 33.7602| -84.5398|14.906844|
| 33.7869| -84.3818| 33.7602| -84.5398|14.906844|
+------------+------------+--------+---------+---------+
Код:
from math import radians, cos, sin, asin, sqrt, atan2, pi
def haversine_distance(lat1, lon1, lat2, lon2):
"""
Calculate the great circle distance between two points
on the earth (specified in decimal degrees)
"""
deg2rad = pi/180.0
# convert decimal degrees to radians
lon1, lat1, lon2, lat2 = map(radians, [lon1, lat1, lon2, lat2])
# haversine formula
dlon = (lon2 - lon1)
dlat = (lat2 - lat1)
a = sin(dlat/2.0)**2.0 + cos(lat1) * cos(lat2) * sin(dlon/2.0)**2.0
c = 2.0 * asin(sqrt(a))
#c = 2.0 * atan2(sqrt(a), sqrt(1.0-a))
r = 6372.8 # Radius of earth in kilometers. Use 3956 for miles #No rounding R = 3959.87433 (miles), 6372.8(km)
return c * r
haversine_distance_udf = udf(haversine_distance, FloatType())
upd_join_final66_df = upd_join_final66_df.withColumn('dist_km', \
haversine_distance_udf(upd_join_final66_df['LATITUDE'],upd_join_final66_df['LONGITUDE']\
,upd_join_final66_df['delv_lat_upd'],upd_join_final66_df['delv_lng_upd'])\
)
upd_join_final66_df.registerTempTable("fac66")
Когда я запускаю приведенную ниже команду, нет ОШИБКИ для выборочной проверки
spark.sql("select delv_lat_upd, delv_lng_upd, LATITUDE, LONGITUDE, dist_km \
from fac66 \
").show()
Когда я пытался опросить dist_km, я получаю сообщение об ошибке
An error occurred while calling o4882.showString.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1041.0 failed 4 times, most recent failure: Lost task 0.3 in stage 1041.0 (TID 81684, 10.0.0.15, executor 2): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/usr/hdp/current/spark2-client/python/pyspark/worker.py", line 171, in main
process()
File "/usr/hdp/current/spark2-client/python/pyspark/worker.py", line 166, in process
serializer.dump_stream(func(split_index, iterator), outfile)
File "/usr/hdp/current/spark2-client/python/pyspark/serializers.py", line 220, in dump_stream
self.serializer.dump_stream(self._batched(iterator), stream)
File "/usr/hdp/current/spark2-client/python/pyspark/serializers.py", line 138, in dump_stream
for obj in iterator:
File "/usr/hdp/current/spark2-client/python/pyspark/serializers.py", line 209, in _batched
for item in iterator:
File "<string>", line 1, in <lambda>
File "/usr/hdp/current/spark2-client/python/pyspark/worker.py", line 70, in <lambda>
return lambda *a: f(*a)
File "<stdin>", line 13, in haversine_distance
TypeError: a float is required
