1b_clean_mexico_survey.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 1b. Survey Cleaning - Mexico \n",
    "Mexico survey data is from the 2015 partial census, obtained from IPUMs. (https://international.ipums.org/international-action/sample_details/country/mx#tab_mx2015a)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import os\n",
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "import seaborn as sns\n",
    "import shutil\n",
    "from sklearn.decomposition import PCA\n",
    "from sklearn.preprocessing import StandardScaler\n",
    "from statsmodels.stats.weightstats import DescrStatsW"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import DataFrame, SparkSession\n",
    "from pyspark.sql import DataFrame as SparkDataFrame\n",
    "from pyspark.sql.functions import *\n",
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"random\") \\\n",
    "    .config(\"spark.sql.files.maxPartitionBytes\", 67108864) \\\n",
    "    .config(\"spark.driver.memory\", \"50g\") \\\n",
    "    .config(\"spark.driver.maxResultSize\", \"2g\")\\\n",
    "    .getOrCreate()\n",
    "spark.sparkContext.setLogLevel(\"ERROR\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "def save_df(df: SparkDataFrame, outfname: str, sep: str = ',') -> None:\n",
    "    \"\"\"\n",
    "    Saves spark dataframe to csv file, using work-around to deal with spark's automatic partitioning and naming\n",
    "    \"\"\"\n",
    "    outfolder = outfname[:-4]\n",
    "    df.repartition(1).write.csv(path=outfolder, mode=\"overwrite\", header=\"true\", sep=sep)\n",
    "    # Work around to deal with spark automatic naming\n",
    "    old_fname = [fname for fname in os.listdir(outfolder) if fname[-4:] == '.csv'][0]\n",
    "    os.rename(outfolder + '/' + old_fname, outfname)\n",
    "    shutil.rmtree(outfolder)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "def grouped_weighted_mean(df, agg_cols, feature_cols, weight_col):\n",
    "    data = df.copy()\n",
    "    for feature_col in feature_cols:\n",
    "        data[feature_col] = data[feature_col]*data[weight_col]\n",
    "    data = data.groupby(agg_cols, as_index=False).agg('sum')\n",
    "    for feature_col in feature_cols:\n",
    "        data[feature_col] = data[feature_col]/data[weight_col] \n",
    "    return data[agg_cols + feature_cols + [weight_col]]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.csv('/data/mosaiks/ipums/raw/ipumsi_00001.csv', header=True)\\\n",
    "    .select(['SERIAL', 'HHWT', 'GEO1_MX2015', 'GEO2_MX2015', 'ELECTRIC', 'PHONE', 'CELL', 'INTERNET', 'AUTOS', \n",
    "             'HOTWATER', 'AIRCON', 'COMPUTER', 'WASHER', 'REFRIG', 'TV', 'RADIO', 'URBAN'])\\\n",
    "    .withColumnRenamed('SERIAL', 'hhid')\\\n",
    "    .withColumnRenamed('HHWT', 'weight')\\\n",
    "    .withColumnRenamed('GEO1_MX2015', 'state')\\\n",
    "    .withColumnRenamed('GEO2_MX2015', 'municipality')\n",
    "\n",
    "df = df.withColumn('ELECTRIC', \n",
    "                   when(col('ELECTRIC') == 1, 1)\\\n",
    "                   .when(col('ELECTRIC') == 2, 0)\\\n",
    "                   .otherwise(np.nan))\n",
    "\n",
    "df = df.withColumn('PHONE', \n",
    "                   when(col('PHONE') == 1, 1)\\\n",
    "                   .when(col('PHONE') == 2, 0)\\\n",
    "                   .otherwise(np.nan))\n",
    "\n",
    "df = df.withColumn('CELL', \n",
    "                   when(col('CELL') == 1, 1)\\\n",
    "                   .when(col('CELL') == 2, 0)\\\n",
    "                   .otherwise(np.nan))\n",
    "\n",
    "df = df.withColumn('INTERNET', \n",
    "                   when(col('INTERNET') == 1, 1)\\\n",
    "                   .when(col('INTERNET') == 2, 0)\\\n",
    "                   .otherwise(np.nan))\n",
    "\n",
    "df = df.withColumn('AUTOS', \n",
    "                   when(col('AUTOS') == 0, 0)\\\n",
    "                   .when(col('AUTOS') == 8, np.nan)\\\n",
    "                   .when(col('AUTOS') == 9, np.nan)\\\n",
    "                   .otherwise(1))\n",
    "\n",
    "df = df.withColumn('HOTWATER', \n",
    "                   when(col('HOTWATER') == 1, 1)\\\n",
    "                   .when(col('HOTWATER') == 2, 0)\\\n",
    "                   .otherwise(np.nan))\n",
    "\n",
    "df = df.withColumn('AIRCON', \n",
    "                   when(col('AIRCON') == 10, 0)\\\n",
    "                   .when(col('AIRCON') == 0, np.nan)\\\n",
    "                   .when(col('AIRCON') == 99, np.nan)\\\n",
    "                   .otherwise(1))\n",
    "\n",
    "df = df.withColumn('COMPUTER', \n",
    "                   when(col('COMPUTER') == 1, 1)\\\n",
    "                   .when(col('COMPUTER') == 2, 0)\\\n",
    "                   .otherwise(np.nan))\n",
    "\n",
    "df = df.withColumn('WASHER', \n",
    "                   when(col('WASHER') == 0, np.nan)\\\n",
    "                   .when(col('WASHER') == 9, np.nan)\\\n",
    "                   .when(col('WASHER') == 1, 0)\\\n",
    "                   .otherwise(1))\n",
    "\n",
    "df = df.withColumn('REFRIG', \n",
    "                   when(col('REFRIG') == 1, 1)\\\n",
    "                   .when(col('REFRIG') == 2, 0)\\\n",
    "                   .otherwise(np.nan))\n",
    "\n",
    "df = df.withColumn('TV', \n",
    "                   when(col('TV') == 0, np.nan)\\\n",
    "                   .when(col('TV') == 99, np.nan)\\\n",
    "                   .when(col('TV') == 10, 0)\n",
    "                   .otherwise(1))\n",
    "\n",
    "df = df.withColumn('RADIO', \n",
    "                   when(col('RADIO') == 1, 1)\\\n",
    "                   .when(col('RADIO') == 2, 0)\\\n",
    "                   .otherwise(np.nan))\n",
    "\n",
    "df = df.withColumn('RURAL', \n",
    "                   when(col('URBAN') == 2, 0)\\\n",
    "                   .when(col('URBAN') == 1, 1)\\\n",
    "                   .otherwise(np.nan))\n",
    "\n",
    "save_df(df, '/data/mosaiks/replication/ipums/raw/mexico_reduced.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of households: 2927196\n",
      "Number of households with all assets: 2871373\n",
      "Explained variance: 0.34\n",
      "Number of individuals with all assets: 11137182\n",
      "Number of municipalities: 2446\n"
     ]
    }
   ],
   "source": [
    "# Read in data\n",
    "indiv = pd.read_csv('/data/mosaiks/replication/ipums/raw/mexico_reduced.csv')\n",
    "df = indiv.drop_duplicates(subset=['hhid'])\n",
    "print('Number of households: %i' % len(df))\n",
    "\n",
    "# Create asset index\n",
    "assets = ['ELECTRIC', 'PHONE', 'CELL', 'INTERNET', 'AUTOS', 'HOTWATER', 'AIRCON', 'COMPUTER', 'WASHER',\n",
    "         'REFRIG', 'TV', 'RADIO']\n",
    "df = df.dropna(subset=assets)\n",
    "print('Number of households with all assets: %i' % len(df))\n",
    "pca = PCA(n_components=1, svd_solver='arpack')\n",
    "scaler = StandardScaler()\n",
    "standardized_assets = scaler.fit_transform(df[assets])\n",
    "asset_index = pca.fit_transform(standardized_assets)\n",
    "df['asset_index'] = asset_index\n",
    "print('Explained variance: %.2f' % pca.explained_variance_ratio_)\n",
    "\n",
    "# Get poor indicator - bottom 41.9% by asset index\n",
    "wq = DescrStatsW(data=df['asset_index'].values, weights=df['weight'].values)\n",
    "cutoff = wq.quantile(probs=np.array([.419]), return_pandas=False)[0]\n",
    "df['poor'] = (df['asset_index'] < cutoff).astype('int')\n",
    "\n",
    "df = indiv.merge(df[['hhid', 'asset_index', 'poor']], on='hhid', how='inner')\n",
    "df = df.rename({'RURAL':'rural'}, axis=1)\n",
    "print('Number of individuals with all assets: %i' % len(df))\n",
    "print('Number of municipalities: %i' % len(df['municipality'].unique()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Percent of municipalities where more than half of HH are rural: 0.56\n",
      "Correlation between rural and wealth: -0.51\n"
     ]
    }
   ],
   "source": [
    "hh = df.drop_duplicates(subset=['hhid'])\n",
    "grouped_hh = grouped_weighted_mean(hh, ['municipality', 'state'], ['asset_index', 'rural'], 'weight')\n",
    "grouped_hh['rural'] = (grouped_hh['rural'] > 0.5).astype('int')\n",
    "grouped = grouped_hh.drop('weight', axis=1)\n",
    "print('Percent of municipalities where more than half of HH are rural: %.2f' % grouped['rural'].mean())\n",
    "print('Correlation between rural and wealth: %.2f' % np.corrcoef(grouped['rural'], grouped['asset_index'])[0][1])\n",
    "grouped[['municipality', 'state', 'asset_index', 'rural']]\\\n",
    "    .to_csv('/data/mosaiks/replication/surveys/mexico/grouped.csv', index=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "py3",
   "language": "python",
   "name": "py3"
  },
  "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.7.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}