PySpark使用Pandas UDF经纬度转H3多节点集群环境配置报错求解

问题描述 #

需要将PySpark DataFrame中的经纬度值转换为Uber H3六边形网格编码，对应调用函数为h3.geo_to_h3(lat=lat, lng=lon, resolution=7)，在PySpark集群运行时持续报错。

已完成的环境配置 #

参考Databricks官方方案配置集群环境，执行命令如下：

1. conda create -y -n pyspark_conda_env -c conda-forge pyarrow pandas h3 numpy python=3.7 conda-pack
2. 执行conda init --all后关闭并重新打开终端窗口
3. conda activate pyspark_conda_env
4. conda pack -f -o pyspark_conda_env.tar.gz
   随后在Jupyter Notebook创建SparkSession时引入生成的tar.gz依赖包，配置代码如下：

spark = SparkSession.builder.master("yarn").appName("test").config("spark.yarn.dist.archives","<path>/pyspark_conda_env.tar.gz#environment").getOrCreate()

已编写的业务代码 #

Pandas UDF代码如下，该代码在单节点Spark集群可正常运行，但切换至多节点工作集群后出现报错：

# 定义经纬度转H3编码的函数
def convert_to_h3(lat : pd.Series, lon : pd.Series) -> pd.Series:
    import h3 as h3
    import numpy as np
    if ((None in [lat, lon]) | (np.isnan(lat))):
        return None
    else:
        return (h3.geo_to_h3(lat=lat, lng=lon, resolution=7))

@f.pandas_udf('string', f.PandasUDFType.SCALAR)
def udf_convert_to_h3(lat : pd.Series, lon : pd.Series) -> pd.Series:
    import pandas as pd
    import numpy as np
    df = pd.DataFrame({'lat' : lat, 'lon' : lon})
    df['h3_res7'] = df.apply(lambda x : convert_to_h3(x['lat'], x['lon']), axis = 1)
    return df['h3_res7']

调用UDF新增H3编码列的代码如下：

trip_starts = trip_starts.withColumn('h3_res7', udf_convert_to_h3(f.col('latitude'), f.col('longitude')))

报错信息 #

21/07/15 20:05:22 WARN YarnSchedulerBackend$YarnSchedulerEndpoint: Requesting driver to remove executor 139 for reason Container marked as failed: container_1626376534301_0015_01_000158 on host: ip-xx-xxx-xx-xxx.aws.com. Exit status: -100. Diagnostics: Container released on a *lost* node.

已尝试缩小处理的数据量，仍未解决该问题。受公司政策限制，启动集群时无法使用bootstrap脚本，希望基于Databricks官方给出的PySpark环境配置方案解决该问题。 #

问题排查与解决方案 #

核心原因定位 #

Exit status -100对应的节点丢失报错，结合单节点正常多节点异常的现象，问题出在两个点：

1. 仅配置了依赖包分发参数，没有指定executor使用打包好的conda环境中的Python解释器，导致执行节点调用系统默认Python环境，缺少h3依赖，进程启动失败被Yarn回收
2. Pandas UDF内部用了逐行apply写法，性能极低，executor负载过高也会触发Yarn主动回收容器

修复步骤 #

1. 补充SparkSession配置，明确指定执行节点的Python解释器路径

spark = SparkSession.builder.master("yarn").appName("test")\
    .config("spark.yarn.dist.archives","<path>/pyspark_conda_env.tar.gz#environment")\
    .config("spark.executorEnv.PYSPARK_PYTHON", "./environment/bin/python")\
    .config("spark.driverEnv.PYSPARK_PYTHON", "python")\
    .getOrCreate()

2. 优化Pandas UDF写法，用向量化操作代替逐行apply，降低执行节点负载

@f.pandas_udf('string', f.PandasUDFType.SCALAR)
def udf_convert_to_h3(lat : pd.Series, lon : pd.Series) -> pd.Series:
    import h3
    import pandas as pd
    # 批量过滤空值
    mask = lat.isna() | lon.isna()
    res = pd.Series([None]*len(lat))
    # 批量转换非空值，避免逐行判断
    res[~mask] = [h3.geo_to_h3(la, lo, 7) for la, lo in zip(lat[~mask], lon[~mask])]
    return res

3. 若仍出现容器被回收的问题，可临时上调executor资源配额，避免负载过高被Yarn kill

.config("spark.executor.memory", "8g")\
.config("spark.executor.cores", "4")\
.config("spark.yarn.executor.memoryOverhead", "2g")

内容的提问来源于stack exchange，提问作者zorrrba