preprocessing.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Preprocessing of TED CAN (contract award notices) CSV files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark import SparkContext, SparkConf\n",
    "from pyspark.sql import SQLContext, Row\n",
    "from pyspark.sql import HiveContext\n",
    "import json\n",
    "import time\n",
    "import sys\n",
    "from datetime import datetime\n",
    "from pyspark.sql.types import *\n",
    "import re\n",
    "\n",
    "nPartitions=4\n",
    "conf = (SparkConf()\n",
    "         .setMaster(\"local[\"+str(nPartitions)+\"]\")\n",
    "       )\n",
    "#sc.stop()\n",
    "sc = SparkContext(conf=conf)\n",
    "sqlContext = SQLContext(sc)\n",
    "sqlContext.sql(\"SET spark.sql.parquet.binaryAsString=true\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Load country code file\n",
    "Used to match country ISO codes to country full names"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "countryCodesRaw = [line.rstrip() for line in open('code/data/countrycodes.csv')]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "countryCodes = {}\n",
    "for i in range(len(countryCodesRaw)):\n",
    "    code_country=countryCodesRaw[i].split(' - ')\n",
    "    countryCodes[code_country[0]] = code_country[1]\n",
    "countryCodes['']=''"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Load CPV meaning file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "CPVmeaningRaw = [line.rstrip() for line in open('code/data/CPVmeaning.csv')]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "CPVmeanings= {}\n",
    "for i in range(len(CPVmeaningRaw)):\n",
    "    CPV_meaning =CPVmeaningRaw[i].split(';')\n",
    "    CPVmeanings[CPV_meaning[0][0:8]] = CPV_meaning[1]\n",
    "CPVmeanings['']=''"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "CPVcorrespondanceRaw = [line.rstrip() for line in open('code/data/CPVcorrespondance2003_2007.csv')]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "CPVcorrespondance= {}\n",
    "for i in range(len(CPVcorrespondanceRaw)):\n",
    "    CPV_correspondance =CPVcorrespondanceRaw[i].split(';')\n",
    "    CPVcorrespondance[CPV_correspondance[0][0:8]] = CPV_correspondance[1][0:8]\n",
    "CPVcorrespondance[None]=''\n",
    "CPVcorrespondance['']=''"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## CSV row processor and schema\n",
    "Convert CSV rows to nicer format and schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def process(row):    \n",
    "    try:\n",
    "        dispatch_date=None if (row['DT_DISPATCH'] is None) else datetime.strptime(row['DT_DISPATCH'], '%d-%b-%y').strftime('%Y-%m-%d')\n",
    "        award_notice_id=None if (row['ID_NOTICE_CAN'] is None) else row['ID_NOTICE_CAN'].strip()\n",
    "        award_notice_id=award_notice_id[4:]+'-'+award_notice_id[0:4]\n",
    "        award_notice_id_link=\"<a href='http://ted.europa.eu/udl?uri=TED:NOTICE:\"+award_notice_id+\":TEXT:EN:HTML' target='_blank'>\"+award_notice_id+\"</a>\"\n",
    "\n",
    "        contracting_authority_country=None if (row['ISO_COUNTRY_CODE'] is None) else countryCodes[row['ISO_COUNTRY_CODE'].strip()]\n",
    "        contracting_authority_name=None if (row['CAE_NAME'] is None) else row['CAE_NAME'].replace (\"_\", \"-\").strip()\n",
    "\n",
    "        contractor_country=None if (row['WIN_COUNTRY_CODE'] is None) else countryCodes[row['WIN_COUNTRY_CODE'].strip()]\n",
    "        contractor_name=None if (row['WIN_NAME'] is None) else row['WIN_NAME'].replace (\"_\", \"-\").strip()\n",
    "\n",
    "        contract_value_euros=None if row['AWARD_VALUE_EURO']=='' else int(float(row['AWARD_VALUE_EURO']))\n",
    "        number_offers_received=None if row['NUMBER_OFFERS']=='' else int(row['NUMBER_OFFERS'])\n",
    "\n",
    "        CPV_code=None if (row['CPV'] is None) else row['CPV'].strip()\n",
    "        CPV_code_2008=CPV_code\n",
    "        #if CPV_code in CPVcorrespondance:\n",
    "        #    CPV_code_2008=CPVcorrespondance[CPV_code]\n",
    "        CPV_code_meaning=CPVmeanings[CPV_code_2008]\n",
    "\n",
    "        YEAR=None if row['YEAR']=='' else int(row['YEAR'])\n",
    "        ID_TYPE=None if (row['ID_TYPE'] is None) else row['ID_TYPE'].strip()\n",
    "        XSD_VERSION=None if (row['XSD_VERSION'] is None) else row['XSD_VERSION'].strip()\n",
    "        CANCELLED=None if (row['CANCELLED'] is None) else row['CANCELLED'].strip()\n",
    "        CORRECTIONS=None if row['CORRECTIONS']=='' else int(float(row['CORRECTIONS']))\n",
    "\n",
    "        CAE_NATIONALID=None if (row['CAE_NATIONALID'] is None) else row['CAE_NATIONALID'].strip()\n",
    "        CAE_ADDRESS=None if (row['CAE_ADDRESS'] is None) else row['CAE_ADDRESS'].strip()\n",
    "        CAE_TOWN=None if (row['CAE_TOWN'] is None) else row['CAE_TOWN'].strip()\n",
    "        CAE_POSTAL_CODE=None if (row['CAE_POSTAL_CODE'] is None) else row['CAE_POSTAL_CODE'].strip()\n",
    "\n",
    "        CAE_TYPE=None if (row['CAE_TYPE'] is None) else row['CAE_TYPE'].strip()\n",
    "        MAIN_ACTIVITY=None if (row['MAIN_ACTIVITY'] is None) else row['MAIN_ACTIVITY'].strip()\n",
    "        B_ON_BEHALF=None if (row['B_ON_BEHALF'] is None) else row['B_ON_BEHALF'].strip()\n",
    "        TYPE_OF_CONTRACT=None if (row['TYPE_OF_CONTRACT'] is None) else row['TYPE_OF_CONTRACT'].strip()\n",
    "        TAL_LOCATION_NUTS=None if (row['TAL_LOCATION_NUTS'] is None) else row['TAL_LOCATION_NUTS'].strip()\n",
    "        B_FRA_AGREEMENT=None if (row['B_FRA_AGREEMENT'] is None) else row['B_FRA_AGREEMENT'].strip()\n",
    "        B_DYN_PURCH_SYST=None if (row['B_DYN_PURCH_SYST'] is None) else row['B_DYN_PURCH_SYST'].strip()\n",
    "        ADDITIONAL_CPVS=None if (row['ADDITIONAL_CPVS'] is None) else row['ADDITIONAL_CPVS'].strip()\n",
    "        B_GPA=None if (row['B_GPA'] is None) else row['B_GPA'].strip()\n",
    "        VALUE_EURO_FIN_1=None if row['VALUE_EURO_FIN_1']=='' else int(float(row['VALUE_EURO_FIN_1']))\n",
    "        VALUE_EURO_FIN_2=None if row['VALUE_EURO_FIN_2']=='' else int(float(row['VALUE_EURO_FIN_2']))\n",
    "        TOP_TYPE=None if (row['TOP_TYPE'] is None) else row['TOP_TYPE'].strip()\n",
    "        CRIT_CODE=None if (row['CRIT_CODE'] is None) else row['CRIT_CODE'].strip()\n",
    "        CRIT_CRITERIA=None if (row['CRIT_CRITERIA'] is None) else row['CRIT_CRITERIA'].strip()\n",
    "        CRIT_WEIGHTS=None if (row['CRIT_WEIGHTS'] is None) else row['CRIT_WEIGHTS'].strip()\n",
    "        B_ELECTRONIC_AUCTION=None if (row['B_ELECTRONIC_AUCTION'] is None) else row['B_ELECTRONIC_AUCTION'].strip()\n",
    "        NUMBER_AWARDS=None if row['NUMBER_AWARDS']=='' else int(float(row['NUMBER_AWARDS']))\n",
    "\n",
    "        WIN_ADDRESS=None if (row['WIN_ADDRESS'] is None) else row['WIN_ADDRESS'].strip()\n",
    "        WIN_TOWN=None if (row['WIN_TOWN'] is None) else row['WIN_TOWN'].strip()\n",
    "        WIN_POSTAL_CODE=None if (row['WIN_POSTAL_CODE'] is None) else row['WIN_POSTAL_CODE'].strip()\n",
    "\n",
    "        ID_AWARD=None if (row['ID_AWARD'] is None) else row['ID_AWARD'].strip()\n",
    "        CONTRACT_NUMBER=None if (row['CONTRACT_NUMBER'] is None) else row['CONTRACT_NUMBER'].strip()\n",
    "        LOT_NUMBER=None if (row['LOT_NUMBER'] is None) else row['LOT_NUMBER'].strip()\n",
    "        TITLE=None if (row['TITLE'] is None) else row['TITLE'].strip()\n",
    "        NUMBER_OFFERS_ELECTR=None if row['NUMBER_OFFERS_ELECTR']=='' else int(float(row['NUMBER_OFFERS_ELECTR'])) \n",
    "        AWARD_EST_VALUE_EURO=None if row['AWARD_EST_VALUE_EURO']=='' else int(float(row['AWARD_EST_VALUE_EURO']))\n",
    "        AWARD_VALUE_EURO=None if row['AWARD_VALUE_EURO']=='' else int(float(row['AWARD_VALUE_EURO']))\n",
    "        AWARD_VALUE_EURO_FIN_1=None if row['AWARD_VALUE_EURO_FIN_1']=='' else int(float(row['AWARD_VALUE_EURO_FIN_1']))\n",
    "        B_SUBCONTRACTED=None if (row['B_SUBCONTRACTED'] is None) else row['B_SUBCONTRACTED'].strip()\n",
    "        B_EU_FUNDS=None if (row['B_EU_FUNDS'] is None) else row['B_EU_FUNDS'].strip() \n",
    "        DT_AWARD=None if row['DT_AWARD']=='' else datetime.strptime(row['DT_AWARD'].strip(), '%d-%b-%y').strftime('%Y-%m-%d')\n",
    "\n",
    "        awardNoticeRow=[\\\n",
    "                        contracting_authority_country,\\\n",
    "                        contracting_authority_name,\\\n",
    "                        dispatch_date,\\\n",
    "                        CPV_code_meaning,\\\n",
    "                        contractor_country,\\\n",
    "                        contractor_name,\\\n",
    "                        contract_value_euros,\\\n",
    "                        number_offers_received,\\\n",
    "                        CPV_code_2008,\\\n",
    "                        award_notice_id_link,\\\n",
    "\n",
    "                        CPV_code,\\\n",
    "                        YEAR,\\\n",
    "                        ID_TYPE,\\\n",
    "                        XSD_VERSION,\\\n",
    "                        CANCELLED,\\\n",
    "                        CORRECTIONS,\\\n",
    "\n",
    "                        CAE_NATIONALID,\\\n",
    "                        CAE_ADDRESS,\\\n",
    "                        CAE_TOWN,\\\n",
    "                        CAE_POSTAL_CODE,\\\n",
    "\n",
    "                        CAE_TYPE,\\\n",
    "                        MAIN_ACTIVITY,\\\n",
    "                        B_ON_BEHALF,\\\n",
    "                        TYPE_OF_CONTRACT,\\\n",
    "                        TAL_LOCATION_NUTS,\\\n",
    "                        B_FRA_AGREEMENT,\\\n",
    "                        B_DYN_PURCH_SYST,\\\n",
    "                        ADDITIONAL_CPVS,\\\n",
    "                        B_GPA,\\\n",
    "                        VALUE_EURO_FIN_1,\\\n",
    "                        VALUE_EURO_FIN_2,\\\n",
    "                        TOP_TYPE,\\\n",
    "                        CRIT_CODE,\\\n",
    "                        CRIT_CRITERIA,\\\n",
    "                        CRIT_WEIGHTS,\\\n",
    "                        B_ELECTRONIC_AUCTION,\\\n",
    "                        NUMBER_AWARDS,\\\n",
    "\n",
    "                        WIN_ADDRESS,\\\n",
    "                        WIN_TOWN,\\\n",
    "                        WIN_POSTAL_CODE,\\\n",
    "\n",
    "                        ID_AWARD,\\\n",
    "                        CONTRACT_NUMBER,\\\n",
    "                        LOT_NUMBER,\\\n",
    "                        TITLE,\\\n",
    "                        NUMBER_OFFERS_ELECTR,\\\n",
    "                        AWARD_EST_VALUE_EURO,\\\n",
    "                        AWARD_VALUE_EURO,\\\n",
    "                        AWARD_VALUE_EURO_FIN_1,\\\n",
    "                        B_SUBCONTRACTED,\\\n",
    "                        B_EU_FUNDS,\\\n",
    "                        DT_AWARD\\\n",
    "\n",
    "                       ]\n",
    "\n",
    "        awardNoticeRow=[None if elt==\"\" else elt for elt in awardNoticeRow] \n",
    "        #awardNoticeRow=(True,tuple(awardNoticeRow),True)\n",
    "        awardNoticeRow=tuple(awardNoticeRow)\n",
    "    except Exception as inst:\n",
    "        awardNoticeRow=(False,row,inst)\n",
    "        \n",
    "    return awardNoticeRow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "schema = StructType([\n",
    "    StructField(\"Contracting_Authority_Country\", StringType(), True),\n",
    "    StructField(\"Contracting_Authority_Name\", StringType(), True),\n",
    "    StructField(\"Dispatch_Date\", StringType(), True),\n",
    "    StructField(\"CPV_Code_Meaning\", StringType(), True),\n",
    "    StructField(\"Contractor_Country\", StringType(), True),\n",
    "    StructField(\"Contractor_Name\", StringType(), True),\n",
    "    StructField(\"Contract_Value_Euros\", LongType(), True),\n",
    "    StructField(\"Number_Offers_Received\", IntegerType(), True),\n",
    "    StructField(\"CPV_Code_2008\", StringType(), True),\n",
    "    StructField(\"Award_Notice_Id_Link\", StringType(), True),\n",
    "    \n",
    "    StructField(\"CPV_Code\", StringType(), True),\n",
    "    StructField(\"YEAR\", IntegerType(), True),\n",
    "    StructField(\"ID_TYPE\", StringType(), True),\n",
    "    StructField(\"XSD_VERSION\", StringType(), True),\n",
    "    StructField(\"CANCELLED\", StringType(), True),\n",
    "    StructField(\"CORRECTIONS\", IntegerType(), True),\n",
    "    \n",
    "    StructField(\"CAE_NATIONALID\", StringType(), True),\n",
    "    StructField(\"CAE_ADDRESS\", StringType(), True),\n",
    "    StructField(\"CAE_TOWN\", StringType(), True),\n",
    "    StructField(\"CAE_POSTAL_CODE\", StringType(), True),\n",
    "    \n",
    "    StructField(\"CAE_TYPE\", StringType(), True),\n",
    "    StructField(\"MAIN_ACTIVITY\", StringType(), True),\n",
    "    StructField(\"B_ON_BEHALF\", StringType(), True),\n",
    "    StructField(\"TYPE_OF_CONTRACT\", StringType(), True),\n",
    "    StructField(\"TAL_LOCATION_NUTS\", StringType(), True),\n",
    "    StructField(\"B_FRA_AGREEMENT\", StringType(), True),\n",
    "    StructField(\"B_DYN_PURCH_SYST\", StringType(), True),\n",
    "    StructField(\"ADDITIONAL_CPVS\", StringType(), True),\n",
    "    StructField(\"B_GPA\", StringType(), True),\n",
    "    StructField(\"VALUE_EURO_FIN_1\", LongType(), True),\n",
    "    StructField(\"VALUE_EURO_FIN_2\", LongType(), True),\n",
    "    StructField(\"TOP_TYPE\", StringType(), True),\n",
    "    StructField(\"CRIT_CODE\", StringType(), True),\n",
    "    StructField(\"CRIT_CRITERIA\", StringType(), True),\n",
    "    StructField(\"CRIT_WEIGHTS\", StringType(), True),\n",
    "    StructField(\"B_ELECTRONIC_AUCTION\", StringType(), True),\n",
    "    StructField(\"NUMBER_AWARDS\", IntegerType(), True),\n",
    "    \n",
    "    StructField(\"WIN_ADDRESS\", StringType(), True),\n",
    "    StructField(\"WIN_TOWN\", StringType(), True),\n",
    "    StructField(\"WIN_POSTAL_CODE\", StringType(), True),\n",
    "    \n",
    "    StructField(\"ID_AWARD\", StringType(), True),\n",
    "    StructField(\"CONTRACT_NUMBER\", StringType(), True),\n",
    "    StructField(\"LOT_NUMBER\", StringType(), True),\n",
    "    StructField(\"TITLE\", StringType(), True),\n",
    "    StructField(\"NUMBER_OFFERS_ELECTR\", IntegerType(), True),\n",
    "    StructField(\"AWARD_EST_VALUE_EURO\", LongType(), True),\n",
    "    StructField(\"AWARD_VALUE_EURO\", LongType(), True),\n",
    "    StructField(\"AWARD_VALUE_EURO_FIN_1\", LongType(), True),\n",
    "    StructField(\"B_SUBCONTRACTED\", StringType(), True),\n",
    "    StructField(\"B_EU_FUNDS\", StringType(), True),\n",
    "    StructField(\"DT_AWARD\", StringType(), True)\n",
    "    ])\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Load CSV data, convert, and save to parquet\n",
    "CSV files are in ../data\n",
    "\n",
    "Four files need to be processed:\n",
    "* TED_CAN_2006.csv\n",
    "* TED_CAN_2007.csv\n",
    "* TED_CAN_2008.csv\n",
    "* TED_CAN_2009_2015.csv\n",
    "\n",
    "which can be downloaded from https://data.europa.eu/euodp/en/data/dataset/ted-csv\n",
    "\n",
    "Preprocess to remove misformed CSV lines (removes \"\" and \\\")\n",
    "* time sed 's/\\\"\\\"//g' TED_CAN_2009_2015.csv |sed 's/\\\\\\\"/\"/g'> TED_CAN_2009_2015_2.csv\n",
    "* time sed 's/\\\"\\\"//g' TED_CAN_2008.csv |sed 's/\\\\\\\"/\"/g'> TED_CAN_2008_2.csv\n",
    "* time sed 's/\\\"\\\"//g' TED_CAN_2007.csv |sed 's/\\\\\\\"/\"/g'> TED_CAN_2007_2.csv\n",
    "* time sed 's/\\\"\\\"//g' TED_CAN_2006.csv |sed 's/\\\\\\\"/\"/g'> TED_CAN_2006_2.csv\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "pathData='data/TED_CAN_2009_2015_3.csv'\n",
    "#pathData='data2/TED_CAN_2015_3.csv'\n",
    "#pathData='data2/TED_CAN_2008_2.csv'\n",
    "#pathData='data2/TED_CAN_2007_2.csv'\n",
    "#pathData='data2/TED_CAN_2006_2.csv'\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Load"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "csvfile = sqlContext.read.format('com.databricks.spark.csv',).options(header='true', inferschema='false').load(pathData)#.repartition(2000)\n",
    "csvfile.registerTempTable(\"csvData\");\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Process"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "processedData=csvfile.rdd.map(process)\n",
    "df=processedData.toDF(schema)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "### Save"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df.write.mode('append').parquet('ted.parquet')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}