dataprep_spark.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import Row"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# load converters table\n",
    "rdd = sc.textFile(\"/insight_converter_site_history.txt\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "rdd2=rdd.map(lambda r: r.split(\"\\t\"))\n",
    "rdd3=rdd2.map(lambda r: Row(date = r[0],\n",
    "                           userid = r[1],\n",
    "                           siteid = r[2],\n",
    "                           timestamp = r[3],\n",
    "                           visit = r[4])).setName(\"conv_rdd\").persist(StorageLevel.MEMORY_AND_DISK_SER)\n",
    "df = sqlContext.createDataFrame(rdd3)\n",
    "df.registerTempTable(\"converters\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# load users table\n",
    "users_rdd = sc.textFile(\"/insight_site_staging_10M.txt\")\\\n",
    "          .map(lambda r: r.split(\"\\t\"))\\\n",
    "          .map(lambda r: Row(date=r[0], \n",
    "                             userid=r[1],\n",
    "                             siteid=r[2],\n",
    "                             timestamp=r[3],\n",
    "                             visit=r[4]))\\\n",
    "          .setName(\"users_rdd\")\\\n",
    "          .persist(StorageLevel.MEMORY_AND_DISK_SER)\n",
    "users_df = sqlContext.createDataFrame(users_rdd)\n",
    "users_df.registerTempTable(\"users\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# clearn users table\n",
    "siteid_nusers = sqlContext.sql(\"\"\"\n",
    "    SELECT siteid, count(userid) as num_users \n",
    "    FROM users WHERE siteid <> 'NULL'\n",
    "    GROUP BY siteid \n",
    "    ORDER BY num_users DESC\n",
    "    \"\"\")\n",
    "siteid_nusers.registerTempTable(\"siteid_nusers\")\n",
    "siteid_nusers_rdd = siteid_nusers.rdd.setName(\"siteid_nusers_rdd\")\\\n",
    "                         .persist(StorageLevel.MEMORY_AND_DISK_SER)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# filter siteids in a given range of number of users\n",
    "siteid_nusers_select = sqlContext.sql(\"\"\"\n",
    "    SELECT siteid, num_users\n",
    "    FROM siteid_nusers\n",
    "    WHERE num_users < 50000 and num_users > 150\n",
    "    \"\"\")\n",
    "siteid_nusers_select_rdd = siteid_nusers_select.rdd.map(lambda x:\n",
    "                        (x.siteid,\" \".join([x.siteid,str(x.num_users)]))\n",
    "                        ).setName(\"siteid_nusers_select_rdd\")\\\n",
    "                         .persist(StorageLevel.MEMORY_AND_DISK_SER)\n",
    "siteid_nusers_select.registerTempTable(\"siteid_nusers_select\")\n",
    "siteid_nusers_select_rdd1 = siteid_nusers_select.rdd.map(lambda x:\n",
    "                        \" \".join([x.siteid,str(x.num_users)]))\\\n",
    "                         .setName(\"siteid_nusers_select_rdd1\")\\\n",
    "                         .persist(StorageLevel.MEMORY_AND_DISK_SER)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# select users wheren siteid is in my range\n",
    "users_sampled=sqlContext.sql(\"\"\"\n",
    "    SELECT users.userid, users.siteid,users.timestamp,users.visit\n",
    "    FROM users,siteid_nusers_select\n",
    "    WHERE users.siteid = siteid_nusers_select.siteid\n",
    "    \"\"\")\n",
    "users_sampled_rdd = users_sampled.rdd.setName(\"users_sampled_rdd\")\\\n",
    "                         .persist(StorageLevel.MEMORY_AND_DISK_SER)\n",
    "users_sampled.registerTempTable(\"users_sampled\")\n",
    "#site_joined = site_rdd.join(siteid_nusers_select_rdd)\n",
    "#site_joined.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# filter converter table in the selected range of sites\n",
    "conv_sampled=sqlContext.sql(\"\"\"\n",
    "    SELECT userid,converters.siteid,timestamp,visit \n",
    "    FROM siteid_nusers_select,converters \n",
    "    WHERE siteid_nusers_select.siteid = converters.siteid\"\"\")\n",
    "conv_sampled_rdd = conv_sampled.rdd.setName(\"conv_sampled_rdd\")\\\n",
    "                         .persist(StorageLevel.MEMORY_AND_DISK_SER)\n",
    "conv_sampled.registerTempTable(\"conv_sampled\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# compute shared users to fill similarity matrix\n",
    "siteids_shared_users=sqlContext.sql(\"\"\"\n",
    "    SELECT c1.siteid AS siteid1,c2.siteid AS siteid2,count(*) as shared\n",
    "    FROM users_sampled AS c1,users_sampled AS c2 \n",
    "    WHERE c1.userid = c2.userid AND c1.siteid <> c2.siteid \n",
    "    GROUP BY c1.siteid,c2.siteid\"\"\")\n",
    "siteids_shared_users.registerTempTable(\"siteids_shared_users\")\n",
    "siteids_shared_users_rdd = siteids_shared_users.rdd.setName(\"siteids_shared_users\")\\\n",
    "                         .persist(StorageLevel.MEMORY_AND_DISK_SER)\n",
    "siteids_shared_users.registerTempTable(\"siteids_shared_users\")\n",
    "# siteids_shared_users.rdd.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# export shared users for PIC clustering\n",
    "rdd6 = siteids_shared_users.rdd.map(lambda r: \", \".join([r.siteid1,r.siteid2,str(r.shared)])).repartition(1)\n",
    "rdd6.saveAsTextFile(\"hdfs:///data/pic_users_50M_100000_100\")\n",
    "# export number of users per site\n",
    "rdd7 = siteid_nusers_select_rdd1.repartition(1)\n",
    "rdd7.saveAsTextFile(\"hdfs:///data/siteids_n_users_10M_50000_150_n\")\n",
    "# export shared users for matrix\n",
    "rdd5 = siteids_shared_users_rdd.map(\n",
    "    lambda r: (r.siteid1, r.siteid2, r.shared)\n",
    "    ).repartition(1)\n",
    "rdd5.saveAsTextFile(\"hdfs:///data/siteids_shared_users_10M_50000_150\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# prepare users table for join with converters\n",
    "users_sampled_join=users_sampled_rdd.map(lambda x: \" \".join(\n",
    "        [x.userid,x.siteid,x.timestamp,x.visit]\n",
    "        ))\n",
    "users_pairs = users_sampled_join.map(lambda x: (x.split(\" \")[0], x))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# prepare converters table for join with users\n",
    "conv_sampled_join=conv_sampled_rdd.map(lambda x: \" \".join(\n",
    "        [x.userid,x.siteid,x.timestamp,x.visit]\n",
    "        ))\n",
    "conv_pairs = conv_sampled_join.map(lambda x: (x.split(\" \")[0], x))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# identify users with no conversion\n",
    "users_subtract_conv = users_pairs.subtractByKey(conv_pairs).setName(\"users_subtract_conv\")\\\n",
    "                         .persist(StorageLevel.MEMORY_AND_DISK_SER)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# identify converters with match in the user population \n",
    "users_conv_joinC = sqlContext.sql(\"\"\"\n",
    "    SELECT \n",
    "        conv_sampled.userid,\n",
    "        conv_sampled.siteid,\n",
    "        conv_sampled.timestamp,\n",
    "        conv_sampled.visit\n",
    "    FROM conv_sampled,users_sampled\n",
    "    WHERE users_sampled.userid = conv_sampled.userid \n",
    "    AND users_sampled.siteid = conv_sampled.siteid\n",
    "    \"\"\")\n",
    "\n",
    "users_conv_joinC_rdd = users_conv_joinC.rdd.setName(\"users_conv_joinC_rdd\")\\\n",
    "                         .persist(StorageLevel.MEMORY_AND_DISK_SER)\n",
    "users_conv_joinC.registerTempTable(\"users_conv_joinC\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# load computed projection of PC for sites\n",
    "rdd20 = sc.textFile(\"/pca.csv\")\n",
    "rdd21=rdd20.map(lambda x: (\" \".join(x.split(\",\"))))\n",
    "rdd22=rdd21.map(lambda x: (x.split(\" \")[0],\" \".join(x.split(\" \")[1:])))\n",
    "#rdd22.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#load computed clusters for sites\n",
    "rdd30 = sc.textFile(\"/clustering_labels6\")\n",
    "rdd31=rdd30.map(lambda x: (x.split(\" \")[0], x.split(\" \")[1]))\n",
    "#rdd31.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# make joins on the converters list\n",
    "conv_pairs1 = users_conv_joinC_rdd.map(lambda x: (x.siteid,\" \".join([x.userid,x.siteid,x.timestamp,x.visit]) ))\n",
    "conv_out = conv_pairs1.join(rdd31).values().map(lambda x: \" \".join([x[0],x[1]]))\n",
    "conv_pairs2 = conv_out.map(lambda x: (x.split(\" \")[1],x))\n",
    "conv_out1 = conv_pairs2.join(rdd22).values().map(lambda x: \" \".join([x[0],x[1]]))\n",
    "conv_out1.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# make joins on the users list\n",
    "users_pairs1 = users_subtract_conv.values().map(lambda x: (x.split(\" \")[1],x))\n",
    "users_out = users_pairs1.join(rdd31).values().map(lambda x: \" \".join([x[0],x[1]]))\n",
    "users_pairs2 = users_out.map(lambda x: (x.split(\" \")[1],x))\n",
    "users_out1 = users_pairs2.join(rdd22).values().map(lambda x: \" \".join([x[0],x[1]]))\n",
    "users_out1.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# print csv for converters set\n",
    "rdd41 = conv_out1.map(lambda x: \",\".join(x.split(\" \"))).repartition(1)\n",
    "rdd41.saveAsTextFile(\"hdfs:///data/converters_10M_50000_150\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# print csv for users set\n",
    "rdd42 = users_out1.map(lambda x: \",\".join(x.split(\" \"))).repartition(1)\n",
    "rdd42.saveAsTextFile(\"hdfs:///data/users_10M_50000_150\")"
   ]
  }
 ],
 "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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}