Packages

package root

Definition Classes: root

package com

Definition Classes: root

package precisely

Definition Classes: com

package bigdata

Definition Classes: precisely

package addressing

Definition Classes: bigdata

package spark

Definition Classes: addressing

package api

This section provides code snippets to simplify usage of Geo Addressing SDK for Big Data.

Geocode Operation Example

import org.apache.spark.sql.functions._
import com.pb.downloadmanager.api._
import com.pb.downloadmanager.api.downloaders._
import com.pb.downloadmanager.api.downloaders.hadoop._
import com.precisely.addressing.v1.model._
import com.precisely.bigdata.addressing.spark.api._
import com.precisely.addressing.v1._

// 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.
val downloadManager = new DownloadManagerBuilder("./downloads")
  .addDownloader(new S3Downloader(spark.sparkContext.hadoopConfiguration))
  .addDownloader(new GoogleDownloader(spark.sparkContext.hadoopConfiguration))
  .addDownloader(new HDFSDownloader(spark.sparkContext.hadoopConfiguration))
  .addDownloader(new LocalFilePassthroughDownloader())
  .build()

// Remove the fields which are not required.
val outputFields = Seq(
    "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.
val geocodeUdf = new AddressingBuilder()
    .withResourcesLocation("s3a:///addressingDistribution/resources")
    .withDownloadManager(downloadManager)
    .withDataLocations("s3a://com.precisely.data/path_to_spd1","s3a://path_to_spd2")
    .withExtractionLocation("./extracted")
    .udfBuilder()
    .withPreferences(new PreferencesBuilder().withReturnAllInfo(true).build())
    .withOutputFields(outputFields: _*)
    .withErrorField("error")
    .withResultAsJSON("jsonOutput")
    .forGeocode()

val geocodeDF = input.withColumn("result", geocodeUdf(map(
  lit("addressLines[0]"), col("address"),
  lit("country"), col("country")
))).select("*", "result.*").drop(colName = "result")

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

Custom Executor Example for Multipass Geocoding:

import com.precisely.bigdata.addressing.spark.api._
import org.apache.spark.sql.functions._

// Refer to Geocode Operation Example for Full Code.
// Use .forCustomExecutor() API
val customUdf = new AddressingBuilder()
 ...
 .forCustomExecutor(new AddressingExecutor {
      override def execute(input: RequestInput, preferences: Option[Preferences], addressing: Addressing): Response = {

          // First Pass
          val response = addressing.geocode(input.requestAddress(), preferences.orNull)
          // Validate the results.
          if (results.get(0).getCustomFields.get("PRECISION_CODE").toUpperCase().startsWith("S8"))
                return response

          // Second Pass
          // Convert MultiLine Input Address to SingleLine
          val singleLineAddress: RequestAddress = new RequestAddress()
          val filterAddress: String = Seq(inputMultiLineAddress.getStreet,
              inputMultiLineAddress.getCity,
              inputMultiLineAddress.getAdmin1,
              inputMultiLineAddress.getPostalCode
          ).filter(isNotNull).mkString(" ")
          singleLineAddress.setAddressLines(java.util.Collections.singletonList(filterAddress: String))
          singleLineAddress.setCountry(inputMultiLineAddress.getCountry)

          // Geocode with SingleLineAddress and CustomPreferences
          val singleLineResponse: Response = addressing.geocode(singleLineAddress,
                        new PreferencesBuilder().withReturnAllInfo(true).build())

          // Adding a Custom Field
          singleLineResponse.getResults.get(0).getCustomFields.put("TOTAL_MATCHES", singleLineResponse.getResults.size().toString)
          return singleLineResponse
      }
      override def execute(lookupType: LookupType, preferences: Option[Preferences], addressing: Addressing, keyValues: KeyValue*): Response = ???
      override def execute(x: Double, y: Double, country: String, preferences: Option[Preferences], addressing: Addressing): Response = ???
  })

val geocodeDF = input.withColumn("result", customUdf(map(
  lit("addressLines[0]"), col("address"),
  lit("country"), col("country")
))).select("*", "result.*").drop(colName = "result")

Verify Operation Example:

import com.precisely.bigdata.addressing.spark.api.AddressingBuilder
import org.apache.spark.sql.functions._

// Refer to Geocode Operation Example for Full Code.
// Use .forVerify() API
val verifyUdf = new AddressingBuilder()
  ...
  .forVerify()

val verifyDf = input.withColumn("result", verifyUdf(map(
  lit("addressLines[0]"), col("address"),
  lit("country"), col("country")
))).select("*", "result.*").drop(colName = "result")

Reverse Geocode Operation Example:

import com.precisely.bigdata.addressing.spark.api.AddressingBuilder
import org.apache.spark.sql.functions._

// Refer to Geocode Operation Example for Full Code.
// Use .forReverseGeocode() API
val reverseGeocodeUdf = new AddressingBuilder()
  ...
  .forReverseGeocode()

val reverseGeocodeDf = input.withColumn("result", reverseGeocodeUdf(
  col("x"), col("y"), col("country")
)).select("*", "result.*").drop(colName = "result")

Lookup Operation Example:

import com.precisely.bigdata.addressing.spark.api.AddressingBuilder
import org.apache.spark.sql.functions._

// Refer to Geocode Operation Example for Full Code.
// Use .forLookup() API
val lookupUdf = new AddressingBuilder()
  ...
  .forLookup()

val lookUpUdf = input.withColumn("result", lookupUdf(
  col("keyType"), col("key"), col("country")
)).select("*", "result.*").drop(colName = "result")

Geocode Operation with SQL Queries:

import com.precisely.bigdata.addressing.spark.api.AddressingBuilder
import org.apache.spark.sql.functions._

// Refer to Geocode Operation Example for Full Code.
// Use .forGeocode(<udf-name>) API
val geocodeUdf = new AddressingBuilder()
  ...
  .forGeocode("PreciselyGeocode")

inputDF.createOrReplaceTempView("inputTable")
val geocodeDF = spark.sql("select *, PreciselyGeocode(map('addressLines[0]', address, 'country', country)) as result from inputTable")

com.precisely.bigdata.addressing

spark

package spark

Package Members

package api

This section provides code snippets to simplify usage of Geo Addressing SDK for Big Data.

Geocode Operation Example

import org.apache.spark.sql.functions._
import com.pb.downloadmanager.api._
import com.pb.downloadmanager.api.downloaders._
import com.pb.downloadmanager.api.downloaders.hadoop._
import com.precisely.addressing.v1.model._
import com.precisely.bigdata.addressing.spark.api._
import com.precisely.addressing.v1._

// 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.
val downloadManager = new DownloadManagerBuilder("./downloads")
  .addDownloader(new S3Downloader(spark.sparkContext.hadoopConfiguration))
  .addDownloader(new GoogleDownloader(spark.sparkContext.hadoopConfiguration))
  .addDownloader(new HDFSDownloader(spark.sparkContext.hadoopConfiguration))
  .addDownloader(new LocalFilePassthroughDownloader())
  .build()

// Remove the fields which are not required.
val outputFields = Seq(
    "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.
val geocodeUdf = new AddressingBuilder()
    .withResourcesLocation("s3a:///addressingDistribution/resources")
    .withDownloadManager(downloadManager)
    .withDataLocations("s3a://com.precisely.data/path_to_spd1","s3a://path_to_spd2")
    .withExtractionLocation("./extracted")
    .udfBuilder()
    .withPreferences(new PreferencesBuilder().withReturnAllInfo(true).build())
    .withOutputFields(outputFields: _*)
    .withErrorField("error")
    .withResultAsJSON("jsonOutput")
    .forGeocode()

val geocodeDF = input.withColumn("result", geocodeUdf(map(
  lit("addressLines[0]"), col("address"),
  lit("country"), col("country")
))).select("*", "result.*").drop(colName = "result")

Custom Executor Example for Multipass Geocoding:

import com.precisely.bigdata.addressing.spark.api._
import org.apache.spark.sql.functions._

// Refer to Geocode Operation Example for Full Code.
// Use .forCustomExecutor() API
val customUdf = new AddressingBuilder()
 ...
 .forCustomExecutor(new AddressingExecutor {
      override def execute(input: RequestInput, preferences: Option[Preferences], addressing: Addressing): Response = {

          // First Pass
          val response = addressing.geocode(input.requestAddress(), preferences.orNull)
          // Validate the results.
          if (results.get(0).getCustomFields.get("PRECISION_CODE").toUpperCase().startsWith("S8"))
                return response

          // Second Pass
          // Convert MultiLine Input Address to SingleLine
          val singleLineAddress: RequestAddress = new RequestAddress()
          val filterAddress: String = Seq(inputMultiLineAddress.getStreet,
              inputMultiLineAddress.getCity,
              inputMultiLineAddress.getAdmin1,
              inputMultiLineAddress.getPostalCode
          ).filter(isNotNull).mkString(" ")
          singleLineAddress.setAddressLines(java.util.Collections.singletonList(filterAddress: String))
          singleLineAddress.setCountry(inputMultiLineAddress.getCountry)

          // Geocode with SingleLineAddress and CustomPreferences
          val singleLineResponse: Response = addressing.geocode(singleLineAddress,
                        new PreferencesBuilder().withReturnAllInfo(true).build())

          // Adding a Custom Field
          singleLineResponse.getResults.get(0).getCustomFields.put("TOTAL_MATCHES", singleLineResponse.getResults.size().toString)
          return singleLineResponse
      }
      override def execute(lookupType: LookupType, preferences: Option[Preferences], addressing: Addressing, keyValues: KeyValue*): Response = ???
      override def execute(x: Double, y: Double, country: String, preferences: Option[Preferences], addressing: Addressing): Response = ???
  })

val geocodeDF = input.withColumn("result", customUdf(map(
  lit("addressLines[0]"), col("address"),
  lit("country"), col("country")
))).select("*", "result.*").drop(colName = "result")

Verify Operation Example:

import com.precisely.bigdata.addressing.spark.api.AddressingBuilder
import org.apache.spark.sql.functions._

// Refer to Geocode Operation Example for Full Code.
// Use .forVerify() API
val verifyUdf = new AddressingBuilder()
  ...
  .forVerify()

val verifyDf = input.withColumn("result", verifyUdf(map(
  lit("addressLines[0]"), col("address"),
  lit("country"), col("country")
))).select("*", "result.*").drop(colName = "result")

Reverse Geocode Operation Example:

import com.precisely.bigdata.addressing.spark.api.AddressingBuilder
import org.apache.spark.sql.functions._

// Refer to Geocode Operation Example for Full Code.
// Use .forReverseGeocode() API
val reverseGeocodeUdf = new AddressingBuilder()
  ...
  .forReverseGeocode()

val reverseGeocodeDf = input.withColumn("result", reverseGeocodeUdf(
  col("x"), col("y"), col("country")
)).select("*", "result.*").drop(colName = "result")

Lookup Operation Example:

import com.precisely.bigdata.addressing.spark.api.AddressingBuilder
import org.apache.spark.sql.functions._

// Refer to Geocode Operation Example for Full Code.
// Use .forLookup() API
val lookupUdf = new AddressingBuilder()
  ...
  .forLookup()

val lookUpUdf = input.withColumn("result", lookupUdf(
  col("keyType"), col("key"), col("country")
)).select("*", "result.*").drop(colName = "result")

Geocode Operation with SQL Queries:

import com.precisely.bigdata.addressing.spark.api.AddressingBuilder
import org.apache.spark.sql.functions._

// Refer to Geocode Operation Example for Full Code.
// Use .forGeocode(<udf-name>) API
val geocodeUdf = new AddressingBuilder()
  ...
  .forGeocode("PreciselyGeocode")

inputDF.createOrReplaceTempView("inputTable")
val geocodeDF = spark.sql("select *, PreciselyGeocode(map('addressLines[0]', address, 'country', country)) as result from inputTable")

Packages

spark

package spark

Package Members

Ungrouped

spark