python-2.7 - 在 PySpark 中比较地理空间数据的最有效方法

Question

我知道在常规 Python shell 中我们可以使用 pyproj 例如

from pyproj import Geod
nyc_geod = Geod(ellps='WGS84')
lat1, lon1 = (70, -74)
lat2, lon2 = (71, -72)
z1,z2,dist = nyc_geod.inv(lon1,lat1,lon2,lat2)
dist --> 134163.09514455328

然后我可以调用 dist 来给出两点之间的距离。 PySpark 中是否有这样的函数，特别是与比较两个 DataFrame 之间的点有关的函数？谢谢

Answer 1

首先稍微修改一下代码，以便有一个距离函数

from pyproj import Geod
import numpy as np
import pandas as pd

nyc_geod = Geod(ellps='WGS84')
def calc_nyc_geod(lon1,lat1,lon2,lat2):
  _, _ ,dist = nyc_geod.inv(lon1,lat1,lon2,lat2)
  return dist
# show a single point
lat1, lon1 = (70, -74)
lat2, lon2 = (71, -72)
print(calc_nyc_geod(lon1,lat1,lon2,lat2))

你可以制作两个随机的 DataFrame(这里是 numpy 和 pandas，但它们可以来自任何来源)

np.random.seed(1234)
df1 = sqlContext.createDataFrame(pd.DataFrame({'lat': np.random.uniform(lat1, lat2, 10),
                    'lon': np.random.uniform(lon1, lon2, 10),
                   }))
df2 = sqlContext.createDataFrame(pd.DataFrame({'lat': np.random.uniform(lat1, lat2, 10),
                    'lon': np.random.uniform(lon1, lon2, 10),
                   }))

很难从你的问题中看出，但听起来目标是加入两个不同的 DataFrame 并使用距离度量作为标准。对于这种情况，它只是 df1 中有多少个点在 df2 的 35000 米(？)范围内(注意:这是一个完整的外部联接，如果您已经了解有关数据的信息，则可以使用更智能的联接)。 为此，您需要在 SQLContext 对象内部创建一个 UserDefinedFunction，并且由于 DataFrame 是类型化的，因​​此输出也必须类型化。

from pyspark.sql.types import DoubleType
sqlContext.registerFunction("nyc_geod", calc_nyc_geod, DoubleType())
df1.registerTempTable("TDF1")
df2.registerTempTable("TDF2")
sqlContext.sql("""
SELECT COUNT(*) as Overlap FROM TDF1 
JOIN TDF2 
WHERE nyc_geod(TDF1.lon, TDF1.lat, TDF2.lon, TDF2.lat)<35000
""").first()

结果是 Row(Overlap=38) 满足条件的点对数量(共 100 个)。