Geo Addressing API Usage

This section provides code snippets to simplify usage of Geo Addressing APIs.

For more information regarding the Usage Guide or API Docs, follow the below links:

Required Spark Configs:

SparkSession.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
from pyspark.sql import SparkSession

# The `allowUntypedScalaUDF` config is required to set run the legacy UDFs used in the Geo Addressing APIs.
# For Databricks, Instead of spark.jars, add the jar and zip as a Library in the cluster.
spark = (SparkSession.builder
        .appName("Example")
        .config("spark.jars", "hdfs:///addressing_distribution/pyspark/sdk/lib/geo-addressing-bigdata-addressing-sdk-spark<spark_version>-<sdk_version>.jar")
        .getOrCreate())

# You need to add the zip file containing the geo addressing sdk python classes along with the sdk jar so that the classes will be available in the spark context.
# This zip file is available in the `pyspark/sdk/lib/` directory.
spark.sparkContext.addPyFile('hdfs:///addressing_distribution/pyspark/sdk/lib/geo-addressing-bigdata-addressing-sdk-pyspark-<sdk_version>.zip')

Geocode Operation Example:

GeocodeOperation.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
# Importing the Required Classes
from addressing.DownloadManagerBuilder import DownloadManagerBuilder
from addressing.AddressingBuilder import AddressingBuilder
from addressing.HDFSDownloader import HDFSDownloader
from addressing.S3Downloader import S3Downloader
from addressing.GoogleDownloader import GoogleDownloader
from addressing.LocalFilePassthroughDownloader import LocalFilePassthroughDownloader
from addressing.PreferencesBuilder import PreferencesBuilder
from addressing.HadoopConfiguration import HadoopConfiguration
from addressing.UDFExecutor import UDFExecutor

# DownloadManager copies all the remote resources (HDFS, S3, Google Storage, etc.) to the
# provided local path (e.g. ./downloads) of every worker node.
# This happens in runtime during the first call. Subsequent calls skips downloading process.
downloadManager = (DownloadManagerBuilder("./downloads")
    .addDownloader(S3Downloader(HadoopConfiguration().getHadoopConfiguration()).getDownloader())
    .addDownloader(GoogleDownloader(HadoopConfiguration().getHadoopConfiguration()).getDownloader())
    .addDownloader(HDFSDownloader(HadoopConfiguration().getHadoopConfiguration()).getDownloader())
    .addDownloader(LocalFilePassthroughDownloader().getDownloader())
    .build())

# Remove the fields which are not required.
outputFields = [
    "location.feature.geometry.coordinates.x as LON",
    "location.feature.geometry.coordinates.y as LAT",
    "address.formattedAddress as fullAddress",
    "address.city.longName as locality",
    "address.country.isoAlpha2Code as countryshortname",
    "address.country.name as country",
    "address.placeName as placename",
    "address.admin1.shortName as administrativearealevel1",
    "address.admin2.longName as administrativearealevel2",
    "address.suburb as suburb",
    "address.borough as borough",
    "address.postalCode as postalcode",
    "address.postalCodeExt as postalcodesuffix",
    "customFields['PB_KEY'] as sourceplaceid",
    "address.addressNumber as AddressNumber",
    "address.street as Street",
    "address.unit as Unit",
    "address.unitType as UnitType",
    "address.formattedStreetAddress as StreetAddress",
    "address.formattedLocationAddress as LocationAddress",
    "address.neighborhood as neighborhood",
    "address.floor as floor",
    "score",
    "explanation.source['label'] as label",
    "customFields['PRECISION_CODE'] as PrecisionCode",
    "customFields['MATCH_TYPE'] as MatchType",
    "customFields['LOC_CODE'] as LocationCode",
    "customFields['MATCH_CODE'] as MatchCode",
    "customFields['CBSA_NAME'] as CBSAName",
    "customFields['PREF_CITY'] as PREFCITY",
    "customFields['PB_KEY'] as PBKEY",
    "customFields['DATATYPE_NAME'] as DataTypeName"
]

# You can also use paths from HDFS or Databricks Volumes
# (e.g. hdfs:///addressingDistribution/resources or /Volumes/precisely/default/geo-addressing/addressingDistribution/resources/)
# The reference data is extracted at runtime in every worker node's extractionLocation path for the first call.
addressingBuilder = (AddressingBuilder()
    .withDownloadManager(downloadManager)
    .withResourcesLocation("s3a:///addressingDistribution/resources")
    .withDataLocations("s3a://com.precisely.data/path_to_spd1","s3a://path_to_spd2")
    .withExtractionLocation("./extracted")
    .udfBuilder()
    .withPreferences(PreferencesBuilder().withReturnAllInfo(True).build())
    .withOutputFields(*outputFields)
    .withErrorField("error")
    .withResultAsJSON("jsonOutput"))

# Create UDF for Geocode Operation
geocodeUDF = addressingBuilder.forGeocode()

# Execute the UDF
geocodeDf = (inputTable.withColumn("result", (
                            UDFExecutor().apply(geocodeUDF, create_map(
                                lit("addressLines[0]"), col("address"),
                                lit("country"), col("country"))))
                            ).persist()
                            .select("*", "result.*").drop("result"))

NOTE: If addressing.yaml is present at the worker node, dataLocations and extractionLocation will be preferred from addressing.yaml over code. In such case where addressing.yaml is present, you should only provide resourcesLocation.

Verify Operation Example:

Verify.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
# Create UDF for Verify Operation
# Refer the addressingBuilder from Geocode Operation Example
verifyUdf = addressingBuilder.forVerify()

# Execute the UDF
verifyDf = (inputTable.withColumn("result", (
                            UDFExecutor().apply(verifyUdf, create_map(
                                lit("addressLines[0]"), col("address"),
                                lit("country"), col("country"))))
                            ).persist()
                            .select("*", "result.*").drop("result"))

Reverse Geocode Operation Example:

ReverseGeocode.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
# Create UDF for Reverse Geocode Operation
# Refer the addressingBuilder from Geocode Operation Example
reverseGeocodeUdf = addressingBuilder.forReverseGeocode()

# Execute the UDF
reverseGeocodeDf = (inputTable.withColumn("result", (
                            UDFExecutor().apply(reverseGeocodeUdf,
                                col("x"), col("y"), col("country"))
                            ).persist()
                            .select("*", "result.*").drop("result"))

Lookup Operation Example:

Lookup.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
# Create UDF for Lookup Operation
# Refer the addressingBuilder from Geocode Operation Example
lookupUdf = addressingBuilder.forLookup()

# Execute the UDF
lookupDf = (inputTable.withColumn("result", (
                            UDFExecutor().apply(lookupUdf,
                                col("keyType"), col("key"), col("country"))
                            ).persist()
                            .select("*", "result.*").drop("result"))

Geocode Operation Using SQL Queries:

GeocodeWithSQL.py
1
2
3
4
5
6
7
# Create UDF for Geocode Operation
# Refer the addressingBuilder from Geocode Operation Example
geocodeUDF = addressingBuilder.forGeocode("PreciselyGeocode")

# Execute the UDF
inputDF.createOrReplaceTempView("inputTable")
geocodeDf = spark.sql("select *, PreciselyGeocode(map('addressLines[0]', address, 'country', country)) as result from inputTable")