average-prices-by-product.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "pycharm": {
     "is_executing": false
    }
   },
   "outputs": [],
   "source": [
    "import logging\n",
    "import apache_beam as beam\n",
    "\n",
    "from apache_beam.options.pipeline_options import PipelineOptions\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "pycharm": {
     "is_executing": false,
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "# logging\n",
    "root = logging.getLogger()\n",
    "root.setLevel(logging.INFO)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "pycharm": {
     "is_executing": false,
     "name": "#%%\n"
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "text": [
      "INFO:root:Missing pipeline option (runner). Executing pipeline using the default runner: DirectRunner.\n",
      "INFO:oauth2client.client:Timeout attempting to reach GCE metadata service.\n",
      "WARNING:root:Unable to find default credentials to use: The Application Default Credentials are not available. They are available if running in Google Compute Engine. Otherwise, the environment variable GOOGLE_APPLICATION_CREDENTIALS must be defined pointing to a file defining the credentials. See https://developers.google.com/accounts/docs/application-default-credentials for more information.\nConnecting anonymously.\n",
      "INFO:oauth2client.client:Timeout attempting to reach GCE metadata service.\n",
      "WARNING:root:Unable to find default credentials to use: The Application Default Credentials are not available. They are available if running in Google Compute Engine. Otherwise, the environment variable GOOGLE_APPLICATION_CREDENTIALS must be defined pointing to a file defining the credentials. See https://developers.google.com/accounts/docs/application-default-credentials for more information.\nConnecting anonymously.\n"
     ],
     "output_type": "stream"
    },
    {
     "data": {
      "text/plain": "<PCollection[Raw transactions count.None] at 0x119881c10>"
     },
     "metadata": {},
     "output_type": "execute_result",
     "execution_count": 36
    }
   ],
   "source": [
    "# create a pipeline, read data from csv\n",
    "pipeline_options = beam.Pipeline(options=PipelineOptions())\n",
    "\n",
    "# PCollection format is = id, customer-type, product-type, price\n",
    "transactions = (pipeline_options\n",
    "                | \"Read Transaction CSV\"\n",
    "                >> beam.io.ReadFromText(\"gs://spark-dataset-1/datasets/sales/sales_transactions.csv\")\n",
    "                )\n",
    "\n",
    "(transactions\n",
    "     | \"Raw transactions\"\n",
    "     >> beam.CombineGlobally(beam.combiners.CountCombineFn())\n",
    "     | \"Raw transactions count\"\n",
    "     >> beam.ParDo(lambda c: logging.info(\"Total lines %s\", str(c)))\n",
    ")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "pycharm": {
     "is_executing": false,
     "name": "#%%\n"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": "<PCollection[Product Type and Price count.None] at 0x119898ed0>"
     },
     "metadata": {},
     "output_type": "execute_result",
     "execution_count": 37
    }
   ],
   "source": [
    "# extract product-type and price\n",
    "class ExtractProductTypePrice(beam.DoFn):\n",
    "\n",
    "    def process(self, element, *args, **kwargs):\n",
    "        line = element.split(',')\n",
    "        return [(line[2], float(line[3]))]\n",
    "\n",
    "\n",
    "# PCollection format now = product-type, price\n",
    "productType_price = (transactions\n",
    "                     | \"Extracting product type and price\"\n",
    "                     >> beam.ParDo(ExtractProductTypePrice())          \n",
    "                     )\n",
    "\n",
    "(productType_price\n",
    " | \"Product Type and Price\"\n",
    " >> beam.CombineGlobally(beam.combiners.CountCombineFn())\n",
    " | \"Product Type and Price count\"\n",
    " >> beam.ParDo(lambda c: logging.info(\"Total lines %s\", str(c)))  \n",
    ")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "pycharm": {
     "is_executing": false,
     "name": "#%%\n"
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "text": [
      "INFO:root:==================== <function annotate_downstream_side_inputs at 0x11712a230> ====================\n",
      "INFO:root:==================== <function fix_side_input_pcoll_coders at 0x11712a2a8> ====================\n",
      "INFO:root:==================== <function lift_combiners at 0x11712a9b0> ====================\n",
      "INFO:root:==================== <function expand_gbk at 0x11712a410> ====================\n",
      "INFO:root:==================== <function sink_flattens at 0x11712a320> ====================\n",
      "INFO:root:==================== <function greedily_fuse at 0x11712a758> ====================\n",
      "INFO:root:==================== <function impulse_to_input at 0x11712a668> ====================\n",
      "INFO:root:==================== <function inject_timer_pcollections at 0x1173c0668> ====================\n",
      "INFO:root:==================== <function sort_stages at 0x1173c06e0> ====================\n",
      "INFO:root:Running ((((ref_AppliedPTransform_Read Transaction CSV/Read_3)+((ref_AppliedPTransform_Raw transactions/KeyWithVoid_5)+(Raw transactions/CombinePerKey/Precombine)))+(ref_AppliedPTransform_Extracting product type and price_18))+((ref_AppliedPTransform_Product Type and Price/KeyWithVoid_20)+((Product Type and Price/CombinePerKey/Precombine)+(Product Type and Price/CombinePerKey/Group/Write))))+(Raw transactions/CombinePerKey/Group/Write)\n",
      "INFO:oauth2client.client:Timeout attempting to reach GCE metadata service.\n",
      "WARNING:root:Unable to find default credentials to use: The Application Default Credentials are not available. They are available if running in Google Compute Engine. Otherwise, the environment variable GOOGLE_APPLICATION_CREDENTIALS must be defined pointing to a file defining the credentials. See https://developers.google.com/accounts/docs/application-default-credentials for more information.\nConnecting anonymously.\n",
      "INFO:oauth2client.client:Timeout attempting to reach GCE metadata service.\n",
      "WARNING:root:Unable to find default credentials to use: The Application Default Credentials are not available. They are available if running in Google Compute Engine. Otherwise, the environment variable GOOGLE_APPLICATION_CREDENTIALS must be defined pointing to a file defining the credentials. See https://developers.google.com/accounts/docs/application-default-credentials for more information.\nConnecting anonymously.\n",
      "INFO:oauth2client.client:Timeout attempting to reach GCE metadata service.\n",
      "WARNING:root:Unable to find default credentials to use: The Application Default Credentials are not available. They are available if running in Google Compute Engine. Otherwise, the environment variable GOOGLE_APPLICATION_CREDENTIALS must be defined pointing to a file defining the credentials. See https://developers.google.com/accounts/docs/application-default-credentials for more information.\nConnecting anonymously.\n",
      "INFO:root:Running ((Raw transactions/CombinePerKey/Group/Read)+(Raw transactions/CombinePerKey/Merge))+((Raw transactions/CombinePerKey/ExtractOutputs)+((ref_AppliedPTransform_Raw transactions/UnKey_13)+(ref_PCollection_PCollection_7/Write)))\n",
      "INFO:root:Running (((Product Type and Price/CombinePerKey/Group/Read)+(Product Type and Price/CombinePerKey/Merge))+(Product Type and Price/CombinePerKey/ExtractOutputs))+((ref_AppliedPTransform_Product Type and Price/UnKey_28)+(ref_PCollection_PCollection_17/Write))\n",
      "INFO:root:Running (ref_AppliedPTransform_Product Type and Price/DoOnce/Read_30)+((ref_AppliedPTransform_Product Type and Price/InjectDefault_31)+(ref_AppliedPTransform_Product Type and Price count_32))\n",
      "INFO:root:Total lines 100\n",
      "INFO:root:Running (ref_AppliedPTransform_Raw transactions/DoOnce/Read_15)+((ref_AppliedPTransform_Raw transactions/InjectDefault_16)+(ref_AppliedPTransform_Raw transactions count_17))\n",
      "INFO:root:Total lines 100\n"
     ],
     "output_type": "stream"
    },
    {
     "data": {
      "text/plain": "'DONE'"
     },
     "metadata": {},
     "output_type": "execute_result",
     "execution_count": 38
    }
   ],
   "source": [
    "# run\n",
    "result = pipeline_options.run()\n",
    "\n",
    "result.wait_until_finish()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "\n"
   ]
  }
 ],
 "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.15"
  },
  "pycharm": {
   "stem_cell": {
    "cell_type": "raw",
    "metadata": {
     "collapsed": false
    },
    "source": []
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}