Dataflow_Tutorial1.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Cloud Dataflow Tutorial\n",
    "\n",
    "## 事前準備\n",
    "- Google Cloud Platform の課金設定\n",
    "- [Dataflow APIの有効化](https://console.cloud.google.com/apis/api/dataflow.googleapis.com/overview)\n",
    "- [GCSのBucketを作る](https://console.cloud.google.com/storage/browser)\n",
    "- [BigQueryに`testdataset`というデータセットを作る](https://console.cloud.google.com/bigquery)\n",
    "- Datalabを起動\n",
    "\n",
    "That's it!\n",
    "\n",
    "## このNotebookをコピーするには\n",
    "Datalabを開いたら、Notebookを新規に開いてください。\n",
    "その後、セルに次のコードを入力して実行してください。\n",
    "\n",
    "```\n",
    "!git clone https://github.com/hayatoy/dataflow-tutorial.git\n",
    "```\n",
    "\n",
    "先頭の\" ! \"を忘れずに入力してください。\n",
    "\n",
    "## 実行する前に・・\n",
    "Project名を変更してください。`Esc`->`F`で一括置換できます。\n",
    "<font color=\"red\">注意：runAllを実行しないでください。全部実行するのに時間がかかります。</font>\n",
    "\n",
    "このNotebookはDatalab (Dataflow 0.6.0)用です。\n",
    "Dataflow 2.0.0以降で使う場合は `beam.utils` の部分を `beam.options` に変更してください。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Apache Beamのインポート"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import apache_beam as beam"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Dataflowの基本設定\n",
    "ジョブ名、プロジェクト名、一時ファイルの置き場を指定します。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "options = beam.utils.pipeline_options.PipelineOptions()\n",
    "gcloud_options = options.view_as(\n",
    "    beam.utils.pipeline_options.GoogleCloudOptions)\n",
    "gcloud_options.job_name = 'dataflow-tutorial1'\n",
    "gcloud_options.project = 'PROJECTID'\n",
    "gcloud_options.staging_location = 'gs://PROJECTID/staging'\n",
    "gcloud_options.temp_location = 'gs://PROJECTID/temp'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Dataflowのスケール設定\n",
    "Workerの最大数や、マシンタイプ等を設定します。  \n",
    "WorkerのDiskサイズは**デフォルトで250GB(Batch)、420GB(Streaming)と大きい**ので、ここで必要サイズを指定する事をオススメします。\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "worker_options = options.view_as(beam.utils.pipeline_options.WorkerOptions)\n",
    "worker_options.disk_size_gb = 20\n",
    "worker_options.max_num_workers = 2\n",
    "# worker_options.num_workers = 2\n",
    "# worker_options.machine_type = 'n1-standard-8'\n",
    "# worker_options.zone = 'asia-northeast1-a'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 実行環境の切り替え\n",
    "- **DirectRunner:** ローカルマシンで実行します\n",
    "- **DataflowRunner:** Dataflow上で実行します"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "options.view_as(beam.utils.pipeline_options.StandardOptions).runner = 'DirectRunner'\n",
    "# options.view_as(beam.utils.pipeline_options.StandardOptions).runner = 'DataflowRunner'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 準備は完了、以下パイプラインの例\n",
    "<br/>\n",
    "<br/>\n",
    "<br/>\n",
    "<br/>\n",
    "<br/>\n",
    "<br/>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## パイプラインその１\n",
    "GCSからファイルを読み込み、GCSにその内容を書き込むだけ  \n",
    "\n",
    "```\n",
    "+----------------+\n",
    "|                |\n",
    "| Read GCS File  |\n",
    "|                |\n",
    "+-------+--------+\n",
    "        |\n",
    "        v\n",
    "+-------+--------+\n",
    "|                |\n",
    "| Write GCS File |\n",
    "|                |\n",
    "+----------------+\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "p1 = beam.Pipeline(options=options)\n",
    "\n",
    "(p1 | 'read' >> beam.io.ReadFromText('gs://dataflow-samples/shakespeare/kinglear.txt')\n",
    "    | 'write' >> beam.io.WriteToText('gs://PROJECTID/test.txt', num_shards=1)\n",
    " )\n",
    "\n",
    "p1.run().wait_until_finish()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## パイプラインその２\n",
    "BigQueryからデータを読み込み、GCSにその内容を書き込むだけ  \n",
    "BigQueryのデータセットは以下  \n",
    "https://bigquery.cloud.google.com/table/bigquery-public-data:samples.shakespeare  \n",
    "\n",
    "```\n",
    "+----------------+\n",
    "|                |\n",
    "| Read BigQuery  |\n",
    "|                |\n",
    "+-------+--------+\n",
    "        |\n",
    "        v\n",
    "+-------+--------+\n",
    "|                |\n",
    "| Write GCS File |\n",
    "|                |\n",
    "+----------------+\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "p2 = beam.Pipeline(options=options)\n",
    "\n",
    "query = 'SELECT * FROM [bigquery-public-data:samples.shakespeare] LIMIT 10'\n",
    "(p2 | 'read' >> beam.io.Read(beam.io.BigQuerySource(project='PROJECTID', use_standard_sql=False, query=query))\n",
    "    | 'write' >> beam.io.WriteToText('gs://PROJECTID/test2.txt', num_shards=1)\n",
    " )\n",
    "\n",
    "p2.run().wait_until_finish()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## パイプラインその３\n",
    "BigQueryからデータを読み込み、BigQueryにデータを書き込む  \n",
    "\n",
    "```\n",
    "+----------------+\n",
    "|                |\n",
    "| Read BigQuery  |\n",
    "|                |\n",
    "+-------+--------+\n",
    "        |\n",
    "        v\n",
    "+-------+--------+\n",
    "|                |\n",
    "| Write BigQuery |\n",
    "|                |\n",
    "+----------------+\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "p3 = beam.Pipeline(options=options)\n",
    "\n",
    "# 注意：データセットを作成しておく\n",
    "query = 'SELECT * FROM [bigquery-public-data:samples.shakespeare] LIMIT 10'\n",
    "(p3 | 'read' >> beam.io.Read(beam.io.BigQuerySource(project='PROJECTID', use_standard_sql=False, query=query))\n",
    "    | 'write' >> beam.io.Write(beam.io.BigQuerySink(\n",
    "        'testdataset.testtable1',\n",
    "        schema='corpus_date:INTEGER, corpus:STRING, word:STRING, word_count:INTEGER',\n",
    "        create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,\n",
    "        write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE))\n",
    " )\n",
    "\n",
    "p3.run().wait_until_finish()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## パイプラインその４\n",
    "- BigQueryからデータを読み込み\n",
    "- データを加工して\n",
    "- BigQueryに書き込む\n",
    "\n",
    "```\n",
    "+----------------+\n",
    "|                |\n",
    "| Read BigQuery  |\n",
    "|                |\n",
    "+-------+--------+\n",
    "        |\n",
    "        v\n",
    "+-------+--------+\n",
    "|                |\n",
    "| Modify Element |\n",
    "|                |\n",
    "+----------------+\n",
    "        |\n",
    "        v\n",
    "+-------+--------+\n",
    "|                |\n",
    "| Write BigQuery |\n",
    "|                |\n",
    "+----------------+\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def modify_data1(element):\n",
    "    # beam.Mapは１行の入力に対し１行の出力をする場合に使う\n",
    "    # element = {u'corpus_date': 0, u'corpus': u'sonnets', u'word': u'LVII', u'word_count': 1}\n",
    "\n",
    "    corpus_upper = element['corpus'].upper()\n",
    "    word_len = len(element['word'])\n",
    "\n",
    "    return {'corpus_upper': corpus_upper,\n",
    "            'word_len': word_len\n",
    "            }\n",
    "\n",
    "\n",
    "p4 = beam.Pipeline(options=options)\n",
    "\n",
    "query = 'SELECT * FROM [bigquery-public-data:samples.shakespeare] LIMIT 10'\n",
    "(p4 | 'read' >> beam.io.Read(beam.io.BigQuerySource(project='PROJECTID', use_standard_sql=False, query=query))\n",
    "    | 'modify' >> beam.Map(modify_data1)\n",
    "    | 'write' >> beam.io.Write(beam.io.BigQuerySink(\n",
    "        'testdataset.testtable2',\n",
    "        schema='corpus_upper:STRING, word_len:INTEGER',\n",
    "        create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,\n",
    "        write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE))\n",
    " )\n",
    "\n",
    "p4.run().wait_until_finish()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## パイプラインその５\n",
    "ブランチを分ける例\n",
    "\n",
    "```\n",
    "+----------------+\n",
    "|                |\n",
    "| Read BigQuery  |\n",
    "|                |\n",
    "+-------+--------+\n",
    "        |\n",
    "        +---------------------+\n",
    "        |                     |\n",
    "+-------v--------+    +-------v--------+\n",
    "|                |    |                |\n",
    "| Modify Element |    | Modify Element |\n",
    "|                |    |                |\n",
    "+-------+--------+    +-------+--------+\n",
    "        |                     |\n",
    "        +---------------------+\n",
    "        |\n",
    "+-------v--------+\n",
    "|                |\n",
    "| Flatten        |\n",
    "|                |\n",
    "+-------+--------+\n",
    "        |\n",
    "        |\n",
    "+-------v--------+\n",
    "|                |\n",
    "| Save BigQuery  |\n",
    "|                |\n",
    "+----------------+\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def modify1(element):\n",
    "    # element = {u'corpus_date': 0, u'corpus': u'sonnets', u'word': u'LVII', u'word_count': 1}\n",
    "    word_count = len(element['corpus'])\n",
    "    count_type = 'corpus only'\n",
    "\n",
    "    return {'word_count': word_count,\n",
    "            'count_type': count_type\n",
    "            }\n",
    "\n",
    "\n",
    "def modify2(element):\n",
    "    # element = {u'corpus_date': 0, u'corpus': u'sonnets', u'word': u'LVII', u'word_count': 1}\n",
    "    word_count = len(element['word'])\n",
    "    count_type = 'word only'\n",
    "\n",
    "    return {'word_count': word_count,\n",
    "            'count_type': count_type\n",
    "            }\n",
    "\n",
    "\n",
    "p5 = beam.Pipeline(options=options)\n",
    "\n",
    "query = 'SELECT * FROM [bigquery-public-data:samples.shakespeare] LIMIT 10'\n",
    "query_results = p5 | 'read' >> beam.io.Read(beam.io.BigQuerySource(\n",
    "    project='PROJECTID', use_standard_sql=False, query=query))\n",
    "\n",
    "# BigQueryの結果を二つのブランチに渡す\n",
    "branch1 = query_results | 'modify1' >> beam.Map(modify1)\n",
    "branch2 = query_results | 'modify2' >> beam.Map(modify2)\n",
    "\n",
    "# ブランチからの結果をFlattenでまとめる\n",
    "((branch1, branch2) | beam.Flatten()\n",
    "                    | 'write' >> beam.io.Write(beam.io.BigQuerySink(\n",
    "                        'testdataset.testtable3',\n",
    "                        schema='word_count:INTEGER, count_type:STRING',\n",
    "                        create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,\n",
    "                        write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE))\n",
    " )\n",
    "\n",
    "p5.run().wait_until_finish()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## パイプラインその６\n",
    "\n",
    "**Groupby**を使う"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def modify_data2(kvpair):\n",
    "    # groupbyによりkeyとそのkeyを持つデータのリストのタプルが渡される\n",
    "    # kvpair = (u'word only', [4, 4, 6, 6, 7, 7, 7, 7, 8, 9])\n",
    "\n",
    "    return {'count_type': kvpair[0],\n",
    "            'sum': sum(kvpair[1])\n",
    "            }\n",
    "\n",
    "\n",
    "p6 = beam.Pipeline(options=options)\n",
    "\n",
    "query = 'SELECT * FROM [PROJECTID:testdataset.testtable3] LIMIT 20'\n",
    "(p6 | 'read' >> beam.io.Read(beam.io.BigQuerySource(project='PROJECTID', use_standard_sql=False, query=query))\n",
    "    | 'pair' >> beam.Map(lambda x: (x['count_type'], x['word_count']))\n",
    "    | \"groupby\" >> beam.GroupByKey()\n",
    "    | 'modify' >> beam.Map(modify_data2)\n",
    "    | 'write' >> beam.io.Write(beam.io.BigQuerySink(\n",
    "        'testdataset.testtable4',\n",
    "        schema='count_type:STRING, sum:INTEGER',\n",
    "        create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,\n",
    "        write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE))\n",
    " )\n",
    "\n",
    "p6.run().wait_until_finish()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## パイプラインその７\n",
    "**Window**で**GroupBy**の区間を区切る"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def assign_timevalue(v):\n",
    "    # pcollectionのデータにタイムスタンプを付加する\n",
    "    # 後段のwindowはこのタイムスタンプを基準に分割される\n",
    "    # ここでは適当に乱数でタイムスタンプを入れている\n",
    "    import apache_beam.transforms.window as window\n",
    "    import random\n",
    "    import time\n",
    "    return window.TimestampedValue(v, int(time.time()) + random.randint(0, 1))\n",
    "\n",
    "\n",
    "def modify_data3(kvpair):\n",
    "    # groupbyによりkeyとそのkeyを持つデータのリストのタプルが渡される\n",
    "    # windowで分割されているのでデータ数が少なくなる\n",
    "    # kvpair = (u'word only', [4, 4, 6, 6, 7])\n",
    "\n",
    "    return {'count_type': kvpair[0],\n",
    "            'sum': sum(kvpair[1])\n",
    "            }\n",
    "\n",
    "\n",
    "p7 = beam.Pipeline(options=options)\n",
    "\n",
    "query = 'SELECT * FROM [PROJECTID:testdataset.testtable3] LIMIT 20'\n",
    "(p7 | 'read' >> beam.io.Read(beam.io.BigQuerySource(project='PROJECTID', use_standard_sql=False, query=query))\n",
    "    | \"assign tv\" >> beam.Map(assign_timevalue)\n",
    "    | 'window' >> beam.WindowInto(beam.window.FixedWindows(1))\n",
    "    | 'pair' >> beam.Map(lambda x: (x['count_type'], x['word_count']))\n",
    "    | \"groupby\" >> beam.GroupByKey()\n",
    "    | 'modify' >> beam.Map(modify_data3)\n",
    "    | 'write' >> beam.io.Write(beam.io.BigQuerySink(\n",
    "        'testdataset.testtable5',\n",
    "        schema='count_type:STRING, sum:INTEGER',\n",
    "        create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,\n",
    "        write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE))\n",
    " )\n",
    "\n",
    "p7.run().wait_until_finish()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 終わり\n",
    "## DataflowRunnerに変えて実行してみよう\n",
    "また、`p*.run().wait_until_finish()` を `p*.run()`にしてください。`wait_until_finish()`が入っていると、かなり待つことになります！"
   ]
  }
 ],
 "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.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}