在 Pyspark 中添加带有坐标的列

Question

我是 Python/Pyspark 编程的新手，所以如果这是一个基本问题，我很抱歉。

我有带有地址的数据框。但请注意，并非所有的拼写都正确。

+-----+--------+-----------------+----------+
|index| Country|City             |      date|
+-----+--------+-----------------+----------+
|    1|  NL    |Staphorst        |2017-01-01|
|    2|  BE    |Herselt-Blauwberg|2017-01-02|
|    3|  NL    |Oosterend        |2017-01-03|
|    4|  BL    |Hoogezand        |2017-01-04|
+-----+--------+-----------------+----------+

可以使用以下代码创建：

from pyspark.sql import SQLContext

df = spark.createDataFrame(
    [
        (1, "NL", 'Staphorst', '2017-01-01'),
        (2, 'BE', 'Herselt-Blauwberg', '2017-01-02'),
        (3, "NL", 'Oosterend', '2017-01-03'),
        (4, "NL", 'Hoogezand', '2017-01-04')
    ],
    ('index', 'Country', 'City', 'date')
)

我想知道所有地址的坐标。如果城市/地址拼写错误，我只想让它为空。所以期望的结果看起来像：

+-----+--------+-----------------+----------+-----------------+-------------------+
|index| Country|City             |      date| lat             | lon               | 
+-----+--------+-----------------+----------+-----------------+-------------------+
|    1|  NL    |Staphorst        |2017-01-01|6.208129803185402|52.636838749999995 |
|    2|  BE    |Herselt-Blauwberg|2017-01-02|                 |                   |
|    3|  NL    |Oosterend        |2017-01-03|4.8731736        |53.0857473         |
|    4|  BL    |Hoogezand        |2017-01-04|6.7556852        |53.1638192         |
+-----+--------+-----------------+----------+-----------------+-------------------+

我尝试使用 geopy。当我将我的 df 转换为 pandas 数据框时，这种代码会起作用（见下文）。但是，我无法提取纬度和经度数据（因为它并不总是可用），并且在我的实际数据帧上运行时间太长。

from geopy.geocoders import Nominatim
geolocator = Nominatim(user_agent="my_app")

df1 = df.toPandas()
df1['geo'] = df1['City'].apply(geolocator.geocode)
df1

因此，我更喜欢 pyspark 替代方案。我尝试使用一种原始方式来让代码在 pyspark 中运行，但是由于我使用循环并且坐标最终在一个单独的列表中，所以它需要永远运行，我不喜欢它，因为我想把所有东西放在一起.

longitude = []
latitude = []

for i in range(len(df.select("City").collect())):          
      loc = geolocator.geocode(df.select("City").collect()[i].City)
        
      if loc != None:
        latitude.append(loc.latitude)
        longitude.append(loc.longitude)
      else:
        latitude.append("")
        longitude.append("")

虽然我可能应该编写一个 UDF，但我未能让它工作。到目前为止我的尝试：

def geo_loc(str_x):
  loc = geolocator.geocode(str_x)
  if loc != None:
    return loc
  else: 
    return ""

geo_locUDF = udf(lambda z:geo_loc(z), StringType()) 

df.withColumn("geo", geo_locUDF(col('City'))) 

有人可以帮我找到正确的坐标吗？

编辑：如果相关，sprak.version 返回 3.1.0。

Answer 1

不幸的是，有人删除了他们的答案，尽管他们帮助我朝着正确的方向前进。他们给我的代码如下：

import pyspark.sql.functions as f
from geopy.geocoders import Nominatim

nominatim = Nominatim(user_agent="my_app")
df = spark.createDataFrame(
    [
        (1, "NL", 'Staphorst', '2017-01-01'),
        (2, 'BE', 'Herselt-Blauwberg', '2017-01-02'),
        (3, "NL", 'Oosterend', '2017-01-03'),
        (4, "NL", 'Hoogezand', '2017-01-04')
    ],
    ('index', 'Country', 'City', 'date')
)


@f.udf(returnType='STRUCT<`Latitude`: double, `Longitude`: double>')
def search(country, city):
    location = nominatim.geocode(query={'country': country, 'city': city}, timeout=60)
    if location is None:
        return {'Latitude': None, 'Longitude': None}
    else:
        return {'Latitude': location.latitude, 'Longitude': location.longitude}


geocode_df = (df
              .select('Country', 'City')
              .distinct()
              .withColumn('Geo', search('Country', 'City')))

new_df = df.join(geocode_df, on=['Country', 'City'], how='inner')
(new_df
 .select('Country', 'City', 'Geo.Latitude', 'Geo.Longitude')
 .show(truncate=False))

这对那个人有用，但在 Databricks 中我遇到了错误。基本上说has no attribute 'Request sHTTPWithSSLContextAdapter__ ssl_context'

我今天再次在谷歌上搜索了 found a more recent question here on SO，有人指出，提名上下文的创建应该在 UDF 中。可能是菜鸟的错误，但我很高兴我的 UDF 现在运行了！