dataflow.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Google Cloud Dataflow\n",
    "\n",
    "Table of contents:\n",
    "\n",
    "- [Beam Programming](#beam-programming)\n",
    "- [Building a Simple Data Transformation Pipeline](#building-a-simple-data-transformation-pipeline)\n",
    "\n",
    "\n",
    "Google Cloud Dataflow provides a serverless, parallel and distributed infrastructure for running jobs for batch and stream data processing.\n",
    "One of the core strengths of Dataflow is its ability to almost seamlessly handle the switch from processing of batch historical data to streaming datasets while elegantly taking into consideration the perks of streaming processing such as windowing.\n",
    "Dataflow is a major component for building an end-to-end ML production pipeline on GCP."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"beam-programming\"></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Beam Programming\n",
    "Apache Beam provides a set of broad concepts to simplify the process of building a transformation pipeline for distributed batch and stream jobs.\n",
    "\n",
    "- **A Pipeline:** A Pipeline object wraps the entire operation and prescribes the transformation process by defining the input data source to the pipeline, how that data will be transformed and where the data will be written.\n",
    "- **A PCollection:** A PCollection is used to define a data source. The data source can either be bounded or unbounded. A bounded data source referes to batch or historical data, whereas an unbounded data source refers to streaming data.\n",
    "- **A PTransform:** PTransforms refers to a particular transformation task carried out on one or more PCollections in the pipeline. A number of core Beam transforms include:\n",
    "  - ParDo: for parallel processing.\n",
    "  - GroupByKey: for processing collections of key/value pairs.\n",
    "  - CoGroupByKey: for a relational join of two or more key/value PCollections with the same key type.\n",
    "  - Combine: for combining collections of elements or values in your data.\n",
    "  - Flatten: for merging multiple PCollection objects.\n",
    "  - Partition: splits a single PCollection into smaller collections. \n",
    "- **I/O Transforms:** These are PTransforms that read or write data to different external storage systems.\n",
    "\n",
    "<div style=\"display: inline-block;width: 100%;\">\n",
    "<img src=\"ieee-ompi/dataflow-sequential-transform.png\" style=\"float:left;\" alt=\"A Simple Linear Pipeline with Sequential Transforms.\" height=90% width=90% />\n",
    "</div>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"enable-dataflow-api\"></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Enable Dataflow API\n",
    "1. Go to API & Services Dashboard\n",
    "2. Click `Enable API & services`\n",
    "3. Search for `Dataflow API`\n",
    "\n",
    "<div style=\"display: inline-block;width: 100%;\">\n",
    "<img src=\"ieee-ompi/search-dataflow-api.png\" style=\"float:left;\" alt=\"Search for Dataflow API.\" height=90% width=90% />\n",
    "</div>\n",
    "\n",
    "4. Enable `Dataflow API`\n",
    "<div style=\"display: inline-block;width: 100%;\">\n",
    "<img src=\"ieee-ompi/enable-dataflow-api.png\" style=\"float:left;\" alt=\"Enable Dataflow API.\" height=90% width=90% />\n",
    "</div>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"building-a-simple-data-transformation-pipeline\"></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Building a Simple Data Transformation Pipeline\n",
    "In this example, a transformation pipeline is built to pre-process the `crypto-markets.csv` dataset by removing the attributes that are not relevant for data modeling, including filtering only bitcoin crypto records."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Creating gs://ieee-ompi-datasets/...\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "# create bucket\n",
    "gsutil mb gs://ieee-ompi-datasets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current\n",
      "                                 Dload  Upload   Total   Spent    Left  Speed\n",
      "\r",
      "  0     0    0     0    0     0      0      0 --:--:-- --:--:-- --:--:--     0\r",
      "  0     0    0     0    0     0      0      0 --:--:-- --:--:-- --:--:--     0Copying from <STDIN>...\n",
      "/ [0 files][    0.0 B/    0.0 B]                                                \r",
      "\r",
      "  0 47.0M    0 81054    0     0  50829      0  0:16:09  0:00:01  0:16:08 50817\r",
      "100 47.0M  100 47.0M    0     0  25.6M      0  0:00:01  0:00:01 --:--:-- 25.6M\n",
      "/ [0 files][ 47.0 MiB/    0.0 B]                                                \r",
      "/ [1 files][    0.0 B/    0.0 B]                                                \r\n",
      "Operation completed over 1 objects.                                              \n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "# transfer data from Github to the bucket.\n",
    "curl https://raw.githubusercontent.com/dvdbisong/IEEE-Carleton-and-OMPI-Machine-Learning-Workshop/master/data/crypto-markets/crypto-markets.csv | gsutil cp - gs://ieee-ompi-datasets/crypto-markets.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# install the apache beam library and other important setup packages.\n",
    "# restart the session after installing apache beam."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%bash\n",
    "source activate py2env\n",
    "pip uninstall -y google-cloud-dataflow\n",
    "conda install -y pytz==2018.4\n",
    "pip install apache-beam[gcp]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "# import relevant libraries\n",
    "import apache_beam as beam\n",
    "from apache_beam.io import ReadFromText\n",
    "from apache_beam.io import WriteToText"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "# transformation code\n",
    "def run(project, source_bucket, target_bucket):\n",
    "    import csv\n",
    "\n",
    "    options = {\n",
    "        'staging_location': 'gs://ieee-ompi-datasets/staging',\n",
    "        'temp_location': 'gs://ieee-ompi-datasets/temp',\n",
    "        'job_name': 'dataflow-crypto',\n",
    "        'project': project,\n",
    "        'max_num_workers': 24,\n",
    "        'teardown_policy': 'TEARDOWN_ALWAYS',\n",
    "        'no_save_main_session': True,\n",
    "        'runner': 'DataflowRunner'\n",
    "      }\n",
    "    options = beam.pipeline.PipelineOptions(flags=[], **options)\n",
    "    \n",
    "    crypto_dataset = 'gs://{}/crypto-markets.csv'.format(source_bucket)\n",
    "    processed_ds = 'gs://{}/transformed-crypto-bitcoin'.format(target_bucket)\n",
    "\n",
    "    pipeline = beam.Pipeline(options=options)\n",
    "\n",
    "    # 0:slug, 3:date, 5:open, 6:high, 7:low, 8:close\n",
    "    rows = (\n",
    "        pipeline |\n",
    "            'Read from bucket' >> ReadFromText(crypto_dataset) |\n",
    "            'Tokenize as csv columns' >> beam.Map(lambda line: next(csv.reader([line]))) |\n",
    "            'Select columns' >> beam.Map(lambda fields: (fields[0], fields[3], fields[5], fields[6], fields[7], fields[8])) |\n",
    "            'Filter bitcoin rows' >> beam.Filter(lambda row: row[0] == 'bitcoin')\n",
    "        )\n",
    "        \n",
    "    combined = (\n",
    "        rows |\n",
    "            'Write to bucket' >> beam.Map(lambda (slug, date, open, high, low, close): '{},{},{},{},{},{}'.format(\n",
    "                slug, date, open, high, low, close)) |\n",
    "            WriteToText(\n",
    "                file_path_prefix=processed_ds,\n",
    "                file_name_suffix=\".csv\", num_shards=2,\n",
    "                shard_name_template=\"-SS-of-NN\",\n",
    "                header='slug, date, open, high, low, close')\n",
    "        )\n",
    "\n",
    "    pipeline.run()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Run pipeline on the cloud\n"
     ]
    }
   ],
   "source": [
    "# execute transfomation\n",
    "if __name__ == '__main__':\n",
    "    print 'Run pipeline on the cloud'\n",
    "    run(project='oceanic-sky-230504', source_bucket='ieee-ompi-datasets', target_bucket='ieee-ompi-datasets')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Open Dataflow Dashboard\n",
    "<div style=\"display: inline-block;width: 100%;\">\n",
    "<img src=\"ieee-ompi/open-dataflow.png\" style=\"float:left;\" alt=\"Open Dataflow.\" height=90% width=90% />\n",
    "</div>\n",
    "\n",
    "### Choose Dataflow Job\n",
    "<div style=\"display: inline-block;width: 100%;\">\n",
    "<img src=\"ieee-ompi/choose-dataflow-job.png\" style=\"float:left;\" alt=\"Choose Dataflow Job.\" height=90% width=90% />\n",
    "</div>\n",
    "\n",
    "### Dataflow Transformation Pipeline\n",
    "<div style=\"display: inline-block;width: 100%;\">\n",
    "<img src=\"ieee-ompi/dataflow-transformation-pipeline.png\" style=\"float:left;\" alt=\"Dataflow transformation pipeline.\" height=90% width=90% />\n",
    "</div>\n",
    "\n",
    "### Transformed Dataset in Bucket\n",
    "<div style=\"display: inline-block;width: 100%;\">\n",
    "<img src=\"ieee-ompi/transformed-dataset.png\" style=\"float:left;\" alt=\"Transformed dataset.\" height=90% width=90% />\n",
    "</div>"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python [conda env:py2dl]",
   "language": "python",
   "name": "conda-env-py2dl-py"
  },
  "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.15"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}