diff --git a/examples/notebooks/get-started/try-apache-beam-yaml.ipynb b/examples/notebooks/get-started/try-apache-beam-yaml.ipynb new file mode 100644 index 000000000000..b18b318e85ea --- /dev/null +++ b/examples/notebooks/get-started/try-apache-beam-yaml.ipynb @@ -0,0 +1,734 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": { + "colab_type": "text", + "id": "view-in-github" + }, + "source": [ + "\"Open\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": { + "cellView": "form" + }, + "outputs": [], + "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." + ] + }, + { + "cell_type": "markdown", + "metadata": { + "colab_type": "text", + "id": "lNKIMlEDZ_Vw" + }, + "source": [ + "# Try Apache Beam - YAML\n", + "\n", + "While Beam provides powerful APIs for authoring sophisticated data processing pipelines, it still has a high barrier for getting started and authoring simple pipelines. Even setting up the environment, installing the dependencies, and setting up the project can be a challenge.\n", + "\n", + "Here we provide a simple declarative syntax for describing pipelines that does not require coding experience or learning how to use an SDK—any text editor will do. Some installation may be required to actually *execute* a pipeline, but we envision various services (such as Dataflow) to accept yaml pipelines directly obviating the need for even that in the future. We also anticipate the ability to generate code directly from these higher-level yaml descriptions, should one want to graduate to a full Beam SDK (and possibly the other direction as well as far as possible).\n", + "\n", + "It should be noted that everything here is still under development, but any features already included are considered stable. Feedback is welcome at dev@apache.beam.org.\n", + "\n", + "In this notebook, you set up your development environment and write a simple pipeline using YAML. Then you run it locally, using the [DirectRunner](https://beam.apache.org/documentation/runners/direct/). You can explore other runners with the [Beam Capability Matrix](https://beam.apache.org/documentation/runners/capability-matrix/).\n", + "\n", + "To navigate through different sections, use the table of contents. From **View** drop-down list, select **Table of contents**.\n", + "\n", + "To run a code cell, click the **Run cell** button at the top left of the cell, or select it and press **`Shift+Enter`**. Try modifying a code cell and re-running it to see what happens.\n", + "\n", + "To learn more about Colab, see [Welcome to Colaboratory!](https://colab.sandbox.google.com/notebooks/welcome.ipynb)." + ] + }, + { + "cell_type": "markdown", + "metadata": { + "colab_type": "text", + "id": "Fz6KSQ13_3Rr" + }, + "source": [ + "# Setup\n", + "\n", + "First, you need to set up your environment. The following code installs `apache-beam` and creates directories for your data, pipelines and results." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": { + "colab": { + "base_uri": "https://localhost:8080/", + "height": 170 + }, + "colab_type": "code", + "id": "GOOk81Jj_yUy", + "outputId": "d283dfb2-4f51-4fec-816b-f57b0cb9b71c" + }, + "outputs": [], + "source": [ + "# Install apache-beam\n", + "! pip install --quiet apache-beam[yaml]\n", + "\n", + "# Create directories for storing data, pipelines and results\n", + "! mkdir -p data\n", + "! mkdir -p pipelines\n", + "! mkdir -p results" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "We'll also create an artificial dataset that represents a simple database. The csv data contains information about different people. Each line represents a single person and their details are separated by commas. The file contains 5 columns: id, firstname, age, country and a profession." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%%writefile 'data/people.csv'\n", + "id,firstname,age,country,profession\n", + "1,Reeba,58,Belgium,unemployed\n", + "2,Maud,45,Spain,firefighter\n", + "3,Meg,11,France,unemployed\n", + "4,Rani,53,Spain,doctor\n", + "5,Natka,26,France,doctor\n", + "6,Aurore,32,Italy,police officer\n", + "7,Elvira,39,Italy,doctor\n", + "8,Asia,10,Belgium,doctor\n", + "9,Lesly,35,Spain,firefighter\n", + "10,Orelia,31,Germany,police officer\n", + "11,Theodora,16,Italy,unemployed\n", + "12,Viviene,44,Germany,police officer\n", + "13,Teriann,50,Belgium,police officer\n", + "14,Carol-Jean,23,Germany,unemployed\n", + "15,Drucie,15,Spain,police officer\n", + "16,Elie,10,Italy,doctor\n", + "17,Shaylyn,34,Belgium,worker\n", + "18,Fayre,33,Spain,police officer\n", + "19,Sabina,52,Germany,police officer\n", + "20,Aryn,20,Belgium,police officer\n", + "21,Darlleen,49,Spain,worker\n", + "22,Jere,18,Italy,worker\n", + "23,Candi,60,Germany,police officer\n", + "24,Sindee,40,Germany,firefighter\n", + "25,Selma,20,Spain,worker\n", + "26,Vonny,35,Germany,doctor\n", + "27,Kate,53,Spain,worker\n", + "28,Annabela,48,Belgium,worker\n", + "29,Jenilee,55,Germany,police officer\n", + "30,Gusella,44,France,police officer\n", + "31,Fawne,35,Spain,worker\n", + "32,Karolina,39,Spain,police officer\n", + "33,Sadie,58,Germany,firefighter\n", + "34,Clo,10,Italy,police officer\n", + "35,Beth,46,Spain,firefighter\n", + "36,Adore,18,Italy,firefighter\n", + "37,Tarra,29,Spain,doctor\n", + "38,Jessamyn,36,France,police officer\n", + "39,Deedee,24,Germany,unemployed\n", + "40,Patricia,45,Italy,doctor\n", + "41,Wileen,39,Spain,police officer\n", + "42,Paola,55,Italy,worker\n", + "43,Gwyneth,37,Italy,worker\n", + "44,Stacey,36,Spain,worker\n", + "45,Camile,60,Germany,unemployed\n", + "46,Sheree,10,Spain,unemployed\n", + "47,Albertina,53,France,police officer\n", + "48,Jinny,30,Spain,firefighter\n", + "49,Kayla,57,Italy,firefighter\n", + "50,Jaime,55,France,doctor" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Let's validate if the file was created correctly. You should see the first few lines from the generated file. Validate if the beginning of the file matches with the declared content above." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "! head data/people.csv" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Your first YAML pipelines\n", + "\n", + "In this section we'll present you the basic structure of a YAML pipeline and present you some available transforms.\n", + "Below is a simple pipeline that reads data from the csv file we've just created and logs the elements for debugging purposes.\n", + "\n", + "The `LogForTesting` transform lets us log the data when developing a pipeline. Remember, it is not advised to use this transform in production.\n", + "\n", + "Let's define the pipeline and save it to a file:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%%writefile 'pipelines/pipeline-01.yaml'\n", + "pipeline:\n", + " type: chain\n", + " transforms:\n", + " - type: ReadFromCsv\n", + " config:\n", + " path: data/people.csv\n", + " - type: LogForTesting" + ] + }, + { + "cell_type": "markdown", + "source": [ + "We can verify the contents of this file by running:" + ], + "metadata": { + "collapsed": false + } + }, + { + "cell_type": "code", + "execution_count": null, + "outputs": [], + "source": [ + "! cat pipelines/pipeline-01.yaml" + ], + "metadata": { + "collapsed": false + } + }, + { + "cell_type": "markdown", + "source": [ + "Now, we can execute the yaml pipeline by passing this file as an argument to the following command:" + ], + "metadata": { + "collapsed": false + } + }, + { + "cell_type": "code", + "execution_count": null, + "outputs": [], + "source": [ + "! python -m apache_beam.yaml.main --pipeline_spec_file=pipelines/pipeline-01.yaml" + ], + "metadata": { + "collapsed": false + } + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Here we use Python and `apache_beam` package to execute the pipeline, but we envision various services (such as Dataflow) to accept yaml pipelines directly obviating the need for that in the future.\n", + "\n", + "If you scroll through the output logs, you'll find entries such as:\n", + "```\n", + "INFO:root:BeamSchema_edf39b51_91da_418a_b28e_af04c9bae811(id=1, firstname='Reeba', age=58, country='Belgium', profession='unemployed')\n", + "INFO:root:BeamSchema_edf39b51_91da_418a_b28e_af04c9bae811(id=2, firstname='Maud', age=45, country='Spain', profession='firefighter')\n", + "INFO:root:BeamSchema_edf39b51_91da_418a_b28e_af04c9bae811(id=3, firstname='Meg', age=11, country='France', profession='unemployed')\n", + "INFO:root:BeamSchema_edf39b51_91da_418a_b28e_af04c9bae811(id=4, firstname='Rani', age=53, country='Spain', profession='doctor')\n", + "```\n", + "This is a representation of records from the input dataset." + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Let's add a transform - `Filter`. To use this transform you need to specify the 'keep' condition and a language your condition is written in. Below you'll find an example with a condition written in Python.\n", + "This pipeline will filter out records containing people that are younger than 18 years old. The only records left to the next transform will be records representing adults. Verify this by scrolling to the bottom of the output logs." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%%writefile 'pipelines/pipeline-filter-01.yaml'\n", + "pipeline:\n", + " type: chain\n", + " transforms:\n", + " - type: ReadFromCsv\n", + " config:\n", + " path: data/people.csv\n", + " - type: Filter\n", + " config:\n", + " language: python\n", + " keep: \"age >= 18\"\n", + " - type: LogForTesting" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "outputs": [], + "source": [ + "! python -m apache_beam.yaml.main --pipeline_spec_file=pipelines/pipeline-filter-01.yaml" + ], + "metadata": { + "collapsed": false + } + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Similarly, we can create a condition in other languages, for example SQL. In this example we filter out people that are younger than 18 and have a profession other than being 'unemployed'." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%%writefile 'pipelines/pipeline-filter-02.yaml'\n", + "pipeline:\n", + " type: chain\n", + " transforms:\n", + " - type: ReadFromCsv\n", + " config:\n", + " path: data/people.csv\n", + " - type: Filter\n", + " config:\n", + " language: sql\n", + " keep: \"age >= 18 or (age < 18 and profession = 'unemployed')\"\n", + " - type: LogForTesting" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "outputs": [], + "source": [ + "! python -m apache_beam.yaml.main --pipeline_spec_file=pipelines/pipeline-filter-02.yaml" + ], + "metadata": { + "collapsed": false + } + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "You'll notice that the output of this pipeline is in a different format than the previous one. That's because this pipeline uses an SQL Filter transform, an example of a [multi-language transform](https://beam.apache.org/documentation/programming-guide/#multi-language-pipelines). Multi-language pipelines are an important feature of Beam, but in this notebook we'll focus on YAML.\n", + "\n", + "To find the output of this pipeline find lines that begin with 'message' keyword and have the associated 'transform_id' set to a transform starting with 'LogForTesting'.\n", + "Example:\n", + "```\n", + "message: \"{\\\"id\\\":49,\\\"firstname\\\":\\\"Kayla\\\",\\\"age\\\":57,\\\"country\\\":\\\"Italy\\\",\\\"profession\\\":\\\"firefighter\\\"}\"\n", + "instruction_id: \"bundle_6\"\n", + "transform_id: \"LogForTesting/beam:schematransform:org.apache.beam:yaml:log_for_testing:v1/LogAsJson/ParMultiDo(Anonymous)\"\n", + "```\n", + "Each log entry represents one element from the output data." + ] + }, + { + "cell_type": "markdown", + "source": [ + "Another useful transform is `MapToFields`. This transform lets us manipulate fields of a record. For example, we can add a field to our records, which is a boolean field specifying if the person is adult or not." + ], + "metadata": { + "collapsed": false + } + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": { + "scrolled": true + }, + "outputs": [], + "source": [ + "pipeline = '''\n", + "pipeline:\n", + " type: chain\n", + " transforms:\n", + " - type: ReadFromCsv\n", + " config:\n", + " path: data/people.csv\n", + " - type: MapToFields\n", + " config:\n", + " language: python\n", + " append: true\n", + " fields:\n", + " is_adult: \"age >= 18\"\n", + " - type: LogForTesting\n", + "'''\n", + "save_to_file(pipeline, 'pipelines/pipeline-map-01.yaml')\n", + "! python -m apache_beam.yaml.main --pipeline_spec_file=pipelines/pipeline-map-01.yaml" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Beam will try to infer the types involved in the mappings, but sometimes this is not possible. In these cases we can explicitly denote the expected output type, e.g." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": { + "scrolled": true + }, + "outputs": [], + "source": [ + "%%writefile 'pipelines/pipeline-map-02.yaml'\n", + "pipeline:\n", + " type: chain\n", + " transforms:\n", + " - type: ReadFromCsv\n", + " config:\n", + " path: data/people.csv\n", + " - type: MapToFields\n", + " config:\n", + " language: python\n", + " append: true\n", + " fields:\n", + " is_adult:\n", + " expression: \"age >= 18\"\n", + " output_type: boolean\n", + " - type: LogForTesting" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "outputs": [], + "source": [ + "! python -m apache_beam.yaml.main --pipeline_spec_file=pipelines/pipeline-map-02.yaml" + ], + "metadata": { + "collapsed": false + } + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "When the `append` field is specified, one can `drop` fields as well, e.g." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": { + "scrolled": true + }, + "outputs": [], + "source": [ + "%%writefile 'pipelines/pipeline-map-03.yaml'\n", + "pipeline:\n", + " type: chain\n", + " transforms:\n", + " - type: ReadFromCsv\n", + " config:\n", + " path: data/people.csv\n", + " - type: MapToFields\n", + " config:\n", + " language: python\n", + " append: true\n", + " fields:\n", + " is_adult: \"age >= 18\"\n", + " drop:\n", + " - age\n", + " - type: LogForTesting" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "outputs": [], + "source": [ + "! python -m apache_beam.yaml.main --pipeline_spec_file=pipelines/pipeline-map-03.yaml" + ], + "metadata": { + "collapsed": false + } + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "We can also create simple UDFs (User Defined Functions) using Python or other languages. In the example below we add a field `random_number` which value is a random number not bigger than the age of the person." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": { + "scrolled": true + }, + "outputs": [], + "source": [ + "%%writefile 'pipelines/pipeline-map-04.yaml'\n", + "pipeline:\n", + " type: chain\n", + " transforms:\n", + " - type: ReadFromCsv\n", + " config:\n", + " path: data/people.csv\n", + " - type: MapToFields\n", + " config:\n", + " language: python\n", + " append: true\n", + " fields:\n", + " random_number:\n", + " callable: |\n", + " import random\n", + " def my_mapping(row):\n", + " return random.randrange(row.age)\n", + " - type: LogForTesting" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "outputs": [], + "source": [ + "! python -m apache_beam.yaml.main --pipeline_spec_file=pipelines/pipeline-map-04.yaml" + ], + "metadata": { + "collapsed": false + } + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Beam YAML has EXPERIMENTAL ability to do aggregations to group and combine values across records. The is accomplished via the `Combine` transform type. Currently `Combine` needs to be in the `yaml_experimental_features` option (see the bottom of the pipeline) to use this transform.\n", + "\n", + "In this example we'll aggregate our records based on the `is_adult` classification. We'll calculate an average age for each of the groups." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": { + "scrolled": true + }, + "outputs": [], + "source": [ + "%%writefile 'pipelines/pipeline-combine-01.yaml'\n", + "pipeline:\n", + " type: chain\n", + " transforms:\n", + " - type: ReadFromCsv\n", + " config:\n", + " path: data/people.csv\n", + " - type: MapToFields\n", + " config:\n", + " language: python\n", + " append: true\n", + " fields:\n", + " is_adult: \"age >= 18\"\n", + " - type: Combine\n", + " config:\n", + " group_by: is_adult\n", + " combine:\n", + " total:\n", + " value: age\n", + " fn: mean\n", + " - type: LogForTesting\n", + "options:\n", + " yaml_experimental_features: Combine" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "outputs": [], + "source": [ + "! python -m apache_beam.yaml.main --pipeline_spec_file=pipelines/pipeline-combine-01.yaml" + ], + "metadata": { + "collapsed": false + } + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "If all was executed correctly, you should see the following lines at the bottom of the output log:\n", + "```\n", + "INFO:root:Result(is_adult=True, total=40.674418604651166)\n", + "INFO:root:Result(is_adult=False, total=11.714285714285714\n", + "```" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "All the previous pipelines were linear - output of one transform was an input to the next transform. This is also known as a `chain` pipeline. This is designated in the top-level pipeline configuration, for example:\n", + "```\n", + "pipeline:\n", + " type: chain\n", + " transforms:\n", + " ...\n", + "```\n", + "In YAML we can also create nonlinear pipelines. To do this, we can specify `type: composite`, or omit this line completely (this is the default pipeline type). In these pipelines, we must specify the `input` in each of the transforms that take the output of previous transforms. This `input` is the name, or collection of names, of the transform(s) that feed into the receiving transform.\n", + "The specification below will create the following pipeline:\n", + "```\n", + " +----> Doctors -----------> SaveDoctors\n", + "InputData ---+\n", + " +----> OtherProfessions --> SaveOtherProfessions\n", + "```" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": { + "scrolled": true + }, + "outputs": [], + "source": [ + "%%writefile 'pipelines/pipeline-nonlinear-01.yaml'\n", + "pipeline:\n", + " type: composite\n", + " transforms:\n", + " - type: ReadFromCsv\n", + " name: InputData\n", + " config:\n", + " path: data/people.csv\n", + " - type: Filter\n", + " name: Doctors\n", + " input: InputData\n", + " config:\n", + " language: python\n", + " keep: \"profession == 'doctor'\"\n", + " - type: Filter\n", + " name: OtherProfessions\n", + " input: InputData\n", + " config:\n", + " language: python\n", + " keep: \"profession != 'doctor'\"\n", + " - type: WriteToCsv\n", + " name: SaveDoctors\n", + " input: Doctors\n", + " config:\n", + " path: results/doctors\n", + " - type: WriteToCsv\n", + " name: SaveOtherProfessions\n", + " input: OtherProfessions\n", + " config:\n", + " path: results/other-professions" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "outputs": [], + "source": [ + "! python -m apache_beam.yaml.main --pipeline_spec_file=pipelines/pipeline-nonlinear-01.yaml" + ], + "metadata": { + "collapsed": false + } + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "The output are 2 files: `results/doctors-00000-of-00001` and `results/other-professions-00000-of-00001`. Let's see their contents:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "! head results/doctors-00000-of-00001" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "! head results/other-professions-00000-of-00001" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Summary\n", + "Congratulations! You've just run Apache Beam pipelines using YAML.\n", + "\n", + "To learn more about Beam YAML visit [Beam YAML API documentation page](https://beam.apache.org/documentation/sdks/yaml/).\n", + "\n", + "To run your pipeline in Dataflow, you'll need to set up your Google Cloud and run the pipeline with the `DataflowRunner`. For more information, follow https://cloud.google.com/dataflow/docs/guides/deploying-a-pipeline#run-on-dataflow" + ] + } + ], + "metadata": { + "colab": { + "collapsed_sections": [], + "include_colab_link": true, + "name": "Try Apache Beam - Python", + "provenance": [], + "toc_visible": true, + "version": "0.3.2" + }, + "kernelspec": { + "display_name": "Python 3 (ipykernel)", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.8.9" + } + }, + "nbformat": 4, + "nbformat_minor": 1 +}