Pyspark - 将数据帧列的值提供给函数并将函数的输出附加到原始数据帧答案

【问题标题】：Pyspark - feed values of dataframe column to function and append output of function to original dataframePyspark - 将数据帧列的值提供给函数并将函数的输出附加到原始数据帧
【发布时间】：2020-09-13 14:59:47
【问题描述】：

我有一个 PYSPARK 数据帧 df，其值为“纬度”和“经度”：

+---------+---------+
| latitude|longitude|
+---------+---------+
|51.822872| 4.905615|
|51.819645| 4.961687|
| 51.81964| 4.961713|
| 51.82256| 4.911187|
|51.819263| 4.904488|
+---------+---------+

我想从数据框列中获取 UTM 坐标（“x”和“y”）。为此，我需要将值“经度”和“纬度”从 pyproj 提供给以下函数。结果“x”和“y”应附加到原始数据帧df。这就是我在 Pandas 中的做法：

from pyproj import Proj
pp = Proj(proj='utm',zone=31,ellps='WGS84', preserve_units=False)
xx, yy = pp(df["longitude"].values, df["latitude"].values)
df["X"] = xx
df["Y"] = yy

我将如何在 Pyspark 中执行此操作？

【问题讨论】：

标签： dataframe pyspark pyspark-dataframes utm

【解决方案1】：

使用pandas_udf，向函数提供一个数组，然后也返回一个数组。见下文：

from pyspark.sql.functions import array, pandas_udf, PandasUDFType
from pyproj import Proj
from pandas import Series

@pandas_udf('array<double>', PandasUDFType.SCALAR)
def get_utm(x):
  pp = Proj(proj='utm',zone=31,ellps='WGS84', preserve_units=False)
  return Series([ pp(e[0], e[1]) for e in x ])

df.withColumn('utm', get_utm(array('longitude','latitude'))) \
  .selectExpr("*", "utm[0] as X", "utm[1] as Y") \
  .show()

【讨论】：

我在df.withColumn('utm', get_utm(array('longitude','latitude'))) \ .selectExpr("*", "utm[0] as X", "utm[1] as Y") 中遇到了各种错误，例如： - Python worker 无法连接回来。 - java.net.SocketTimeoutException: Accept timed out 有没有其他方法或者这些错误值得解决？
@Jeroen，关于 pandas_udf 和 pyarrow 版本存在一个已知问题，请参见链接：stackoverflow.com/questions/61202005/…。如果不是同一个问题，您可以发布错误吗？
我检查了我的 Pyspark 版本，它是 v2.4.5，pyarrow 版本是 0.13。根据链接，我不得不降级到 0.10 版，我这样做了。我检查了 conda list，它显示 Pyarrow 的 0.10 版本。但现在我得到了错误：ImportError: PyArrow >= 0.8.0 must be installed; however, it was not found.
你能给我你的 PyArrow 和 Pyspark 版本吗？
我接受了答案，但仍未解决 Pyarrow 问题。我会在这个周末更多地研究它。