GeoNames Cities Pipeline.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# GeoNames Cities processing"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 1. Setup the local Spark instance.\n",
    "\n",
    "In this first step, we load and instantiate the different global Spark contexts: `SparkContext` and `SQLContext`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init(\"/opt/spark\")\n",
    "\n",
    "# Load the global SparkContext for this Spark application.\n",
    "from pyspark import SparkContext, SparkConf\n",
    "conf = SparkConf()\n",
    "conf.set(\"spark.app.name\", \"SparkAlgoliaPipeline\")\n",
    "conf.set(\"spark.master\", \"local[30]\")\n",
    "conf.set(\"spark.ui.port\", \"4040\")\n",
    "\n",
    "SparkContext.setSystemProperty('spark.executor.memory', '16g')\n",
    "SparkContext.setSystemProperty('spark.driver.memory', '8g')\n",
    "SparkContext.setSystemProperty('spark.driver.maxResultSize', '8g')\n",
    "\n",
    "sc = SparkContext(conf=conf)\n",
    "\n",
    "# Load the global Spark SQLContext for this Spark application.\n",
    "from pyspark.sql import SQLContext\n",
    "sqlContext = SQLContext(sc)\n",
    "\n",
    "# Load the Spark packages that we will use throughout this notebook.\n",
    "from pyspark.sql.types import ArrayType, StructType, StructField, DoubleType, IntegerType, StringType, BooleanType\n",
    "from pyspark.sql.functions import col, udf, collect_list, collect_set, size, levenshtein, abs, length, ceil\n",
    "\n",
    "# Load the tools we need for the string similarity computation\n",
    "from collections import Counter\n",
    "from sklearn.feature_extraction.text import CountVectorizer\n",
    "from sklearn.metrics.pairwise import cosine_similarity"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2. Load the different GeoNames datasets into Spark DataFrame instances.\n",
    "\n",
    " - The GeoNames dataset (DataFrame variable `df_gn`); the specifications for this dataset are available [here](http://download.geonames.org/export/dump/readme.txt).\n",
    " - The Postal Codes dataset (DataFrame variable `df_pc`); the specifications for this dataset are available [here](http://download.geonames.org/export/zip/readme.txt).\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define the local path of the datasets.\n",
    "DATA_PATH = \"path/to/geonames/\"\n",
    "POSTAL_CODES_FILE = DATA_PATH + \"postcodes-allCountries.txt\"\n",
    "GEONAMES_FILE = DATA_PATH + \"geonames-allCountries.txt\"\n",
    "\n",
    "\n",
    "# Define the DataFrame schema for each dataset.\n",
    "schema_postal_codes = StructType([\n",
    "    StructField(\"CC\", StringType()),\n",
    "    StructField(\"PC\", StringType()),\n",
    "    StructField(\"NAME\", StringType()),\n",
    "    StructField(\"AN1\", StringType()),\n",
    "    StructField(\"AC1\", StringType()),\n",
    "    StructField(\"AN2\", StringType()),\n",
    "    StructField(\"AC2\", StringType()),\n",
    "    StructField(\"AN3\", StringType()),\n",
    "    StructField(\"AC3\", StringType()),\n",
    "    StructField(\"LAT\", DoubleType()),\n",
    "    StructField(\"LON\", DoubleType()),\n",
    "    StructField(\"ACC\", IntegerType())\n",
    "])\n",
    "\n",
    "schema_geonames = StructType([\n",
    "    StructField(\"GNID\", StringType()),\n",
    "    StructField(\"NAME\", StringType()),\n",
    "    StructField(\"NAME_ASCII\", StringType()),\n",
    "    StructField(\"ALT_NAMES\", StringType()),\n",
    "    StructField(\"LAT\", DoubleType()),\n",
    "    StructField(\"LON\", DoubleType()),\n",
    "    StructField(\"F_CLASS\", StringType()),\n",
    "    StructField(\"F_CODE\", StringType()),\n",
    "    StructField(\"CC\", StringType()),\n",
    "    StructField(\"CC2\", StringType()),\n",
    "    StructField(\"AC1\", StringType()),\n",
    "    StructField(\"AC2\", StringType()),\n",
    "    StructField(\"AC3\", StringType()),\n",
    "    StructField(\"AC4\", StringType()),\n",
    "    StructField(\"POP\", IntegerType()),\n",
    "    StructField(\"ALT\", StringType()),\n",
    "    StructField(\"DEM\", StringType()),\n",
    "    StructField(\"TZ\", StringType()),\n",
    "    StructField(\"DATE\", StringType())\n",
    "])\n",
    "\n",
    "# Load both datasets in their respective DataFrame instances.\n",
    "df_pc = sqlContext.read.format(\"com.databricks.spark.csv\").schema(schema_postal_codes).option(\"header\", \"false\").option(\"mode\", \"DROPMALFORMED\").option(\"delimiter\", '\\t').load(POSTAL_CODES_FILE)\n",
    "df_gn = sqlContext.read.format(\"com.databricks.spark.csv\").schema(schema_geonames).option(\"header\", \"false\").option(\"mode\", \"DROPMALFORMED\").option(\"delimiter\", '\\t').load(GEONAMES_FILE)\n",
    "\n",
    "# Give aliases to the DataFrames we just loaded\n",
    "df_pc = df_pc.alias(\"pc\")\n",
    "df_gn = df_gn.alias(\"gn\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3. Filter the Postal Codes dataset according to the country-specific postal code formats.\n",
    "\n",
    "The country-specific postal codes format is given by this crowdsourced Wikipedia page: https://en.wikipedia.org/wiki/List_of_postal_codes.  \n",
    "For this step, we parse the file that we have made from the formats detailed in the Wikipedia page and load each postal code format string as a regex pattern instance."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import re\n",
    "\n",
    "# Define the local path of the csv file containing the postal code formats.\n",
    "POSTAL_CODES_FORMAT_FILE = DATA_PATH + \"postcodes.csv\"\n",
    "\n",
    "# Instanciate the list of postal code formats as a global variable.\n",
    "country_postal_codes = {}\n",
    "with open(POSTAL_CODES_FORMAT_FILE, 'r') as f:\n",
    "    for line in f:\n",
    "        fields = line.split('\\t')\n",
    "        cc = fields[0]\n",
    "        pattern = None\n",
    "        if fields[1] != '':\n",
    "            pattern = re.compile(fields[1].strip())\n",
    "        \n",
    "        country_postal_codes[cc] = pattern\n",
    "\n",
    "print(\"Loaded %s postal code formats.\" % len(country_postal_codes))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define a Python function to check the validity of a postal code given the country code.\n",
    "def is_valid_postal_code(cc, postal_code):\n",
    "    \"\"\"Returns True if the postal code postal_code is valid \n",
    "       for the given country code cc, False otherwise.\"\"\"\n",
    "    if cc not in country_postal_codes:\n",
    "        return False\n",
    "    \n",
    "    m = country_postal_codes[cc].match(postal_code)\n",
    "    return m is not None\n",
    "    \n",
    "    \n",
    "# Define a Spark user-defined function to check the validity of a postal code given the country code.\n",
    "is_valid_postal_code_udf = udf(lambda x,y: is_valid_postal_code(x,y), BooleanType())\n",
    "\n",
    "# Filter the Postal Codes DataFrame to remove the invalid postal codes.\n",
    "df_pc_filtered = df_pc.filter(is_valid_postal_code_udf(col(\"pc.CC\"), col(\"pc.PC\")))\n",
    "\n",
    "# Print to check that it has filtered.\n",
    "print(\"Number of postal codes features (not filtered): %s.\" % df_pc.count())\n",
    "print(\"Number of postal codes features (filtered): %s.\" % df_pc_filtered.count())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 4. Filter the cities from the GeoNames dataset.\n",
    "\n",
    "The GeoNames features are associated to class and subclass codes. The feature class and subclass codes are defined in the document at this link: https://www.geonames.org/export/codes.html. The list of feature class codes is as follows:\n",
    " - `A`: country, state, region,...\n",
    " - `H`: stream, lake, ...\n",
    " - `L`: parks, area, ...\n",
    " - `P`: city, village, ...\n",
    " - `R`: road, railroad \n",
    " - `S`: spot, building, farm\n",
    " - `T`: mountain, hill, rock, ... \n",
    " - `U`: undersea\n",
    " - `V`: forest, heath, ...\n",
    "\n",
    "The feature class that is of interest for this project is the feature class of code `P` that corresponds to populated places such as cities and villages. In the following, we filter the GeoNames dataset to remove the features that are not of class code `P` and only keep the cities."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Filter the cities (feature class code P) from the GeoNames DataFrame.\n",
    "df_gn_cities = df_gn.filter(col(\"gn.F_CLASS\")=='P')\n",
    "\n",
    "print(\"Number of GeoNames features (not filtered): %s.\" % df_gn.count())\n",
    "print(\"Number of GeoNames features (filtered): %s.\" % df_gn_cities.count())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 5. Join the GeoNames dataset to the Postal Codes dataset.\n",
    "\n",
    "In this step, we join the GeoNames dataset to the Postal Codes dataset based on the longitudes and latitudes of the features. A precision of one decimal place in the coordinates corresponds to an approximate length of 7.871 kilometers. Therefore, comparing coordinates at this precision level would be precise enough to absorb the difference between the coordinates in both datasets, while avoiding overlap on multiple cities. \n",
    "\n",
    "#### 6. Aggregate the joined dataset.\n",
    "\n",
    "In this last step, we aggregate the joined DataFrame by grouping the Postal Code name features. The identifiers and names of the GeoNames features are then aggregated into two separate lists for a given matching Postal Codes name feature. We select the most likely GeoNames feature candidate using the cosine distance between the Postal Code feature name and the GeoNames feature name. The [cosine similarity](https://en.wikipedia.org/wiki/Cosine_similarity) performs a loose match between two strings. In particular, it removes the special characters, tokenizes the string into words, and then compares the words shared between two strings. If the similarity between the name in the GeoNames dataset and the one in the Postal Codes dataset is more than 20%, the two names will be matched.\n",
    "\n",
    "The other features that match the name features will be joined to the aggregated dataframe using the GeoNames identifier. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "import os\n",
    "import time\n",
    "\n",
    "# Define some constants for this part of the pipeline.\n",
    "COORDINATES_PRECISION = 0.01\n",
    "STRING_MATCHING_THRESHOLD = 0.2\n",
    "TEMP_CSV_FOLDER = \"temp_output\"\n",
    "CSV_FOLDER = \"geonames_output\"\n",
    "\n",
    "# Define some functions that we will use in the user-defined function below.\n",
    "def get_best_match_idx(ref, strs): \n",
    "    \"\"\"Get the index of the best matching string in strs against ref.\"\"\"\n",
    "    vectors = [t for t in get_vectors([ref]+strs)]\n",
    "    sim = cosine_similarity(vectors)\n",
    "    match_idx = sim[0,1:].argmax(axis=0)\n",
    "    if sim[0,1:][match_idx] >= STRING_MATCHING_THRESHOLD:\n",
    "        return int(match_idx)\n",
    "    else:\n",
    "        return -1\n",
    "    \n",
    "def get_vectors(strs):\n",
    "    \"\"\"Vectorize the word tokens normalized in lower case contained in strings of list strs.\"\"\"\n",
    "    text = [t.lower() for t in strs if t is not None]\n",
    "    vectorizer = CountVectorizer(text)\n",
    "    vectorizer.fit(text)\n",
    "    return vectorizer.transform(text).toarray()\n",
    "\n",
    "\n",
    "# Define the join conditions.\n",
    "cond1 = [col(\"pc.CC\") == col(\"gn.CC\")]\n",
    "cond2 = [abs(col(\"pc.LON\") - col(\"gn.LON\")) <= COORDINATES_PRECISION, \n",
    "         abs(col(\"pc.LAT\") - col(\"gn.LAT\")) <= COORDINATES_PRECISION]\n",
    "\n",
    "# Define the fields to select after the first join.\n",
    "sel_join = [col(\"gn.GNID\"), col(\"gn.CC\"), col(\"pc.AC1\"), col(\"pc.AC2\"), col(\"pc.AC3\"), \n",
    "            col(\"pc.AN1\"), col(\"pc.AN2\"), col(\"pc.AN3\"), col(\"gn.NAME\").alias(\"NAME_gn\"), \n",
    "            col(\"pc.NAME\").alias(\"NAME_pc\"), col('pc.PC'), col('pc.LAT'), col(\"pc.LON\")]\n",
    "\n",
    "# Define the fields to select after the aggregation.\n",
    "sel_agg = [col(\"jn.GNID\"), col(\"jn.NAME_gn\"), col(\"jn.NAME_pc\"), col(\"jn.PC\"),\n",
    "           col(\"jn.AC1\"), col(\"jn.AC2\"), col(\"jn.AC3\"), \n",
    "           col(\"jn.AN1\"), col(\"jn.AN2\"), col(\"jn.AN3\")]\n",
    "\n",
    "# Define the final fiels to select after the final join.\n",
    "sel_final = [col(\"gn.GNID\"), col(\"gn.CC\"), col(\"agg.AC1\"), col(\"agg.AC2\"), col(\"gn.LAT\"), col(\"gn.LON\"),\n",
    "             col(\"agg.AC3\"), col(\"agg.AN1\"), col(\"agg.AN2\"), col(\"agg.AN3\"), col(\"agg.NAME_pc\"), \n",
    "             col(\"gn.NAME\"), col(\"agg.PC_set\"), col(\"gn.ALT_NAMES\"), col(\"gn.POP\")]\n",
    "\n",
    "# User-defined function to transform a list of elements into a set.\n",
    "to_set_udf = udf(lambda x: list(set(x)), ArrayType(StringType()))\n",
    "\n",
    "# User-defined function to select the most similar GeoNames name with the given postal code feature name.\n",
    "best_match_idx_udf = udf(lambda x,y: get_best_match_idx(x,y), IntegerType())\n",
    "\n",
    "# User-defined function to get the value of the element in a list at a given index.\n",
    "get_list_index_udf = udf(lambda l,i: l[i] if i >= 0 else '', StringType())\n",
    "\n",
    "# User-defined function to transform an array of strings into a string (eq. str.join).\n",
    "array_to_string_udf = udf(lambda x: \",\".join(x))\n",
    "\n",
    "# Get the countries present in the Postal Codes dataframe\n",
    "countries = df_pc_filtered.groupby('CC').count().toPandas().set_index('CC').T.to_dict()\n",
    "\n",
    "print(\"Computing the join...\")\n",
    "for cc, count in countries.items():            \n",
    "    print(\"[.] %s: Processing...\" % cc)\n",
    "    start_time = time.time()\n",
    "    \n",
    "    # Filter both dataframes by their country code.\n",
    "    df_pc_filtered_cc = df_pc_filtered.filter((col(\"pc.CC\")==cc) & (col(\"pc.NAME\").isNotNull()))\n",
    "    df_gn_cities_cc = df_gn_cities.filter((col(\"gn.CC\")==cc) & (col(\"gn.NAME\").isNotNull()))\n",
    "\n",
    "    # Perform the inner join operation on the fitered dataframes.\n",
    "    df_joined_cc = df_pc_filtered_cc.join(df_gn_cities_cc, cond1+cond2, \"inner\").select(sel_join)\n",
    "        \n",
    "    # Aggregate the names features of the Postal Codes dataset in the joined DataFrame.\n",
    "    # Reduce the GeoName features with the best matching name\n",
    "    df_joined_cc = df_joined_cc.alias(\"jn\")\n",
    "    df_agg_cc = df_joined_cc.select(sel_agg)\\\n",
    "    .groupBy(\"NAME_pc\", \"AC1\", \"AC2\", \"AC3\", \"AN1\", \"AN2\", \"AN3\").agg(\n",
    "        collect_list(\"GNID\").alias(\"GNID_list\"),\n",
    "        collect_list(\"NAME_gn\").alias(\"NAME_gn_list\"),\n",
    "        collect_set(\"PC\").alias(\"PC_set\"))\\\n",
    "    .withColumn(\"match_idx\", best_match_idx_udf(col(\"NAME_pc\"), col(\"NAME_gn_list\")))\\\n",
    "    .withColumn(\"GNID\", get_list_index_udf(col(\"GNID_list\"), col(\"match_idx\")))\n",
    "    \n",
    "    df_agg_cc = df_agg_cc.alias(\"agg\")\n",
    "    df_agg_filtered_cc = df_agg_cc.filter(col(\"agg.GNID\").isNotNull())\n",
    "    \n",
    "    # Join the aggregated dataframe with the attributes of the GeoNames dataframe.\n",
    "    df_agg_joined_cc = df_agg_filtered_cc.join(df_gn_cities_cc, col(\"gn.GNID\") == col(\"agg.GNID\"), \"inner\").select(sel_final)\n",
    "    \n",
    "    # Transform the postal code set (array) into a string compatible with csv write.\n",
    "    try:\n",
    "        df_agg_joined_cc = df_agg_joined_cc.alias(\"agg_jn\")\n",
    "        df_agg_joined_cc.withColumn('PCs', array_to_string_udf(col(\"agg_jn.PC_set\")))\\\n",
    "        .drop(\"PC_set\").write.csv(\"%s\" % TEMP_CSV_FOLDER)\n",
    "\n",
    "        print(\"[x] %s: Completed in %.2f seconds.\" % (cc, time.time() - start_time))   \n",
    "    except Exception as e:\n",
    "        print(\"Exception when dumping the file in a csv: %s\" % e)\n",
    "        \n",
    "    # Put all the csv files dumped in the temporary folder into a single csv file \n",
    "    # and remove the temporary folder.\n",
    "    os.system(\"cat %s/p* > %s/%s.csv; rm -rf %s\" % (TEMP_CSV_FOLDER, CSV_FOLDER, cc, TEMP_CSV_FOLDER))     \n",
    "    \n",
    "    # Release the dataframes.\n",
    "    try:\n",
    "        del df_agg_joined_cc\n",
    "        del df_agg_filtered_cc\n",
    "        del df_agg_cc\n",
    "        del df_joined_cc\n",
    "        del df_gn_cities_cc\n",
    "        del df_pc_filtered_cc\n",
    "    except KeyError:\n",
    "            pass\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 7. Prepare and send the data to Algolia\n",
    "In this last part, we prepare and then send the data to Algolia."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from algoliasearch import algoliasearch\n",
    "\n",
    "client = algoliasearch.Client(\"N3CD4LLDJP\", '7b36e4dc1ad98f44e8868c1b1aa184d0')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import csv\n",
    "import glob\n",
    "\n",
    "# Batch add to the Algolia index.\n",
    "# Doc: https://www.algolia.com/doc/api-reference/api-methods/batch/\n",
    "# Search doc: https://www.algolia.com/doc/guides/searching/geo-search/#getting-geo-search-info-with-rankinginfo\n",
    "\n",
    "INDEX_NAME = \"geonames\"\n",
    "\n",
    "cc_data = []\n",
    "count = 0\n",
    "for file in glob.glob(CSV_FOLDER+\"/*.csv\"):\n",
    "    # Prepare the data.        \n",
    "    with open(file, 'r') as f:\n",
    "        csvreader = csv.reader(f, delimiter=',', quotechar='\"')\n",
    "        for row in csvreader:\n",
    "            d = {\n",
    "                \"action\": \"addObject\",\n",
    "                \"indexName\": INDEX_NAME,\n",
    "                \"body\": {\n",
    "                    \"geonameid\": row[0],\n",
    "                    \"cc\": row[1],\n",
    "                    \"name\": row[10],\n",
    "                    \"altnames\": row[12].split(','),\n",
    "                    \"admin1\": row[7],\n",
    "                    \"admin2\": row[8],\n",
    "                    \"admin3\": row[9],\n",
    "                    \"postalCodes\": row[14].split(',')[:100],\n",
    "                    \"population\": int(row[13]),\n",
    "                    \"_geoloc\": {\n",
    "                        \"lat\": float(row[4]),\n",
    "                        \"lng\": float(row[5])\n",
    "                    }\n",
    "                }\n",
    "            }\n",
    "            cc_data.append(d)\n",
    "            count += 1\n",
    "\n",
    "    \n",
    "# Batch-send the data.\n",
    "print(\"Sending the data...\")\n",
    "res = client.batch(cc_data)\n",
    "print(\"Data sent with %s records.\" % (count))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python (py36)",
   "language": "python",
   "name": "py36"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}