notebooks/HelloWorld.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Hello World\n",
    "This notebook presents a very simple workflow writing some strings to a file. It showcases the ability to run Python Dataflow workflows from Jupyter notebooks.\n",
    "\n",
    "## Prerequisites\n",
    "\n",
    "The steps required to use this notebook are:\n",
    "1. Install the tools for creating conda environments: http://conda.pydata.org/miniconda.html\n",
    "2. Create and activate a virtual environment: conda create -n NAME jupyter; source activate NAME\n",
    "3. Install the Python Dataflow package (current latest version is v0.2.3): pip install https://github.com/GoogleCloudPlatform/DataflowPythonSDK/archive/v0.2.3.tar.gz\n",
    "4. Start a Jupyter notebook: jupyter notebook"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Workflow\n",
    "\n",
    "Start with an idiomatic import statement. Most objects are in the _df_ or _df.io_ namespace"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import google.cloud.dataflow as df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "There are three steps involved in creating and running a pipeline:\n",
    "* Create the `Pipeline` object\n",
    "* Create the graph of data transforms\n",
    "* Run the pipeline graph\n",
    "\n",
    "The graph of transforms is sometimes called workflow and therefore running a workflow is the same thing as running a pipeline. \n",
    "\n",
    "## Create a Pipeline object\n",
    "The code below associates with the pipeline a runner that will execute the workflow locally. This is very useful for testing your code. For running at scale the same identical code can be run in the Google Cloud: https://github.com/GoogleCloudPlatform/DataflowPythonSDK/#signing-up-for-alpha-batch-cloud-execution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "p = df.Pipeline('DirectPipelineRunner')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Build the graph of transforms\n",
    "A `Create` transform creates a PCollection out if its initializer elements. The `Write` transform writes the input PCollection to a text file sink one element per line."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<PValue transform=<_NativeWrite(PTransform) label=[native_write]> at 0x7fa8e6742890>"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "p | df.Create(['hello', 'world']) | df.io.Write(df.io.TextFileSink('./test.txt'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Run the pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<google.cloud.dataflow.runners.direct_runner.DirectPipelineResult at 0x7fa8e67bde50>"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "p.run()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The output file contains the written elements. There is no guarantee that a transform (including a Write) will output elements in the same order as they are received as input. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "hello\r\n",
      "world\r\n"
     ]
    }
   ],
   "source": [
    "!more ./test.txt"
   ]
  }
 ],
 "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
}