将Spark DataFrame列传递到GeoHash功能-Pyspark.无法将列转换为bool:

Question

import pygeohash as pgh
pgh.encode(45,55)
'tpzpgxczbzur'

上述步骤运作良好。下面我正在尝试创建一个数据框架：

l = [(45,25),(75,22),(85,20),(89,26)]
rdd = sc.parallelize(l)
geoCords = rdd.map(lambda x: Row(lat=x[0], long=int(x[1])))
geoCordsSchema = sqlContext.createDataFrame(geoCords)
geoCordsSchema.show()
+---+----+
|lat|long|
+---+----+
| 45|  25|
| 75|  22|
| 85|  20|
| 89|  26|
+---+----+

这成功创建了Spark DataFrame。现在，我正在使用Pygeohash编码，并按以下丢弃错误：

pgh.encode(geoCordsSchema.lat, geoCordsSchema.long, precision = 7)
Traceback (most recent call last):
   File "<stdin>", line 1, in <module>
   File "/Library/Python/2.7/site-packages/pygeohash/geohash.py", line 96, in encode
   if longitude > mid:
   File "/usr/local/spark/python/pyspark/sql/column.py", line 427, in __nonzero__
   raise ValueError("Cannot convert column into bool: please use '&' for 'and', '|' for 'or', "
ValueError: Cannot convert column into bool: please use '&' for 'and', '|' for 'or', '~' for 'not' when building DataFrame boolean expressions.

Answer 1

您不能直接在某些函数上使用列来转换它。您可以使用UDF实现它，

from pyspark.sql import function as F
udf1 = F.udf(lambda x,y: pgh.encode(x,y,precision=7))
geoCordsSchema.select('lat','long',udf1('lat','long').alias('encodedVal')).show()
+---+----+-----------+
|lat|long|encodedeVal|
+---+----+-----------+
| 45|  25|    sxczbzu|
| 75|  22|    umrdst7|
| 85|  20|    urn5x1g|
| 89|  26|    uxf6r9u|
+---+----+-----------+