GFS Wordcount.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#!/usr/bin/python\n",
    "\"\"\"GFS I/O PySpark example.\"\"\"\n",
    "import json\n",
    "import pprint\n",
    "import subprocess\n",
    "import pyspark\n",
    "\n",
    "sc = pyspark.SparkContext('local[*]')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Use the Google Cloud Storage bucket for temporary BigQuery export data used\n",
    "# by the InputFormat. This assumes the Google Cloud Storage connector for\n",
    "# Hadoop is configured.\n",
    "bq_project = \"savvy-aileron-127413\"\n",
    "fs_project = \"owners-681171445480\"\n",
    "bucket = \"spark_tmp\"\n",
    "account_email = \"savvy-aileron-127413@appspot.gserviceaccount.com\"\n",
    "input_directory = 'gs://{}/hadoop/tmp/bigquery/pyspark_input'.format(bucket)\n",
    "localKeyfile = '/mnt/key.p12'\n",
    "\n",
    "# Output Parameters\n",
    "output_dataset = 'tutorial'\n",
    "output_table = 'wordcount_table'\n",
    "\n",
    "conf = {\n",
    "    # Input Parameters\n",
    "    'fs.gs.impl': 'com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem',\n",
    "    'fs.AbstractFileSystem.gs.impl': 'com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS',\n",
    "    'fs.gs.project.id': fs_project,\n",
    "    'mapred.bq.project.id': bq_project,\n",
    "    'mapred.bq.gcs.bucket': bucket,\n",
    "    'mapred.bq.temp.gcs.path': input_directory,\n",
    "    'mapred.bq.input.project.id': 'publicdata',\n",
    "    'mapred.bq.input.dataset.id': 'samples',\n",
    "    'mapred.bq.input.table.id': 'shakespeare',\n",
    "    'google.cloud.auth.service.account.enable': 'true',\n",
    "    'google.cloud.auth.service.account.email': account_email,\n",
    "    'google.cloud.auth.service.account.keyfile': localKeyfile,\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Load data in from BigQuery.\n",
    "table_data = sc.newAPIHadoopRDD(\n",
    "    'com.google.cloud.hadoop.io.bigquery.JsonTextBigQueryInputFormat',\n",
    "    'org.apache.hadoop.io.LongWritable',\n",
    "    'com.google.gson.JsonObject',\n",
    "    conf=conf)\n",
    "\n",
    "# Perform word count.\n",
    "word_counts = (\n",
    "    table_data\n",
    "    .map(lambda (_, record): json.loads(record))\n",
    "    .map(lambda x: (x['word'].lower(), int(x['word_count'])))\n",
    "    .reduceByKey(lambda x, y: x + y))\n",
    "\n",
    "# Display 10 results.\n",
    "pprint.pprint(word_counts.take(10))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Stage data formatted as newline-delimited JSON in Google Cloud Storage.\n",
    "output_directory = 'gs://{}/hadoop/tmp/bigquery/pyspark_output'.format(bucket)\n",
    "partitions = range(word_counts.getNumPartitions())\n",
    "output_files = [output_directory + '/part-{:05}'.format(i) for i in partitions]\n",
    "\n",
    "(word_counts\n",
    " .map(lambda (w, c): json.dumps({'word': w, 'word_count': c}))\n",
    " .saveAsTextFile(output_directory))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Google cloud authentication\n",
    "subprocess.check_call(\n",
    "    '/home/jovyan/google-cloud-sdk/bin/gcloud auth activate-service-account {0} --key-file {1}'.format(account_email, localKeyfile).split())\n",
    "\n",
    "# Shell out to bq CLI to perform BigQuery import.\n",
    "subprocess.call(\n",
    "    '/home/jovyan/google-cloud-sdk/bin/bq load --source_format NEWLINE_DELIMITED_JSON '\n",
    "    '--project_id {project}'\n",
    "    '--schema word:STRING,word_count:INTEGER '\n",
    "    '{project}:{dataset}.{table} {files}'.format(\n",
    "        project=bq_project, dataset=output_dataset, table=output_table, files=','.join(output_files)\n",
    "    ).split())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Manually clean up the staging_directories, otherwise BigQuery\n",
    "# files will remain indefinitely.\n",
    "input_path = sc._jvm.org.apache.hadoop.fs.Path(input_directory)\n",
    "input_path.getFileSystem(sc._jsc.hadoopConfiguration()).delete(input_path, True)\n",
    "output_path = sc._jvm.org.apache.hadoop.fs.Path(output_directory)\n",
    "output_path.getFileSystem(sc._jsc.hadoopConfiguration()).delete(\n",
    "    output_path, True)"
   ]
  }
 ],
 "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
}