Merge pull request #18 from dime-worldbank/liberia

Bringing in Liberia Indicators data and updating to include Liberia in pipeline

Author: elysenko

population/COD/cod_subnational_population.py

@@ -27,10 +27,6 @@
 ddf = df[df.ISO_Code=='COD'][['Country', 'Region', 'year', 'population_millions']]
 
 
-# COMMAND ----------
-
-pop.sample(5)
-
 # COMMAND ----------
 
 # Write to indicator_intermediate

population/LBR/lbr_subnational_population.py

@@ -0,0 +1,43 @@
+# Databricks notebook source
+import pandas as pd
+
+# COMMAND ----------
+
+def process_to_indicator_intermediate(country_name:str, country_code:str, adm1_drop:list=[]):
+
+    spark_df = spark.table(f'prd_mega.indicator.global_data_lab_subnational_population')
+    df = spark_df.toPandas()
+
+    ddf = df[df.ISO_Code==country_code.upper()][['Country', 'Region', 'year', 'population_millions']]
+    ddf.columns = ['country_name', 'adm1_name', 'year', 'population']
+    ddf['population'] = ddf.population.map(lambda x: x*1_000_000)
+    ddf['adm1_name'] = ddf['adm1_name'].str.lower()
+    ddf = ddf[ddf.adm1_name!='total']
+    ddf['adm1_name'] = ddf['adm1_name'].str.strip().str.title()
+    ddf = ddf[~ddf['adm1_name'].isin(adm1_drop)]
+
+    pop = ddf.sort_values(['year', 'adm1_name'])
+    pop.country_name = country_name
+    pop['data_source'] = 'Global Data Lab'
+
+    return pop
+
+def write_to_indicator_intermediate(pop:pd.DataFrame, country_code:str):
+
+    database_name = "prd_mega.indicator_intermediate"
+    if not spark.catalog.databaseExists(database_name):
+        print(f"Database '{database_name}' does not exist. Creating the database.")
+        spark.sql(f"CREATE DATABASE {database_name}")
+
+    sdf = spark.createDataFrame(pop)
+    sdf.write.mode("overwrite").saveAsTable(f"{database_name}.{country_code.lower()}_subnational_population")
+
+    return
+
+# COMMAND ----------
+
+country_code = 'LBR'
+country_name = 'Liberia'
+adm1_drop = ['North Central','North Western','Monrovia','South Eastern A','South Eastern B','South Central']
+pop = process_to_indicator_intermediate(country_name, country_code, adm1_drop)
+write_to_indicator_intermediate(pop, country_code)

population/global_data_lab_subnational_population.r

@@ -1,8 +1,5 @@
 # Databricks notebook source
 install.packages("gdldata")
-
-# COMMAND ----------
-
 library(gdldata)
 library(magrittr)
 
@@ -20,7 +17,7 @@ sess <- gdl_session(api_token)
 # COMMAND ----------
 
 sess <- sess %>%
-    set_dataset('areadata') %>%
+    set_dataset('demographics') %>%
     set_countries_all() %>%
     set_indicators(c('regpopm'))
     # by default linear extrapolation for 3 years

population/subnational_population_official_dlt.py

@@ -3,7 +3,7 @@
 from pyspark.sql import functions as F
 
 # Adding a new country requires adding the country here
-country_codes = ['moz', 'pry', 'ken', 'pak', 'bfa', 'col', 'cod', 'tun', 'btn', 'chl', 'nga', 'bgd', 'alb', "zaf", 'chl', 'gha']
+country_codes = ['moz', 'pry', 'ken', 'pak', 'bfa', 'col', 'cod', 'tun', 'btn', 'chl', 'nga', 'bgd', 'alb', "zaf", 'chl', 'gha', 'lbr']
 
 @dlt.table(name=f'subnational_population')
 def subnational_population():