我有一个CSV文档，正在加载到包含经度和纬度列的SQLContext中。
val sqlContext = new org.apache.spark.sql.SQLContext(sc);
val df = sqlContext.read.format("com.databricks.spark.csv").option("header", "false").option("delimiter","\t").schema(customSchema).load(inputFile);
CSV范例
metro_code, resolved_lat, resolved_lon
602, 40.7201, -73.2001
我正在尝试找出添加新列并计算每一行的GeoHex的最佳方法。使用geohex包可以很容易地散列经纬度。我想我需要运行parallelize方法，或者我已经看到一些将函数传递给withColumn的示例。
用UDF包装必需的功能应该可以解决问题：
import org.apache.spark.sql.functions.udf
import org.geohex.geohex4j.GeoHex
val df = sc.parallelize(Seq(
(Some(602), 40.7201, -73.2001), (None, 5.7805, 139.5703)
)).toDF("metro_code", "resolved_lat", "resolved_lon")
def geoEncode(level: Int) = udf(
(lat: Double, long: Double) => GeoHex.encode(lat, long, level))
df.withColumn("code", geoEncode(9)($"resolved_lat", $"resolved_lon")).show
// +----------+------------+------------+-----------+
// |metro_code|resolved_lat|resolved_lon| code|
// +----------+------------+------------+-----------+
// | 602| 40.7201| -73.2001|PF384076026|
// | null| 5.7805| 139.5703|PR081331784|
// +----------+------------+------------+-----------+
本文收集自互联网，转载请注明来源。
如有侵权，请联系 [email protected] 删除。
我来说两句