PySpark Pandas UDF 最佳实践答案

【问题标题】：PySpark Pandas UDF Best PracticesPySpark Pandas UDF 最佳实践
【发布时间】：2021-08-13 20:08:15
【问题描述】：

我已经编写了以下 pandas_udf 来计算 PySpark 中的半正弦距离：

def haversine(witness_lat : pd.Series, witness_lon: pd.Series, beacon_lat: pd.Series, beacon_lon: pd.Series) -> pd.Series:
    if None in [witness_lat, witness_lon, beacon_lat, beacon_lon]:
        return None
    else:
        lon1 = witness_lon
        lat1 = witness_lat
        lon2 = beacon_lon
        lat2 = beacon_lat

        lon1, lat1, lon2, lat2 = map(math.radians, [lon1, lat1, lon2, lat2])
        dlon = lon2 - lon1 
        dlat = lat2 - lat1 
        a = np.sin(dlat/2)**2 + np.cos(lat1) * np.cos(lat2) * np.sin(dlon/2)**2
        c = 2 * np.arcsin(np.sqrt(a)) 
        m = 6367000 * c
        return m

@pandas_udf("float", PandasUDFType.SCALAR)
def udf_calc_distance(st_y_witness, st_x_witness, st_y_transmitter, st_x_transmitter):
    distance_df = pd.DataFrame({'st_y_witness' : st_y_witness, 'st_x_witness' : st_x_witness, 'st_y_transmitter' : st_y_transmitter, 'st_x_transmitter' : st_x_transmitter})
    distance_df['distance'] = distance_df.apply(lambda x : haversine(x['st_y_witness'], x['st_x_witness'], x['st_y_transmitter'], x['st_x_transmitter']), axis = 1)
    return distance_df['distance']

此代码运行正常，并给出了我期望的答案，但是我收到了如下所示的折旧警告。

UserWarning: In Python 3.6+ and Spark 3.0+, it is preferred to specify type hints for pandas UDF instead of specifying pandas UDF type which will be deprecated in the future releases. See SPARK-28264 for more details.
  warnings.warn(

我在此处查看了有关 databricks 的最新 pandas_udf 文档：https://docs.databricks.com/spark/latest/spark-sql/udf-python-pandas.html，但我不确定如何将提示与应用格式一起使用。我根据我在堆栈溢出中看到的其他示例设置了我的代码，例如：Passing multiple columns in Pandas UDF PySpark，它遵循将被折旧的格式。

感谢您的帮助！

【问题讨论】：

标签： python pandas apache-spark pyspark bigdata

【解决方案1】：

只需像为 hasrsine 函数一样添加函数类型：

@pandas_udf("float")
def udf_calc_distance(st_y_witness: pd.Series, st_x_witness: pd.Series, st_y_transmitter: pd.Series, st_x_transmitter: pd.Series) -> pd.Series:
    distance_df = pd.DataFrame({'st_y_witness' : st_y_witness, 'st_x_witness' : st_x_witness, 'st_y_transmitter' : st_y_transmitter, 'st_x_transmitter' : st_x_transmitter})
    distance_df['distance'] = distance_df.apply(lambda x : haversine(x['st_y_witness'], x['st_x_witness'], x['st_y_transmitter'], x['st_x_transmitter']), axis = 1)
    return distance_df['distance']

【讨论】：