| { | |
| "nbformat": 4, | |
| "nbformat_minor": 0, | |
| "metadata": { | |
| "colab": { | |
| "name": "Getting started -- Tour of Beam", | |
| "provenance": [], | |
| "toc_visible": true, | |
| "include_colab_link": true | |
| }, | |
| "kernelspec": { | |
| "name": "python3", | |
| "display_name": "Python 3" | |
| } | |
| }, | |
| "cells": [ | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "view-in-github", | |
| "colab_type": "text" | |
| }, | |
| "source": [ | |
| "<a href=\"https://colab.research.google.com/github/apache/beam/blob/master/examples/notebooks/interactive-overview/getting-started.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>" | |
| ] | |
| }, | |
| { | |
| "cell_type": "code", | |
| "metadata": { | |
| "id": "SSKEd7tP-b2k", | |
| "cellView": "form" | |
| }, | |
| "source": [ | |
| "#@title ###### Licensed to the Apache Software Foundation (ASF), Version 2.0 (the \"License\")\n", | |
| "\n", | |
| "# Licensed to the Apache Software Foundation (ASF) under one\n", | |
| "# or more contributor license agreements. See the NOTICE file\n", | |
| "# distributed with this work for additional information\n", | |
| "# regarding copyright ownership. The ASF licenses this file\n", | |
| "# to you under the Apache License, Version 2.0 (the\n", | |
| "# \"License\"); you may not use this file except in compliance\n", | |
| "# with the License. You may obtain a copy of the License at\n", | |
| "#\n", | |
| "# http://www.apache.org/licenses/LICENSE-2.0\n", | |
| "#\n", | |
| "# Unless required by applicable law or agreed to in writing,\n", | |
| "# software distributed under the License is distributed on an\n", | |
| "# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n", | |
| "# KIND, either express or implied. See the License for the\n", | |
| "# specific language governing permissions and limitations\n", | |
| "# under the License." | |
| ], | |
| "execution_count": null, | |
| "outputs": [] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "5UC_aGanx6oE" | |
| }, | |
| "source": [ | |
| "# Getting started: _Tour of Beam_\n", | |
| "\n", | |
| "[Apache Beam](https://beam.apache.org/get-started/beam-overview/)\n", | |
| "is a library for parallel data processing.\n", | |
| "\n", | |
| "Beam is commonly used for\n", | |
| "[Extract-Transform-Load (ETL)](https://en.wikipedia.org/wiki/Extract,_transform,_load)\n", | |
| "jobs, where we _extract_ data from a data source, _transform_ that data, and _load_ it into a data sink like a database.\n", | |
| "It does particularly well with large amounts of data since it can use mutliple machines to process everything at the same time.\n", | |
| "\n", | |
| "Let's begin by installing the `apache-beam` package with `pip`." | |
| ] | |
| }, | |
| { | |
| "cell_type": "code", | |
| "metadata": { | |
| "id": "R_Yhoc6N_Flg" | |
| }, | |
| "source": [ | |
| "# Install apache-beam with pip.\n", | |
| "!pip install --quiet apache-beam" | |
| ], | |
| "execution_count": null, | |
| "outputs": [] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "WwxLB5EiVKP_" | |
| }, | |
| "source": [ | |
| "You can express a _data processing pipeline_, and then run it on the\n", | |
| "[_runner_ of your choice](https://beam.apache.org/documentation/runners/capability-matrix/).\n", | |
| "For now, we use the `DirectRunner` which runs locally for simplicity." | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "8rfQeLk2y9lx" | |
| }, | |
| "source": [ | |
| "# What is a _pipeline_?\n", | |
| "\n", | |
| "A __pipeline__ is a __sequence of data transformations__.\n", | |
| "You can think of it like a production line,\n", | |
| "data comes in from one end,\n", | |
| "it gets transformed by each step.\n", | |
| "The outputs from one step are passed as inputs to the next step.\n", | |
| "\n", | |
| "In Beam, your data lives in a __`PCollection`__,\n", | |
| "which stands for _Parallel Collection_.\n", | |
| "A `PCollection` is like a __list of elements__,\n", | |
| "but without any order guarantees.\n", | |
| "This allows Beam to easily parallelize and distribute\n", | |
| "the `PCollection`'s elements." | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "81VNY478gep_" | |
| }, | |
| "source": [ | |
| "" | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "LaljqKvs4UU5" | |
| }, | |
| "source": [ | |
| "Once you have your data, the next step is to transform it.\n", | |
| "In Beam, you transform data using **`PTransform`**s,\n", | |
| "which stands for _Parallel Transform_.\n", | |
| "A `PTransform` is like a __function__,\n", | |
| "they take some inputs, transform them and create some outputs." | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "Me_hvd6XgtPC" | |
| }, | |
| "source": [ | |
| "" | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "SVk4BsuKVlLj" | |
| }, | |
| "source": [ | |
| "Now let's dive into creating our first pipeline.\n", | |
| "\n", | |
| "For this first pipeline, let's just feed it some data from a Python list and print the results.\n", | |
| "\n", | |
| "Each _step_ in the pipeline is delimited by the _pipe operator_ `|`.\n", | |
| "The outputs of each transform are passed to the next transform as inputs.\n", | |
| "And we can save the final results into a `PCollection` variable.\n", | |
| "\n", | |
| "```py\n", | |
| "# We pass the elements from step1 through step3 and save the results into `outputs`.\n", | |
| "outputs = pipeline | step1 | step2 | step3\n", | |
| "```\n", | |
| "\n", | |
| "Pipelines can quickly grow long, so it's sometimes easier to read if we surround them with parentheses and break them into multiple lines.\n", | |
| "\n", | |
| "```py\n", | |
| "# This is equivalent to the example above.\n", | |
| "outputs = (\n", | |
| " pipeline\n", | |
| " | step1\n", | |
| " | step2\n", | |
| " | step3\n", | |
| ")\n", | |
| "```\n", | |
| "\n", | |
| "Also, Beam expects each transform, or step, to have a unique _label_, or description.\n", | |
| "This makes it a lot easier to debug, and it's in general a good practice to start.\n", | |
| "You can use the _right shift operator_ `>>` to add a label to your transforms, like `'My description' >> MyTransform`.\n", | |
| "\n", | |
| "```py\n", | |
| "# Try to give short but descriptive labels.\n", | |
| "# These serve both as comments and help debug later on.\n", | |
| "outputs = (\n", | |
| " pipeline\n", | |
| " | 'First step' >> step1\n", | |
| " | 'Second step' >> step2\n", | |
| " | 'Third step' >> step3\n", | |
| ")\n", | |
| "```\n", | |
| "\n", | |
| "> ℹ️ The syntax might seem a little different at first, but you'll become familiar with it.\n", | |
| "\n", | |
| "We use the `Create` transform to feed the pipeline with an\n", | |
| "[`iterable`](https://docs.python.org/3/glossary.html#term-iterable)\n", | |
| "of elements, like a `list`.\n", | |
| "\n", | |
| "Let's try to see what happens if we try to `print` a PCollection." | |
| ] | |
| }, | |
| { | |
| "cell_type": "code", | |
| "metadata": { | |
| "id": "XAke4d5lV5f5", | |
| "outputId": "eaf5954c-3a66-4ef7-a258-35797ce0bbb9", | |
| "colab": { | |
| "base_uri": "https://localhost:8080/" | |
| } | |
| }, | |
| "source": [ | |
| "import apache_beam as beam\n", | |
| "\n", | |
| "inputs = [0, 1, 2, 3]\n", | |
| "\n", | |
| "# Create a pipeline.\n", | |
| "with beam.Pipeline() as pipeline:\n", | |
| " # Feed it some input elements with `Create`.\n", | |
| " outputs = (\n", | |
| " pipeline\n", | |
| " | 'Create initial values' >> beam.Create(inputs)\n", | |
| " )\n", | |
| "\n", | |
| " # `outputs` is a PCollection with our input elements.\n", | |
| " # But printing it directly won't show us its contents :(\n", | |
| " print(f\"outputs: {outputs}\")" | |
| ], | |
| "execution_count": 3, | |
| "outputs": [ | |
| { | |
| "output_type": "stream", | |
| "text": [ | |
| "outputs: PCollection[[3]: Create initial values/Map(decode).None]\n" | |
| ], | |
| "name": "stdout" | |
| } | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "h0UUmpwRADqA" | |
| }, | |
| "source": [ | |
| "> ℹ️ In Beam, you can __NOT__ access the elements from a `PCollection` directly like a Python list.\n", | |
| "> This means, we can't simply `print` the output `PCollection` to see the elements.\n", | |
| ">\n", | |
| "> This is because, depending on the runner,\n", | |
| "> the `PCollection` elements might live in multiple worker machines.\n", | |
| "\n", | |
| "To print the elements in the PCollection, we'll do a little trick, but we'll explain it shortly." | |
| ] | |
| }, | |
| { | |
| "cell_type": "code", | |
| "metadata": { | |
| "id": "IblMlbE8_4CJ", | |
| "outputId": "dc825083-d157-4fae-fff9-4db33dbc0876", | |
| "colab": { | |
| "base_uri": "https://localhost:8080/" | |
| } | |
| }, | |
| "source": [ | |
| "import apache_beam as beam\n", | |
| "\n", | |
| "inputs = [0, 1, 2, 3]\n", | |
| "\n", | |
| "with beam.Pipeline() as pipeline:\n", | |
| " outputs = (\n", | |
| " pipeline\n", | |
| " | 'Create initial values' >> beam.Create(inputs)\n", | |
| " )\n", | |
| "\n", | |
| " # We can only access the elements through another transform.\n", | |
| " # Don't worry if you don't know what's happening here,\n", | |
| " # we'll get to it just next :)\n", | |
| " outputs | beam.Map(print)" | |
| ], | |
| "execution_count": 4, | |
| "outputs": [ | |
| { | |
| "output_type": "stream", | |
| "text": [ | |
| "0\n", | |
| "1\n", | |
| "2\n", | |
| "3\n" | |
| ], | |
| "name": "stdout" | |
| } | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "FY9TP3Tw5eZt" | |
| }, | |
| "source": [ | |
| "# Transforming data\n", | |
| "\n", | |
| "Apache Beam is designed with a [functional paradigm](https://en.wikipedia.org/wiki/Functional_programming).\n", | |
| "This means that, instead of _loops_, it uses `PTransform`s alongside with _functions_ to process each element in a `PCollection`.\n", | |
| "\n", | |
| "Let's go through some of the most common and basic data transforms in Beam." | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "UMKgi9LD6-xb" | |
| }, | |
| "source": [ | |
| "## Map: _one-to-one_\n", | |
| "\n", | |
| "Let's say we have some elements and we want to do something with each element.\n", | |
| "\n", | |
| "We want to `map` a function to each element of the collection.\n", | |
| "\n", | |
| "`map` takes a _function_ that transforms a single input `a` into a single output `b`.\n", | |
| "\n", | |
| "> ℹ️ For example, we want to multiply each element by 2." | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "tDtzHrSdjQx8" | |
| }, | |
| "source": [ | |
| "" | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "G0NJrjja9YRF" | |
| }, | |
| "source": [ | |
| "In Python, this is commonly done with the\n", | |
| "[built-in `map` function](https://docs.python.org/3/library/functions.html#map), or with\n", | |
| "[list comprehensions](https://docs.python.org/3/tutorial/datastructures.html#list-comprehensions)." | |
| ] | |
| }, | |
| { | |
| "cell_type": "code", | |
| "metadata": { | |
| "id": "MHzR7CuZqcQq", | |
| "outputId": "295e24fb-506e-4080-ee04-755ae533888b", | |
| "colab": { | |
| "base_uri": "https://localhost:8080/" | |
| } | |
| }, | |
| "source": [ | |
| "inputs = [0, 1, 2, 3]\n", | |
| "\n", | |
| "# Using the `map` function.\n", | |
| "outputs = map(lambda x: x * 2, inputs)\n", | |
| "print(list(outputs))\n", | |
| "\n", | |
| "# Using a list comprehension.\n", | |
| "outputs = [x * 2 for x in inputs]\n", | |
| "print(outputs)\n", | |
| "\n", | |
| "# Roughly equivalent for loop.\n", | |
| "outputs = []\n", | |
| "for x in inputs:\n", | |
| " outputs.append(x * 2)\n", | |
| "print(outputs)" | |
| ], | |
| "execution_count": 5, | |
| "outputs": [ | |
| { | |
| "output_type": "stream", | |
| "text": [ | |
| "[0, 2, 4, 6]\n", | |
| "[0, 2, 4, 6]\n", | |
| "[0, 2, 4, 6]\n" | |
| ], | |
| "name": "stdout" | |
| } | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "4vaBuNFW-5FG" | |
| }, | |
| "source": [ | |
| "In Beam, there is the\n", | |
| "[`Map` transform](https://beam.apache.org/documentation/transforms/python/elementwise/map/),\n", | |
| "but we must use it within a pipeline.\n", | |
| "\n", | |
| "First we create a pipeline and feed it our input elements.\n", | |
| "Then we _pipe_ those elements into a `Map` transform where we apply our function." | |
| ] | |
| }, | |
| { | |
| "cell_type": "code", | |
| "metadata": { | |
| "id": "LoYq871Q96iu", | |
| "outputId": "57c2155f-cebb-42ae-e175-e1554f2f806b", | |
| "colab": { | |
| "base_uri": "https://localhost:8080/" | |
| } | |
| }, | |
| "source": [ | |
| "import apache_beam as beam\n", | |
| "\n", | |
| "inputs = [0, 1, 2, 3]\n", | |
| "\n", | |
| "with beam.Pipeline() as pipeline:\n", | |
| " outputs = (\n", | |
| " pipeline\n", | |
| " | 'Create values' >> beam.Create(inputs)\n", | |
| " | 'Multiply by 2' >> beam.Map(lambda x: x * 2)\n", | |
| " )\n", | |
| "\n", | |
| " outputs | beam.Map(print)" | |
| ], | |
| "execution_count": null, | |
| "outputs": [ | |
| { | |
| "output_type": "stream", | |
| "text": [ | |
| "0\n", | |
| "2\n", | |
| "4\n", | |
| "6\n" | |
| ], | |
| "name": "stdout" | |
| } | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "Q06jvwmqOzer" | |
| }, | |
| "source": [ | |
| "> ℹ️ Now that we know how `Map` works, we can see what's happening when we print the elements.\n", | |
| ">\n", | |
| "> We have our outputs stored in the `outputs` `PCollection`, so we _pipe_ it to a `Map` transform to apply the\n", | |
| "> [`print`](https://docs.python.org/3/library/functions.html#print)\n", | |
| "> function.\n", | |
| ">\n", | |
| "> Note that `print` returns `None`, so we get an output `PCollection` of all `None` elements.\n", | |
| "> But we are not saving its results to any variable,\n", | |
| "> so they get discarded.\n", | |
| ">\n", | |
| "> This does _not_ affect the values in `outputs` in any way." | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "m8AVwzVyCAPD" | |
| }, | |
| "source": [ | |
| "## FlatMap: _one-to-many_\n", | |
| "\n", | |
| "`Map` allows us to transform each individual element,\n", | |
| "but we can't change the number of elements with it.\n", | |
| "\n", | |
| "We want to `map` a function to each element of a collection.\n", | |
| "That function returns a _list of output elements_,\n", | |
| "so we would get a _list of lists of elements_.\n", | |
| "Then we want to _flatten_ the _list of lists_ into a single _list_.\n", | |
| "\n", | |
| "`flatMap` takes a function that transforms a single input `a` into an `iterable` of outputs `b`.\n", | |
| "But we get a _single collection_ containing the outputs of _all_ the elements.\n", | |
| "\n", | |
| "> ℹ️ For example, we want to have as many elements as the element's value.\n", | |
| "> For a value `1` we want one element, and three elements for a value `3`." | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "t3KpC-tej1CS" | |
| }, | |
| "source": [ | |
| "" | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "QdQEEbtNKpQJ" | |
| }, | |
| "source": [ | |
| "In Python this could be done with a _nested list comprehension_,\n", | |
| "but it's a little tricky to read." | |
| ] | |
| }, | |
| { | |
| "cell_type": "code", | |
| "metadata": { | |
| "id": "0xqSUP5mKNqI", | |
| "outputId": "58251379-ed6f-4452-e9ea-60e59394ee5a", | |
| "colab": { | |
| "base_uri": "https://localhost:8080/" | |
| } | |
| }, | |
| "source": [ | |
| "inputs = [0, 1, 2, 3]\n", | |
| "\n", | |
| "# Using a list comprehensions.\n", | |
| "mapOutputs = [[x for _ in range(x)] for x in inputs]\n", | |
| "# After the map function, flatten the results.\n", | |
| "outputs = [x for xs in mapOutputs for x in xs]\n", | |
| "print(outputs)\n", | |
| "\n", | |
| "# Roughly equivalent for loop.\n", | |
| "outputs = []\n", | |
| "for x in inputs:\n", | |
| " outputs += [x for _ in range(x)]\n", | |
| "print(outputs)" | |
| ], | |
| "execution_count": null, | |
| "outputs": [ | |
| { | |
| "output_type": "stream", | |
| "text": [ | |
| "[1, 2, 2, 3, 3, 3]\n", | |
| "[1, 2, 2, 3, 3, 3]\n" | |
| ], | |
| "name": "stdout" | |
| } | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "OP3bSQoYMOH6" | |
| }, | |
| "source": [ | |
| "The good news is that Beam already has a\n", | |
| "[`FlatMap` transform](https://beam.apache.org/documentation/transforms/python/elementwise/flatmap/)\n", | |
| "built-in, so it's actually easier than plain Python.\n", | |
| "\n", | |
| "`FlatMap` accepts a function that takes a single input element and outputs an `iterable` of elements." | |
| ] | |
| }, | |
| { | |
| "cell_type": "code", | |
| "metadata": { | |
| "id": "gZbsyAWaMT-F", | |
| "outputId": "f2623c5c-d709-4229-c293-952b9bbcbef4", | |
| "colab": { | |
| "base_uri": "https://localhost:8080/" | |
| } | |
| }, | |
| "source": [ | |
| "import apache_beam as beam\n", | |
| "\n", | |
| "inputs = [0, 1, 2, 3]\n", | |
| "\n", | |
| "with beam.Pipeline() as pipeline:\n", | |
| " outputs = (\n", | |
| " pipeline\n", | |
| " | 'Create values' >> beam.Create(inputs)\n", | |
| " | 'Expand elements' >> beam.FlatMap(lambda x: [x for _ in range(x)])\n", | |
| " )\n", | |
| "\n", | |
| " outputs | beam.Map(print)" | |
| ], | |
| "execution_count": null, | |
| "outputs": [ | |
| { | |
| "output_type": "stream", | |
| "text": [ | |
| "1\n", | |
| "2\n", | |
| "2\n", | |
| "3\n", | |
| "3\n", | |
| "3\n" | |
| ], | |
| "name": "stdout" | |
| } | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "YHcWp18cMefc" | |
| }, | |
| "source": [ | |
| "> ℹ️ Try replacing the `FlatMap` transform with `Map` to see how they behave differently." | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "ad-O7lUxOoyD" | |
| }, | |
| "source": [ | |
| "## Filter: _one-to-zero_\n", | |
| "\n", | |
| "Sometimes we want to *only* process certain elements while ignoring others.\n", | |
| "\n", | |
| "We want to `filter` each element in a collection using a function.\n", | |
| "\n", | |
| "`filter` takes a function that checks a single element `a`,\n", | |
| "and returns `True` to keep the element, or `False` to discard it.\n", | |
| "\n", | |
| "> ℹ️ For example, we only want to keep number that are *even*, or divisible by two.\n", | |
| "> We can use the\n", | |
| "> [modulo operator `%`](https://en.wikipedia.org/wiki/Modulo_operation)\n", | |
| "> for a simple check." | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "CUVXfvixkckY" | |
| }, | |
| "source": [ | |
| "" | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "Y-gWtFNockwq" | |
| }, | |
| "source": [ | |
| "In Python we can do this with *list comprehensions* as well." | |
| ] | |
| }, | |
| { | |
| "cell_type": "code", | |
| "metadata": { | |
| "id": "4xXSy5Bga9td", | |
| "outputId": "90d5d9a7-d925-4b11-91c8-e6e844f9d21e", | |
| "colab": { | |
| "base_uri": "https://localhost:8080/" | |
| } | |
| }, | |
| "source": [ | |
| "inputs = [0, 1, 2, 3]\n", | |
| "\n", | |
| "# Using a list comprehensions.\n", | |
| "outputs = [x for x in inputs if x % 2 == 0]\n", | |
| "print(outputs)\n", | |
| "\n", | |
| "# Roughly equivalent for loop.\n", | |
| "outputs = []\n", | |
| "for x in inputs:\n", | |
| " if x % 2 == 0:\n", | |
| " outputs.append(x)\n", | |
| "print(outputs)" | |
| ], | |
| "execution_count": null, | |
| "outputs": [ | |
| { | |
| "output_type": "stream", | |
| "text": [ | |
| "[0, 2]\n", | |
| "[0, 2]\n" | |
| ], | |
| "name": "stdout" | |
| } | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "mtVSijNAc0Fi" | |
| }, | |
| "source": [ | |
| "In Beam, there is the\n", | |
| "[`Filter` transform](https://beam.apache.org/documentation/transforms/python/elementwise/filter/)." | |
| ] | |
| }, | |
| { | |
| "cell_type": "code", | |
| "metadata": { | |
| "id": "l7W9UafLkiO1", | |
| "outputId": "a59e0cb0-3cb9-43a6-8043-5789380d938c", | |
| "colab": { | |
| "base_uri": "https://localhost:8080/" | |
| } | |
| }, | |
| "source": [ | |
| "import apache_beam as beam\n", | |
| "\n", | |
| "inputs = [0, 1, 2, 3]\n", | |
| "\n", | |
| "with beam.Pipeline() as pipeline:\n", | |
| " outputs = (\n", | |
| " pipeline\n", | |
| " | 'Create values' >> beam.Create(inputs)\n", | |
| " | 'Keep only even numbers' >> beam.Filter(lambda x: x % 2 == 0)\n", | |
| " )\n", | |
| "\n", | |
| " outputs | beam.Map(print)" | |
| ], | |
| "execution_count": null, | |
| "outputs": [ | |
| { | |
| "output_type": "stream", | |
| "text": [ | |
| "0\n", | |
| "2\n" | |
| ], | |
| "name": "stdout" | |
| } | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "_yYNTszgktLY" | |
| }, | |
| "source": [ | |
| "## Combine: _many-to-one_\n", | |
| "\n", | |
| "We also need a way to get a single value from an entire `PCollection`.\n", | |
| "We might want to get the total number of elements, or the average value, or any other type of _aggregation_ of values.\n", | |
| "\n", | |
| "We want to `combine` the elements in a collection into a single output.\n", | |
| "\n", | |
| "`combine` takes a function that transforms an `iterable` of inputs `a`, and returns a single output `a`.\n", | |
| "\n", | |
| "Other common names for this function are `fold` and `reduce`.\n", | |
| "\n", | |
| "> ℹ️ For example, we want to add all numbers together." | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "Too-Ru1xGj7e" | |
| }, | |
| "source": [ | |
| "" | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "F60IfKPOmlkg" | |
| }, | |
| "source": [ | |
| "In Python this is usually achieved with the\n", | |
| "[`reduce` function](https://docs.python.org/3/library/functools.html#functools.reduce)." | |
| ] | |
| }, | |
| { | |
| "cell_type": "code", | |
| "metadata": { | |
| "id": "4qjUZvT2k878", | |
| "outputId": "1868d167-529e-4e5f-ec27-3ba405226a1d", | |
| "colab": { | |
| "base_uri": "https://localhost:8080/" | |
| } | |
| }, | |
| "source": [ | |
| "from functools import reduce\n", | |
| "\n", | |
| "inputs = [0, 1, 2, 3]\n", | |
| "\n", | |
| "# Using reduce (most general way).\n", | |
| "output = reduce(lambda x, y: x + y, inputs, 0)\n", | |
| "print(output)\n", | |
| "\n", | |
| "# Using the built-in sum function, which is itself a \"reduce\" function.\n", | |
| "output = sum(inputs)\n", | |
| "print(output)\n", | |
| "\n", | |
| "# Roughly equivalent for loop.\n", | |
| "output = 0\n", | |
| "for x in inputs:\n", | |
| " y = output\n", | |
| " output = x + y\n", | |
| "print(output)" | |
| ], | |
| "execution_count": null, | |
| "outputs": [ | |
| { | |
| "output_type": "stream", | |
| "text": [ | |
| "6\n", | |
| "6\n", | |
| "6\n" | |
| ], | |
| "name": "stdout" | |
| } | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "kotSZnlZoO8D" | |
| }, | |
| "source": [ | |
| "In Beam, there are\n", | |
| "[aggregation transforms](https://beam.apache.org/documentation/transforms/python/overview/#aggregation).\n", | |
| "\n", | |
| "For this particular example, we can use the\n", | |
| "[`CombineGlobally` transform](https://beam.apache.org/documentation/transforms/python/aggregation/sum/)\n", | |
| "which accepts a function that takes an iterable of elements as an input and outputs a single value.\n", | |
| "\n", | |
| "We can pass the\n", | |
| "[built-in function `sum`](https://docs.python.org/3/library/functions.html#sum)\n", | |
| "into `CombineGlobally`." | |
| ] | |
| }, | |
| { | |
| "cell_type": "code", | |
| "metadata": { | |
| "id": "PoRd7hlnoOu5", | |
| "outputId": "b34f987f-ddb3-4808-95eb-9ac6140d2ea2", | |
| "colab": { | |
| "base_uri": "https://localhost:8080/" | |
| } | |
| }, | |
| "source": [ | |
| "import apache_beam as beam\n", | |
| "\n", | |
| "inputs = [0, 1, 2, 3]\n", | |
| "\n", | |
| "with beam.Pipeline() as pipeline:\n", | |
| " outputs = (\n", | |
| " pipeline\n", | |
| " | 'Create values' >> beam.Create(inputs)\n", | |
| " | 'Sum all values together' >> beam.CombineGlobally(sum)\n", | |
| " )\n", | |
| "\n", | |
| " outputs | beam.Map(print)" | |
| ], | |
| "execution_count": null, | |
| "outputs": [ | |
| { | |
| "output_type": "stream", | |
| "text": [ | |
| "6\n" | |
| ], | |
| "name": "stdout" | |
| } | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "pFb98ioSp9YU" | |
| }, | |
| "source": [ | |
| "> ℹ️ There are many ways to combine values in Beam.\n", | |
| "> You could even combine them into a different data type by defining a custom `CombineFn`.\n", | |
| ">\n", | |
| "> You can learn more about them by checking the available\n", | |
| "> [aggregation transforms](https://beam.apache.org/documentation/transforms/python/overview/#aggregation)." | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "cOnJoaE3qmMv" | |
| }, | |
| "source": [ | |
| "## GroupByKey: _group related elements_\n", | |
| "\n", | |
| "Sometimes it's useful to pair each element with a *key* that we can use to group related elements together.\n", | |
| "\n", | |
| "Think of it as creating a\n", | |
| "[Python `dict`](https://docs.python.org/3/tutorial/datastructures.html#dictionaries)\n", | |
| "from a list of `(key, value)` pairs,\n", | |
| "but instead of replacing the value on a \"duplicate\" key,\n", | |
| "you would get a list of all the values associated with that key.\n", | |
| "\n", | |
| "> ℹ️ For example, we want to group each animal with the list of foods they like, and we start with `(animal, food)` pairs." | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "IX9VmTH7xluM" | |
| }, | |
| "source": [ | |
| "" | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "Sic8de3btYli" | |
| }, | |
| "source": [ | |
| "There's no built-in function for `groupByKey` in plain Python,\n", | |
| "but here's a simple implementation." | |
| ] | |
| }, | |
| { | |
| "cell_type": "code", | |
| "metadata": { | |
| "id": "EzJPZJCWtWF9", | |
| "outputId": "cd972c46-7c05-49ab-a57d-e6e655349544", | |
| "colab": { | |
| "base_uri": "https://localhost:8080/" | |
| } | |
| }, | |
| "source": [ | |
| "from functools import reduce\n", | |
| "\n", | |
| "inputs = [\n", | |
| " ('🐹', '🌽'),\n", | |
| " ('🐼', '🎋'),\n", | |
| " ('🐰', '🥕'),\n", | |
| " ('🐹', '🌰'),\n", | |
| " ('🐰', '🥒'),\n", | |
| "]\n", | |
| "\n", | |
| "\n", | |
| "# Since we're getting a single dict from all the elements,\n", | |
| "# we can use reduce for this.\n", | |
| "def groupByKey(result, keyValue):\n", | |
| " key, value = keyValue\n", | |
| " values = result.get(key, []) + [value]\n", | |
| " return {**result, key: values}\n", | |
| "output = reduce(groupByKey, inputs, {})\n", | |
| "print(output)\n", | |
| "\n", | |
| "\n", | |
| "# Roughly equivalent for loop.\n", | |
| "output = {}\n", | |
| "for key, value in inputs:\n", | |
| " values = output.get(key, []) + [value]\n", | |
| " output = {**output, key: values}\n", | |
| "print(output)" | |
| ], | |
| "execution_count": null, | |
| "outputs": [ | |
| { | |
| "output_type": "stream", | |
| "text": [ | |
| "{'🐹': ['🌽', '🌰'], '🐼': ['🎋'], '🐰': ['🥕', '🥒']}\n", | |
| "{'🐹': ['🌽', '🌰'], '🐼': ['🎋'], '🐰': ['🥕', '🥒']}\n" | |
| ], | |
| "name": "stdout" | |
| } | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "1EmT-9Zawa_q" | |
| }, | |
| "source": [ | |
| "In Beam, there is the\n", | |
| "[`GroupByKey` transform](https://beam.apache.org/documentation/transforms/python/aggregation/groupbykey/)." | |
| ] | |
| }, | |
| { | |
| "cell_type": "code", | |
| "metadata": { | |
| "id": "67fZUiWiwnAf", | |
| "outputId": "f95c455c-c554-41de-bc8c-e755584064d2", | |
| "colab": { | |
| "base_uri": "https://localhost:8080/" | |
| } | |
| }, | |
| "source": [ | |
| "import apache_beam as beam\n", | |
| "\n", | |
| "inputs = [\n", | |
| " ('🐹', '🌽'),\n", | |
| " ('🐼', '🎋'),\n", | |
| " ('🐰', '🥕'),\n", | |
| " ('🐹', '🌰'),\n", | |
| " ('🐰', '🥒'),\n", | |
| "]\n", | |
| "\n", | |
| "with beam.Pipeline() as pipeline:\n", | |
| " outputs = (\n", | |
| " pipeline\n", | |
| " | 'Create (animal, food) pairs' >> beam.Create(inputs)\n", | |
| " | 'Group foods by animals' >> beam.GroupByKey()\n", | |
| " )\n", | |
| "\n", | |
| " outputs | beam.Map(print)" | |
| ], | |
| "execution_count": null, | |
| "outputs": [ | |
| { | |
| "output_type": "stream", | |
| "text": [ | |
| "('🐹', ['🌽', '🌰'])\n", | |
| "('🐼', ['🎋'])\n", | |
| "('🐰', ['🥕', '🥒'])\n" | |
| ], | |
| "name": "stdout" | |
| } | |
| ] | |
| }, | |
| { | |
| "cell_type": "markdown", | |
| "metadata": { | |
| "id": "gnoz_mWtxSjW" | |
| }, | |
| "source": [ | |
| "# What's next?\n", | |
| "\n", | |
| "* \n", | |
| " [Reading and writing data](https://colab.research.google.com/github/apache/beam/blob/master/examples/notebooks/interactive-overview/reading-and-writing-data.ipynb) --\n", | |
| " how to read and write data to and from different data formats. \n", | |
| "* [Transform catalog](https://beam.apache.org/documentation/transforms/python/overview) --\n", | |
| " check out all the available transforms.\n", | |
| "* [Mobile gaming example](https://beam.apache.org/get-started/mobile-gaming-example) --\n", | |
| " learn more about windowing, triggers, and streaming through a complete example pipeline.\n", | |
| "* [Runners](https://beam.apache.org/documentation/runners/capability-matrix) --\n", | |
| " check the available runners, their capabilities, and how to run your pipeline in them." | |
| ] | |
| } | |
| ] | |
| } |