PySpark Dataframe 根据函数返回值创建新列

Question

我有一个数据框，我想根据函数返回的值添加一个新列。此函数的参数是来自同一数据帧的四列。这个和这个有点类似于我想要的，但没有回答我的问题。

这是我的数据框（有更多的列然后这四个）

 + ------ + ------ + ------ + ------ +

 | lat1   | lng1   | lat2   | lng2   |

 + ------ + ------ + ------ + ------ +

 | -32.92 | 151.80 | -32.89 | 151.71 |

 | -32.92 | 151.80 | -32.89 | 151.71 |

 | -32.92 | 151.80 | -32.89 | 151.71 |

 | -32.92 | 151.80 | -32.89 | 151.71 |

 | -32.92 | 151.80 | -32.89 | 151.71 |

 + ------ + ------ + ------ + ------ +

我想添加另一列“距离”，它是两个位置点（纬度/经度）之间的总距离。我有一个函数，它将四个位置点作为参数，并将差值作为浮点数返回。

def get_distance(lat_1, lng_1, lat_2, lng_2): 

  d_lat = lat_2 - lat_1

  d_lng = lng_2 - lng_1 

  temp = (  

  math.sin(d_lat / 2) ** 2 

    + math.cos(lat_1) 

    * math.cos(lat_2) 

    * math.sin(d_lng / 2) ** 2

  )

  return 6367.0 * (2 * math.asin(math.sqrt(temp))) 

这是我的尝试导致错误，我也不确定这种方法，它基于我已经提到的其他问题。

udf_func = udf(lambda lat_1, lng_1, lat_2, lng_2: get_distance(lat_1, lng_1, lat_2, lng_2), returnType=FloatType())

df1 = df.withColumn('difference', udf_func(df.lat1, df_lng1, df.lat2, df.lng2))

df_subset1.show()

Answer 1

关于 unicode 的堆栈跟踪部分表明该列的类型是 StringType，因为您不能减去两个字符串。您可以使用df.printSchema().如果您float(lat1)在任何计算之前将所有 lats 和 longs 转换为浮点数（例如），您的 udf 应该可以正常执行。