PySpark经纬度推断时区并行运行报错及解决方案咨询

针对你遇到的情况，我来逐个解答你的三个问题：

1. Spark原生方法是否可完成该任务？ #

Spark本身并没有直接根据经纬度推断时区的原生API。Spark SQL提供的from_utc_timestamp、to_timestamp等函数是用来做时区转换的，但前提是你已经知道目标时区。所以要完成经纬度到时区的映射，还是得依赖第三方库，不过我们可以优化库的使用方式来适配Spark的分布式环境。

2. 其他加载timezonefinder库的方式？ #

你原来的代码里，每次调用UDF都在函数内部导入并初始化TimezoneFinder实例，这不仅效率低，还会导致numba在分布式环境下的缓存路径问题。推荐你调整代码结构，让每个Executor进程只初始化一次TimezoneFinder：

优化后的普通UDF写法 #

from timezonefinder import TimezoneFinder
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

# 在模块级别初始化，每个Executor进程启动时只会执行一次
tzf = TimezoneFinder()

def get_timezone(longitude, latitude):
    # 先处理空值，避免报错
    if longitude is None or latitude is None:
        return None
    return tzf.timezone_at(lng=longitude, lat=latitude)

udf_timezone = udf(get_timezone, StringType())

# 后续使用逻辑不变
df = sqlContext.read.parquet(INPUT)
df.withColumn("local_timezone", udf_timezone(df.longitude, df.latitude))\
  .write.parquet(OUTPUT)

更高效的Pandas UDF写法 #

如果你的数据量较大，推荐用Pandas UDF按批次处理，性能会更好：

from timezonefinder import TimezoneFinder
from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import StringType
import pandas as pd

tzf = TimezoneFinder()

@pandas_udf(StringType())
def get_timezone_udf(longitudes: pd.Series, latitudes: pd.Series) -> pd.Series:
    def map_tz(lng, lat):
        if pd.isna(lng) or pd.isna(lat):
            return None
        return tzf.timezone_at(lng=lng, lat=lat)
    return pd.Series(map(map_tz, longitudes, latitudes))

# 使用方式
df.withColumn("local_timezone", get_timezone_udf(df.longitude, df.latitude))\
  .write.parquet(OUTPUT)

这样调整后，TimezoneFinder实例会在每个Executor进程启动时初始化一次，避免了重复创建实例的开销，同时也能让numba在Executor进程的环境下正常编译缓存。

3. 是否可禁用numba的缓存功能？ #

当然可以，有几种方式可以实现：

• 全局禁用numba缓存：在导入timezonefinder之前，设置numba的配置：

  import numba
  # 禁用numba缓存
  numba.config.DISABLE_CACHE = True
  # 再导入timezonefinder
  from timezonefinder import TimezoneFinder
  

  这种方式会全局关闭numba的缓存功能，解决找不到缓存路径的RuntimeError，但代价是第一次运行时JIT编译的时间会变长，后续每次运行都需要重新编译，性能会有一定下降。

• 禁用单个函数的缓存：如果你能修改timezonefinder库的代码，可以找到helpers_numba.py里的numba装饰器，把cache=True改成cache=False，但这种方式需要修改第三方库的源码，不推荐在生产环境使用。

• 通过环境变量禁用：在提交Spark任务时，设置环境变量NUMBA_DISABLE_JIT=1，但这会完全禁用numba的JIT编译，会大幅降低timezonefinder的性能，仅作为临时调试方案。

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