{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "ETL Pipelines with Prefect\n", "==========================\n", "\n", "[Prefect](https://github.com/PrefectHQ/prefect) is a platform for automating data workflows. Data engineers and data scientists can build, test and deploy production pipelines without worrying about all of the [\"negative engineering\" aspects](https://medium.com/the-prefect-blog/positive-and-negative-data-engineering-a02cb497583d) of production. For example, Prefect makes it easy to deploy a workflow that runs on a complicated schedule, requires task retries in the event of failures, and sends notifications when certain tasks are complete. Prefect was built on top of Dask, and [relies on Dask](http://stories.dask.org/en/latest/prefect-workflows.html#how-dask-helps) to schedule and manage the execution of a Prefect workflow in a distributed environment.\n", "\n", "This example demonstrates running a Prefect ETL Flow on Dask which ultimately creates a GIF. While this is a somewhat unconventional use case of Prefect, we're no strangers to [unconventional use cases](https://medium.com/the-prefect-blog/prefect-runs-on-prefect-3e6df553c3a4).\n", "\n", "In the world of workflow engines, Prefect supports many unique features; in this particular example we will see:\n", "\n", "- parametrization of workflows\n", "- dynamic runtime \"mapping\" of workflow tasks\n", "- customizable execution logic\n", "\n", "You wouldn't get this from any other engine.\n", "\n", "**Contents**\n", "\n", "0. [Description of goal](#Goal)\n", "1. [Setting up our environment](#Setting-up-our-environment)\n", "2. Building our Flow\n", " 1. [Extract](#Extract)\n", " 1. [Transform](#Transform)\n", " 1. [Load](#Load)\n", " 1. [Putting the pieces together](#Build-the-Flow)\n", "3. [Running our Flow on Dask](#Running-our-Flow-on-Dask)\n", "4. [Watching our GIF](#Play)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "[Goal](#Goal)\n", "----\n", "\n", "To demonstrate how Prefect and Dask work together, we are going to build and execute a standard \"Extract / Transform / Load\" (ETL) workflow for processing some basic image data. Most ETL workflows involve a scheduled migration of data from one database to another. In our case, we will be moving data from a file located at a known URL to our local hard disk, converting the individual file into a series of frames, and compiling those frames into a GIF. The URL references a file containing raw bytes such as:\n", "\n", "```python\n", "b\"\"\"aÙ\u0001w\u0001˜\u0000\u0000≠•∆≠≠fi#!\b\u0015\u0016\u0003≠≠÷≠•Ω≠úΩ••µú•µîúµ•úΩ••Ω3&\u0015µ•Ω\u0018!\u000b", "µ≠∆≠•¥4(%µú∑≠≠Œ≠î≠≠≠∆≠îµúî≠úîµE5.≠ú≠≠•Œµµfi••∆•≠ŒµµŒúúΩ62&)1&623µ•∆Ωµ÷úî•ßjxΩΩÁú•Ωµ≠Œ••≠ú•≠Ω≠∆≠µÁâUV≠µ‹ΩµŒîî•NC5µ≠Ÿôãô•î•µ•µîú≠#\u0017\bVHCuhl≠≠ΩôchâRIoc]™≠Á≠î•™ú»öis•ú•f7,íYfL9?îî≠≠•÷∑ò™gWVxGEΩ≠–))1qB5µ≠Ω81\u0018R,\u0015´tÜñWV\u001e", "!\u001e", "HCDBB5;5?\"\"\"\n", "```\n", "\n", "The steps of our workflow will be as follows:\n", "- Extract: pull the data file from a URL (speicified by a `Parameter`) to disk\n", "- Transform: split the file into multiple files, each corresponding to a single frame\n", "- Load: Store each frame individually, and compile the frames together into a GIF\n", "\n", "Once we have built our Flow, we can execute it with different values for the `Parameter` or even run it on a nightly schedule.\n", "\n", "**NOTE:** If we planned on executing this Flow in a truly distributed environment, writing the images to the local filesystem would _not_ be appropriate. We would instead use an external datastore such as Google Cloud Storage, or a proper database." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "[Setting up our environment](#Setting-up-our-environment)\n", "--------------------------\n", "\n", "Before proceeding, we need to install both the [`prefect`](https://pypi.org/project/prefect/) and [`imageio`](https://pypi.org/project/imageio/) packages." ] }, { "cell_type": "code", "execution_count": 1, "metadata": { "execution": { "iopub.execute_input": "2021-01-14T10:43:33.742171Z", "iopub.status.busy": "2021-01-14T10:43:33.741593Z", "iopub.status.idle": "2021-01-14T10:43:38.789572Z", "shell.execute_reply": "2021-01-14T10:43:38.788635Z" } }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Collecting imageio\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ " Downloading imageio-2.9.0-py3-none-any.whl (3.3 MB)\r\n", "\u001b[?25l\r", "\u001b[K | | 10 kB 36.5 MB/s eta 0:00:01\r", "\u001b[K |▏ | 20 kB 37.0 MB/s eta 0:00:01\r", "\u001b[K |▎ | 30 kB 41.7 MB/s eta 0:00:01\r", "\u001b[K |▍ | 40 kB 9.5 MB/s eta 0:00:01\r", "\u001b[K |▌ | 51 kB 11.3 MB/s eta 0:00:01\r", "\u001b[K |▋ | 61 kB 13.0 MB/s eta 0:00:01\r", "\u001b[K |▊ | 71 kB 11.1 MB/s eta 0:00:01\r", "\u001b[K |▉ | 81 kB 12.3 MB/s eta 0:00:01\r", "\u001b[K |█ | 92 kB 13.5 MB/s eta 0:00:01\r", "\u001b[K |█ | 102 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█ | 112 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█▏ | 122 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█▎ | 133 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█▍ | 143 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█▌ | 153 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█▋ | 163 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█▊ | 174 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█▉ | 184 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█▉ | 194 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██ | 204 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██ | 215 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██▏ | 225 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██▎ | 235 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██▍ | 245 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██▌ | 256 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██▋ | 266 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██▊ | 276 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██▊ | 286 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██▉ | 296 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███ | 307 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███ | 317 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███▏ | 327 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███▎ | 337 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███▍ | 348 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███▌ | 358 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███▋ | 368 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███▋ | 378 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███▊ | 389 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███▉ | 399 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████ | 409 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████ | 419 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████▏ | 430 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████▎ | 440 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████▍ | 450 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████▌ | 460 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████▋ | 471 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████▋ | 481 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████▊ | 491 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████▉ | 501 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████ | 512 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████ | 522 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████▏ | 532 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████▎ | 542 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████▍ | 552 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████▌ | 563 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████▌ | 573 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████▋ | 583 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████▊ | 593 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████▉ | 604 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████ | 614 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████ | 624 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████▏ | 634 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████▎ | 645 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████▍ | 655 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████▍ | 665 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████▌ | 675 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████▋ | 686 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████▊ | 696 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████▉ | 706 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████ | 716 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████ | 727 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████▏ | 737 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████▎ | 747 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████▎ | 757 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████▍ | 768 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████▌ | 778 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████▋ | 788 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████▊ | 798 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████▉ | 808 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████ | 819 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████ | 829 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████▏ | 839 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████▏ | 849 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████▎ | 860 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████▍ | 870 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████▌ | 880 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████▋ | 890 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████▊ | 901 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████▉ | 911 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████ | 921 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████ | 931 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████▏ | 942 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████▏ | 952 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████▎ | 962 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████▍ | 972 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████▌ | 983 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████▋ | 993 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████▊ | 1.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████▉ | 1.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████ | 1.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████ | 1.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████ | 1.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████▏ | 1.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████▎ | 1.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████▍ | 1.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████▌ | 1.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████▋ | 1.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████▊ | 1.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████▉ | 1.1 MB 10.9 MB/s eta 0:00:01" ] }, { "name": "stdout", "output_type": "stream", "text": [ "\r", "\u001b[K |███████████ | 1.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████ | 1.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████ | 1.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████▏ | 1.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████▎ | 1.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████▍ | 1.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████▌ | 1.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████▋ | 1.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████▊ | 1.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████▉ | 1.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████▉ | 1.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████ | 1.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████ | 1.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████▏ | 1.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████▎ | 1.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████▍ | 1.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████▌ | 1.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████▋ | 1.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████▊ | 1.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████▉ | 1.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████▉ | 1.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████ | 1.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████ | 1.4 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████▏ | 1.4 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████▎ | 1.4 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████▍ | 1.4 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████▌ | 1.4 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████▋ | 1.4 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████▊ | 1.4 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████▊ | 1.4 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████▉ | 1.4 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████ | 1.4 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████ | 1.5 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████▏ | 1.5 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████▎ | 1.5 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████▍ | 1.5 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████▌ | 1.5 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████▋ | 1.5 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████▋ | 1.5 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████▊ | 1.5 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████▉ | 1.5 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████ | 1.5 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████ | 1.6 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████▏ | 1.6 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████▎ | 1.6 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████▍ | 1.6 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████▌ | 1.6 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████▌ | 1.6 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████▋ | 1.6 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████▊ | 1.6 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████▉ | 1.6 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████ | 1.6 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████ | 1.7 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████▏ | 1.7 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████▎ | 1.7 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████▍ | 1.7 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████▍ | 1.7 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████▌ | 1.7 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████▋ | 1.7 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████▊ | 1.7 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████▉ | 1.7 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████ | 1.8 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████ | 1.8 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████▏ | 1.8 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████▎ | 1.8 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████▍ | 1.8 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████▍ | 1.8 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████▌ | 1.8 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████▋ | 1.8 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████▊ | 1.8 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████▉ | 1.8 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████ | 1.9 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████ | 1.9 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████▏ | 1.9 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████▎ | 1.9 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████▎ | 1.9 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████▍ | 1.9 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████▌ | 1.9 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████▋ | 1.9 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████▊ | 1.9 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████▉ | 1.9 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████ | 2.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████ | 2.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████▏ | 2.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████▏ | 2.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████▎ | 2.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████▍ | 2.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████▌ | 2.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████▋ | 2.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████▊ | 2.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████▉ | 2.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████ | 2.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████ | 2.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████ | 2.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████▏ | 2.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████▎ | 2.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████▍ | 2.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████▌ | 2.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████▋ | 2.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████▊ | 2.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████▉ | 2.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████ | 2.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████ | 2.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████ | 2.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████▏ | 2.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████▎ | 2.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████▍ | 2.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████▌ | 2.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████▋ | 2.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████▊ | 2.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████▉ | 2.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████ | 2.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████ | 2.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████ | 2.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████▏ | 2.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████▎ | 2.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████▍ | 2.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████▌ | 2.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████▋ | 2.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████▊ | 2.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████▉ | 2.4 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████▉ | 2.4 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████ | 2.4 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████ | 2.4 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████▏ | 2.4 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████▎ | 2.4 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████▍ | 2.4 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████▌ | 2.4 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████▋ | 2.4 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████▊ | 2.4 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████▊ | 2.5 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████▉ | 2.5 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████ | 2.5 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████ | 2.5 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████▏ | 2.5 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████▎ | 2.5 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████▍ | 2.5 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████▌ | 2.5 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████▋ | 2.5 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████▋ | 2.5 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████▊ | 2.6 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████▉ | 2.6 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████ | 2.6 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████ | 2.6 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████▏ | 2.6 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████▎ | 2.6 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████▍ | 2.6 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████▌ | 2.6 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████▋ | 2.6 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████▋ | 2.7 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████▊ | 2.7 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████▉ | 2.7 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████ | 2.7 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████ | 2.7 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████▏ | 2.7 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████▎ | 2.7 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████▍ | 2.7 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████▌ | 2.7 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████▌ | 2.7 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████▋ | 2.8 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████▊ | 2.8 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████▉ | 2.8 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████ | 2.8 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████ | 2.8 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████▏ | 2.8 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████▎ | 2.8 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████▍ | 2.8 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████▍ | 2.8 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████▌ | 2.8 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████▋ | 2.9 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████▊ | 2.9 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████▉ | 2.9 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████ | 2.9 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████ | 2.9 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████▏ | 2.9 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████▎ | 2.9 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████▎ | 2.9 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████▍ | 2.9 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████▌ | 2.9 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████▋ | 3.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████▊ | 3.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████▉ | 3.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████████ | 3.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████████ | 3.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████████▏ | 3.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████████▏ | 3.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████████▎ | 3.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████████▍ | 3.0 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████████▌ | 3.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████████▋ | 3.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████████▊ | 3.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████████▉ | 3.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████████ | 3.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████████ | 3.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████████▏ | 3.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████████▏ | 3.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████████▎ | 3.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████████▍ | 3.1 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████████▌ | 3.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████████▋ | 3.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████████▊ | 3.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████████▉ | 3.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████████ | 3.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████████ | 3.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████████ | 3.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████████▏| 3.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████████▎| 3.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████████▍| 3.2 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████████▌| 3.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████████▋| 3.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████████▊| 3.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████████▉| 3.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████████| 3.3 MB 10.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████████| 3.3 MB 10.9 MB/s eta 0:00:01" ] }, { "name": "stdout", "output_type": "stream", "text": [ "\r", "\u001b[K |████████████████████████████████| 3.3 MB 10.9 MB/s \r\n", "\u001b[?25hRequirement already satisfied: numpy in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from imageio) (1.18.5)\r\n", "Requirement already satisfied: pillow in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from imageio) (8.1.0)\r\n", "Collecting prefect[viz]\r\n", " Downloading prefect-0.14.3-py3-none-any.whl (476 kB)\r\n", "\u001b[?25l\r", "\u001b[K |▊ | 10 kB 64.7 MB/s eta 0:00:01\r", "\u001b[K |█▍ | 20 kB 75.0 MB/s eta 0:00:01\r", "\u001b[K |██ | 30 kB 89.6 MB/s eta 0:00:01\r", "\u001b[K |██▊ | 40 kB 93.9 MB/s eta 0:00:01\r", "\u001b[K |███▍ | 51 kB 96.5 MB/s eta 0:00:01\r", "\u001b[K |████▏ | 61 kB 103.5 MB/s eta 0:00:01\r", "\u001b[K |████▉ | 71 kB 108.0 MB/s eta 0:00:01\r", "\u001b[K |█████▌ | 81 kB 113.7 MB/s eta 0:00:01\r", "\u001b[K |██████▏ | 92 kB 115.7 MB/s eta 0:00:01\r", "\u001b[K |██████▉ | 102 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |███████▋ | 112 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |████████▎ | 122 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |█████████ | 133 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |█████████▋ | 143 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |██████████▎ | 153 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |███████████ | 163 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |███████████▊ | 174 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K" ] }, { "name": "stdout", "output_type": "stream", "text": [ " |████████████▍ | 184 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |█████████████ | 194 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |█████████████▊ | 204 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |██████████████▌ | 215 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |███████████████▏ | 225 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |███████████████▉ | 235 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |████████████████▌ | 245 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |█████████████████▏ | 256 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |██████████████████ | 266 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |██████████████████▋ | 276 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |███████████████████▎ | 286 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |████████████████████ | 296 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |████████████████████▋ | 307 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████▎ | 317 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████ | 327 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████▊ | 337 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████▍ | 348 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████ | 358 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████▊ | 368 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████▌ | 378 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████▏ | 389 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████▉ | 399 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████▌ | 409 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████▏ | 419 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████████ | 430 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████████▋ | 440 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████████▎ | 450 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████████ | 460 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████████▋| 471 kB 118.0 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████████| 476 kB 118.0 MB/s \r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "\u001b[?25hRequirement already satisfied: urllib3>=1.24.3 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from prefect[viz]) (1.26.2)\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Requirement already satisfied: msgpack>=0.6.0 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from prefect[viz]) (1.0.2)\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Requirement already satisfied: dask>=2.17.0 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from prefect[viz]) (2.20.0)\r\n", "Requirement already satisfied: click>=7.0 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from prefect[viz]) (7.1.2)\r\n", "Requirement already satisfied: cloudpickle>=1.3.0 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from prefect[viz]) (1.6.0)\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Requirement already satisfied: toml>=0.9.4 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from prefect[viz]) (0.10.2)\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Requirement already satisfied: distributed>=2.17.0 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from prefect[viz]) (2.30.1)\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Requirement already satisfied: requests>=2.20 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from prefect[viz]) (2.25.1)\r\n", "Requirement already satisfied: pytz>=2018.7 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from prefect[viz]) (2020.5)\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Requirement already satisfied: pyyaml>=3.13 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from prefect[viz]) (5.3.1)\r\n", "Requirement already satisfied: python-dateutil>=2.7.0 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from prefect[viz]) (2.8.1)\r\n", "Requirement already satisfied: graphviz>=0.8.3 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from prefect[viz]) (0.16)\r\n", "Collecting croniter<1.0,>=0.3.24\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ " Downloading croniter-0.3.37-py2.py3-none-any.whl (13 kB)\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Requirement already satisfied: zict>=0.1.3 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from distributed>=2.17.0->prefect[viz]) (2.0.0)\r\n", "Requirement already satisfied: toolz>=0.8.2 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from distributed>=2.17.0->prefect[viz]) (0.11.1)\r\n", "Requirement already satisfied: sortedcontainers!=2.0.0,!=2.0.1 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from distributed>=2.17.0->prefect[viz]) (2.3.0)\r\n", "Requirement already satisfied: tornado>=6.0.3 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from distributed>=2.17.0->prefect[viz]) (6.1)\r\n", "Requirement already satisfied: psutil>=5.0 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from distributed>=2.17.0->prefect[viz]) (5.8.0)\r\n", "Requirement already satisfied: setuptools in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from distributed>=2.17.0->prefect[viz]) (49.6.0.post20210108)\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Requirement already satisfied: tblib>=1.6.0 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from distributed>=2.17.0->prefect[viz]) (1.6.0)\r\n", "Collecting docker>=3.4.1\r\n", " Downloading docker-4.4.1-py2.py3-none-any.whl (146 kB)\r\n", "\u001b[?25l\r", "\u001b[K |██▎ | 10 kB 46.9 MB/s eta 0:00:01\r", "\u001b[K |████▌ | 20 kB 50.5 MB/s eta 0:00:01\r", "\u001b[K |██████▊ | 30 kB 61.6 MB/s eta 0:00:01\r", "\u001b[K |█████████ | 40 kB 60.4 MB/s eta 0:00:01\r", "\u001b[K |███████████▏ | 51 kB 61.1 MB/s eta 0:00:01\r", "\u001b[K |█████████████▍ | 61 kB 66.8 MB/s eta 0:00:01\r", "\u001b[K |███████████████▋ | 71 kB 68.8 MB/s eta 0:00:01\r", "\u001b[K |█████████████████▉ | 81 kB 73.9 MB/s eta 0:00:01\r", "\u001b[K |████████████████████ | 92 kB 73.7 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████▎ | 102 kB 76.3 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████▌ | 112 kB 76.3 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████▉ | 122 kB 76.3 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████████ | 133 kB 76.3 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████████▎| 143 kB 76.3 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████████| 146 kB 76.3 MB/s \r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "\u001b[?25hRequirement already satisfied: six>=1.4.0 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from docker>=3.4.1->prefect[viz]) (1.15.0)\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Collecting importlib-resources>=3.0.0\r\n", " Downloading importlib_resources-5.0.0-py3-none-any.whl (22 kB)\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Collecting marshmallow>=3.0.0b19\r\n", " Downloading marshmallow-3.10.0-py2.py3-none-any.whl (46 kB)\r\n", "\u001b[?25l\r", "\u001b[K |███████ | 10 kB 47.5 MB/s eta 0:00:01\r", "\u001b[K |██████████████▏ | 20 kB 52.2 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████▏ | 30 kB 63.6 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████▎ | 40 kB 66.6 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████████| 46 kB 14.2 MB/s \r\n", "\u001b[?25hCollecting marshmallow-oneofschema>=2.0.0b2\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ " Downloading marshmallow_oneofschema-2.1.0-py2.py3-none-any.whl (5.7 kB)\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Collecting mypy-extensions>=0.4.0\r\n", " Downloading mypy_extensions-0.4.3-py2.py3-none-any.whl (4.5 kB)\r\n", "Collecting pendulum>=2.0.4\r\n", " Downloading pendulum-2.1.2-cp38-cp38-manylinux1_x86_64.whl (155 kB)\r\n", "\u001b[?25l\r", "\u001b[K |██ | 10 kB 45.8 MB/s eta 0:00:01\r", "\u001b[K |████▏ | 20 kB 49.8 MB/s eta 0:00:01\r", "\u001b[K |██████▎ | 30 kB 56.0 MB/s eta 0:00:01\r", "\u001b[K |████████▍ | 40 kB 62.8 MB/s eta 0:00:01\r", "\u001b[K |██████████▌ | 51 kB 64.6 MB/s eta 0:00:01\r", "\u001b[K |████████████▋ | 61 kB 66.8 MB/s eta 0:00:01\r", "\u001b[K |██████████████▊ | 71 kB 66.5 MB/s eta 0:00:01\r", "\u001b[K |████████████████▉ | 81 kB 61.4 MB/s eta 0:00:01\r", "\u001b[K |███████████████████ | 92 kB 61.6 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████ | 102 kB 64.3 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████▏ | 112 kB 64.3 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████▎ | 122 kB 64.3 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████▍ | 133 kB 64.3 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████████▌ | 143 kB 64.3 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████████▋| 153 kB 64.3 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████████| 155 kB 64.3 MB/s \r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "\u001b[?25hCollecting python-box>=5.1.0\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ " Downloading python_box-5.2.0-py3-none-any.whl (20 kB)\r\n", "Collecting python-slugify>=1.2.6\r\n", " Downloading python-slugify-4.0.1.tar.gz (11 kB)\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Collecting pytzdata>=2020.1\r\n", " Downloading pytzdata-2020.1-py2.py3-none-any.whl (489 kB)\r\n", "\u001b[?25l" ] }, { "name": "stdout", "output_type": "stream", "text": [ "\r", "\u001b[K |▊ | 10 kB 25.3 MB/s eta 0:00:01\r", "\u001b[K |█▍ | 20 kB 7.7 MB/s eta 0:00:01\r", "\u001b[K |██ | 30 kB 10.9 MB/s eta 0:00:01\r", "\u001b[K |██▊ | 40 kB 13.8 MB/s eta 0:00:01\r", "\u001b[K |███▍ | 51 kB 16.3 MB/s eta 0:00:01\r", "\u001b[K |████ | 61 kB 18.7 MB/s eta 0:00:01\r", "\u001b[K |████▊ | 71 kB 20.6 MB/s eta 0:00:01\r", "\u001b[K |█████▍ | 81 kB 22.7 MB/s eta 0:00:01\r", "\u001b[K |██████ | 92 kB 25.0 MB/s eta 0:00:01\r", "\u001b[K |██████▊ | 102 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |███████▍ | 112 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |████████ | 122 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |████████▊ | 133 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |█████████▍ | 143 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |██████████ | 153 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |██████████▊ | 163 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |███████████▍ | 174 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |████████████ | 184 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |████████████▊ | 194 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |█████████████▍ | 204 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |██████████████ | 215 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |██████████████▊ | 225 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |███████████████▍ | 235 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |████████████████ | 245 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |████████████████▊ | 256 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |█████████████████▍ | 266 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |██████████████████ | 276 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |██████████████████▊ | 286 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |███████████████████▍ | 296 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |████████████████████ | 307 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |████████████████████▊ | 317 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████▍ | 327 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████ | 337 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████▊ | 348 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████▍ | 358 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████ | 368 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████▊ | 378 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████▍ | 389 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████ | 399 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████▊ | 409 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████▍ | 419 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████ | 430 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████▊ | 440 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████████▍ | 450 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████████ | 460 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████████▊ | 471 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████████▍| 481 kB 26.8 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████████| 489 kB 26.8 MB/s \r\n", "\u001b[?25hRequirement already satisfied: idna<3,>=2.5 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from requests>=2.20->prefect[viz]) (2.10)\r\n", "Requirement already satisfied: certifi>=2017.4.17 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from requests>=2.20->prefect[viz]) (2020.12.5)\r\n", "Requirement already satisfied: chardet<5,>=3.0.2 in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from requests>=2.20->prefect[viz]) (3.0.4)\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Collecting tabulate>=0.8.0\r\n", " Downloading tabulate-0.8.7-py3-none-any.whl (24 kB)\r\n", "Collecting text-unidecode>=1.3\r\n", " Downloading text_unidecode-1.3-py2.py3-none-any.whl (78 kB)\r\n", "\u001b[?25l\r", "\u001b[K |████▏ | 10 kB 61.6 MB/s eta 0:00:01\r", "\u001b[K |████████▍ | 20 kB 61.0 MB/s eta 0:00:01\r", "\u001b[K |████████████▋ | 30 kB 71.8 MB/s eta 0:00:01\r", "\u001b[K |████████████████▊ | 40 kB 75.4 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████ | 51 kB 78.9 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████▏ | 61 kB 77.1 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████████▍ | 71 kB 81.4 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████████| 78 kB 21.2 MB/s \r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "\u001b[?25hCollecting websocket-client>=0.32.0\r\n", " Downloading websocket_client-0.57.0-py2.py3-none-any.whl (200 kB)\r\n", "\u001b[?25l\r", "\u001b[K |█▋ | 10 kB 67.9 MB/s eta 0:00:01\r", "\u001b[K |███▎ | 20 kB 76.2 MB/s eta 0:00:01\r", "\u001b[K |█████ | 30 kB 86.1 MB/s eta 0:00:01\r", "\u001b[K |██████▌ | 40 kB 92.5 MB/s eta 0:00:01\r", "\u001b[K |████████▏ | 51 kB 96.3 MB/s eta 0:00:01\r", "\u001b[K |█████████▉ | 61 kB 103.2 MB/s eta 0:00:01\r", "\u001b[K |███████████▍ | 71 kB 107.0 MB/s eta 0:00:01\r", "\u001b[K |█████████████ | 81 kB 112.2 MB/s eta 0:00:01\r", "\u001b[K |██████████████▊ | 92 kB 114.6 MB/s eta 0:00:01\r", "\u001b[K |████████████████▎ | 102 kB 116.7 MB/s eta 0:00:01\r", "\u001b[K |██████████████████ | 112 kB 116.7 MB/s eta 0:00:01\r", "\u001b[K |███████████████████▋ | 122 kB 116.7 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████▏ | 133 kB 116.7 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████▉ | 143 kB 116.7 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████▌ | 153 kB 116.7 MB/s eta 0:00:01\r", "\u001b[K |██████████████████████████ | 163 kB 116.7 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████▊ | 174 kB 116.7 MB/s eta 0:00:01\r", "\u001b[K |█████████████████████████████▍ | 184 kB 116.7 MB/s eta 0:00:01\r", "\u001b[K |███████████████████████████████ | 194 kB 116.7 MB/s eta 0:00:01\r", "\u001b[K |████████████████████████████████| 200 kB 116.7 MB/s \r\n", "\u001b[?25hRequirement already satisfied: heapdict in /usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages (from zict>=0.1.3->distributed>=2.17.0->prefect[viz]) (1.0.1)\r\n", "Collecting natsort\r\n", " Downloading natsort-7.1.0-py3-none-any.whl (35 kB)\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Building wheels for collected packages: python-slugify\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ " Building wheel for python-slugify (setup.py) ... \u001b[?25l-" ] }, { "name": "stdout", "output_type": "stream", "text": [ "\b \bdone\r\n", "\u001b[?25h Created wheel for python-slugify: filename=python_slugify-4.0.1-py2.py3-none-any.whl size=6769 sha256=40020786f9a316b9e4fe0d1c2120af19bed5f6d850e918f3977fb59c8713eb0a\r\n", " Stored in directory: /home/runner/.cache/pip/wheels/91/4d/4f/e740a68c215791688c46c4d6251770a570e8dfea91af1acb5c\r\n", "Successfully built python-slugify\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Installing collected packages: websocket-client, text-unidecode, pytzdata, natsort, marshmallow, tabulate, python-slugify, python-box, pendulum, mypy-extensions, marshmallow-oneofschema, importlib-resources, docker, croniter, prefect, imageio\r\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Successfully installed croniter-0.3.37 docker-4.4.1 imageio-2.9.0 importlib-resources-5.0.0 marshmallow-3.10.0 marshmallow-oneofschema-2.1.0 mypy-extensions-0.4.3 natsort-7.1.0 pendulum-2.1.2 prefect-0.14.3 python-box-5.2.0 python-slugify-4.0.1 pytzdata-2020.1 tabulate-0.8.7 text-unidecode-1.3 websocket-client-0.57.0\r\n" ] } ], "source": [ "!pip install imageio prefect[viz]" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### [Extract](#Extract)\n", "\n", "First, we will define our tasks for _extracting_ the image data file from a given URL and saving it to a given file location. To do so, we will utilize two methods for creating Prefect Tasks:\n", "- the `task` decorator for converting any Python function into a task\n", "- a pre-written, configurable Task from the [Prefect \"Task Library\"](https://docs.prefect.io/guide/task_library/) which helps us abstract some standard boilerplate\n", "\n", "Additionally, we will utilize the following Prefect concepts:\n", "- a [Prefect signal](https://docs.prefect.io/guide/core_concepts/execution.html#state-signals) for marking this task and its downstream depedencies as successfully \"Skipped\" if the file is already present in our local filesystem\n", "- retry semantics: if, for whatever reason, our `curl` command fails to connect, we want it to retry up to 2 times with a 10 second delay. This way, if we run this workflow on a schedule we won't need to concern ourselves with temporary intermittent connection issues.\n", "\n", "Right now we are simply defining our individual tasks - we won't actually set up our dependency structure until we create the full Flow." ] }, { "cell_type": "code", "execution_count": 2, "metadata": { "execution": { "iopub.execute_input": "2021-01-14T10:43:38.792625Z", "iopub.status.busy": "2021-01-14T10:43:38.791758Z", "iopub.status.idle": "2021-01-14T10:43:39.240208Z", "shell.execute_reply": "2021-01-14T10:43:39.240902Z" } }, "outputs": [], "source": [ "import datetime\n", "import os\n", "\n", "import prefect\n", "from prefect import task\n", "from prefect.engine.signals import SKIP\n", "from prefect.tasks.shell import ShellTask\n", "\n", "\n", "@task\n", "def curl_cmd(url: str, fname: str) -> str:\n", " \"\"\"\n", " The curl command we wish to execute.\n", " \"\"\"\n", " if os.path.exists(fname):\n", " raise SKIP(\"Image data file already exists.\")\n", " return \"curl -fL -o {fname} {url}\".format(fname=fname, url=url)\n", "\n", "\n", "# ShellTask is a task from the Task library which will execute a given command in a subprocess\n", "# and fail if the command returns a non-zero exit code\n", "\n", "download = ShellTask(name=\"curl_task\", max_retries=2, retry_delay=datetime.timedelta(seconds=10))" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### [Transform](#Transform)\n", "\n", "Next up, we need to define our task which loads the image data file and splits it into multiple frames. In this case, each frame is delimited by 4 newlines. Note that, in the event the previous two tasks are \"Skipped\", the default behavior in Prefect is to skip downstream dependencies as well. However, as with most things in Prefect, this behavior is customizable. In this case, we want this task to run regardless of whether the upstreams skipped or not, so we set the `skip_on_upstream_skip` flag to `False`." ] }, { "cell_type": "code", "execution_count": 3, "metadata": { "execution": { "iopub.execute_input": "2021-01-14T10:43:39.243672Z", "iopub.status.busy": "2021-01-14T10:43:39.242800Z", "iopub.status.idle": "2021-01-14T10:43:39.247293Z", "shell.execute_reply": "2021-01-14T10:43:39.247887Z" } }, "outputs": [], "source": [ "@task(skip_on_upstream_skip=False)\n", "def load_and_split(fname: str) -> list:\n", " \"\"\"\n", " Loads image data file at `fname` and splits it into\n", " multiple frames. Returns a list of bytes, one element\n", " for each frame.\n", " \"\"\"\n", " with open(fname, \"rb\") as f:\n", " images = f.read()\n", " \n", " return [img for img in images.split(b\"\\n\" * 4) if img]" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### [Load](#Load)\n", "\n", "Finally, we want to write our frames to disk as well as combine the frames into a single GIF. In order to achieve this goal, we are going to utilize [Prefect's task \"mapping\" feature](https://docs.prefect.io/guide/core_concepts/mapping.html) which conveniently spawns new tasks in response to upstream outputs. In this case, we will write a single task for writing an image to disk, and \"map\" this task over all the image frames returned by `load_and_split` above! To infer which frame we are on, we look in [`prefect.context`](https://docs.prefect.io/guide/core_concepts/execution.html#context).\n", "\n", "Additionally, we can \"reduce\" over a mapped task - in this case, we will take the collection of mapped tasks and pass them into our `combine_to_gif` task for creating and saving our GIF." ] }, { "cell_type": "code", "execution_count": 4, "metadata": { "execution": { "iopub.execute_input": "2021-01-14T10:43:39.250312Z", "iopub.status.busy": "2021-01-14T10:43:39.249489Z", "iopub.status.idle": "2021-01-14T10:43:39.253493Z", "shell.execute_reply": "2021-01-14T10:43:39.254052Z" } }, "outputs": [], "source": [ "@task\n", "def write_to_disk(image: bytes) -> bytes:\n", " \"\"\"\n", " Given a single image represented as bytes, writes the image\n", " to the present working directory with a filename determined\n", " by `map_index`. Returns the image bytes.\n", " \"\"\"\n", " frame_no = prefect.context.get(\"map_index\")\n", " with open(\"frame_{0:0=2d}.gif\".format(frame_no), \"wb\") as f:\n", " f.write(image)\n", " return image" ] }, { "cell_type": "code", "execution_count": 5, "metadata": { "execution": { "iopub.execute_input": "2021-01-14T10:43:39.256524Z", "iopub.status.busy": "2021-01-14T10:43:39.255702Z", "iopub.status.idle": "2021-01-14T10:43:39.270060Z", "shell.execute_reply": "2021-01-14T10:43:39.270616Z" } }, "outputs": [], "source": [ "import imageio\n", "from io import BytesIO\n", "\n", "\n", "@task\n", "def combine_to_gif(image_bytes: list) -> None:\n", " \"\"\"\n", " Given a list of ordered images represented as bytes,\n", " combines them into a single GIF stored in the present working directory.\n", " \"\"\"\n", " images = [imageio.imread(BytesIO(image)) for image in image_bytes]\n", " imageio.mimsave('./clip.gif', images)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "[Build the Flow](#Build-the-Flow)\n", "---------------\n", "\n", "Finally, we need to put our tasks together into a Prefect \"Flow\". Similar to Dask's `delayed` interface, all computation is deferred and no Task code will be executed in this step. Because Prefect maintains a stricter contract between tasks and additionally needs the ability to run in non-Dask execution environments, the mechanism for deferring execution is independent of Dask.\n", "\n", "In addition to the tasks we have already defined, we introduce two \"Parameters\" for specifying the URL and local file location of our data. At runtime, we can optionally override these tasks to return different values." ] }, { "cell_type": "code", "execution_count": 6, "metadata": { "execution": { "iopub.execute_input": "2021-01-14T10:43:39.273045Z", "iopub.status.busy": "2021-01-14T10:43:39.272216Z", "iopub.status.idle": "2021-01-14T10:43:39.319207Z", "shell.execute_reply": "2021-01-14T10:43:39.318863Z" } }, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "Unable to revert mtime: /usr/share/fonts/cMap\n", "Unable to revert mtime: /usr/share/fonts/cmap/adobe-cns1\n", "Unable to revert mtime: /usr/share/fonts/cmap/adobe-gb1\n", "Unable to revert mtime: /usr/share/fonts/cmap/adobe-japan1\n", "Unable to revert mtime: /usr/share/fonts/cmap/adobe-japan2\n", "Unable to revert mtime: /usr/share/fonts/cmap/adobe-korea1\n" ] }, { "data": { "image/svg+xml": [ "\n", "\n", "\n", "\n", "\n", "\n", "%3\n", "\n", "\n", "\n", "139692964902560\n", "\n", "load_and_split\n", "\n", "\n", "\n", "139692964902896\n", "\n", "write_to_disk <map>\n", "\n", "\n", "\n", "139692964902560->139692964902896\n", "\n", "\n", "image\n", "\n", "\n", "\n", "139692964903424\n", "\n", "combine_to_gif\n", "\n", "\n", "\n", "139692964901840\n", "\n", "curl_cmd\n", "\n", "\n", "\n", "139692964901936\n", "\n", "curl_task\n", "\n", "\n", "\n", "139692964901840->139692964901936\n", "\n", "\n", "command\n", "\n", "\n", "\n", "139692964901936->139692964902560\n", "\n", "\n", "\n", "\n", "\n", "139692964900976\n", "\n", "DATA_URL\n", "\n", "\n", "\n", "139692964900976->139692964901840\n", "\n", "\n", "url\n", "\n", "\n", "\n", "139692964901072\n", "\n", "DATA_FILE\n", "\n", "\n", "\n", "139692964901072->139692964902560\n", "\n", "\n", "fname\n", "\n", "\n", "\n", "139692964901072->139692964901840\n", "\n", "\n", "fname\n", "\n", "\n", "\n", "139692964902896->139692964903424\n", "\n", "\n", "image_bytes\n", "\n", "\n", "\n" ], "text/plain": [ "" ] }, "execution_count": 6, "metadata": {}, "output_type": "execute_result" } ], "source": [ "from prefect import Parameter, Flow\n", "\n", "\n", "DATA_URL = Parameter(\"DATA_URL\", \n", " default=\"https://github.com/cicdw/image-data/blob/master/all-images.img?raw=true\")\n", "\n", "DATA_FILE = Parameter(\"DATA_FILE\", default=\"image-data.img\")\n", "\n", "\n", "with Flow(\"Image ETL\") as flow:\n", " \n", " # Extract\n", " command = curl_cmd(DATA_URL, DATA_FILE)\n", " curl = download(command=command)\n", " \n", " # Transform\n", " # we use the `upstream_tasks` keyword to specify non-data dependencies\n", " images = load_and_split(fname=DATA_FILE, upstream_tasks=[curl])\n", " \n", " # Load \n", " frames = write_to_disk.map(images)\n", " result = combine_to_gif(frames)\n", " \n", "\n", "flow.visualize()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### [Running the Flow on Dask](#Running-the-Flow-on-Dask)\n", "\n", "Now we have built our Flow, independently of Dask. We could execute this Flow sequentially, Task after Task, but there is inherent parallelism in our mapping of the images to files that we want to exploit. Luckily, Dask makes this easy to achieve.\n", "\n", "First, we will start a local Dask cluster. Then, we will run our Flow against Prefect's `DaskExecutor`, which will submit each Task to our Dask cluster and use Dask's distributed scheduler for determining when and where each Task should run. Essentially, we built a Directed Acylic Graph (DAG) and are simply \"submitting\" that DAG to Dask for handling its execution in a distributed way." ] }, { "cell_type": "code", "execution_count": 7, "metadata": { "execution": { "iopub.execute_input": "2021-01-14T10:43:39.326425Z", "iopub.status.busy": "2021-01-14T10:43:39.326020Z", "iopub.status.idle": "2021-01-14T10:43:43.992290Z", "shell.execute_reply": "2021-01-14T10:43:43.992637Z" }, "scrolled": true }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "[2021-01-14 10:43:40+0000] INFO - prefect.FlowRunner | Beginning Flow run for 'Image ETL'\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ ":11: UserWarning: prefect.engine.executors.DaskExecutor has been moved to `prefect.executors.DaskExecutor`, please update your imports\n", " executor = DaskExecutor(address=client.scheduler.address)\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "/usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages/distributed/worker.py:3373: UserWarning: Large object of size 2.03 MB detected in task graph: \n", " {'task': , 'state': None, 'up ... _parent': True}\n", "Consider scattering large objects ahead of time\n", "with client.scatter to reduce scheduler burden and \n", "keep data on workers\n", "\n", " future = client.submit(func, big_data) # bad\n", "\n", " big_future = client.scatter(big_data) # good\n", " future = client.submit(func, big_future) # good\n", " warnings.warn(\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "[2021-01-14 10:43:43+0000] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded\n" ] }, { "data": { "text/plain": [ "" ] }, "execution_count": 7, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# start our Dask cluster\n", "from dask.distributed import Client\n", "\n", "\n", "client = Client(n_workers=4, threads_per_worker=1)\n", "\n", "# point Prefect's DaskExecutor to our Dask cluster\n", "\n", "from prefect.engine.executors import DaskExecutor\n", "\n", "executor = DaskExecutor(address=client.scheduler.address)\n", "flow.run(executor=executor)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "[Next Steps](#Next-Steps)\n", "----------\n", "\n", "Now that we've built our workflow, what next? The interested reader should try to:\n", "\n", "- run the Flow again to see how the `SKIP` signal behaves\n", "- use different parameters for both the URL and the file location (Parameter values can be overriden by simply passing their names as keyword arguments to `flow.run()`)\n", "- introduce a new Parameter for the filename of the final GIF\n", "- use Prefect's [scheduler interface](https://docs.prefect.io/guide/core_concepts/schedules.html) to run our workflow on a schedule" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "[Play](#Play)\n", "----\n", "\n", "Finally, let's watch our creation!" ] }, { "cell_type": "code", "execution_count": 8, "metadata": { "execution": { "iopub.execute_input": "2021-01-14T10:43:43.996348Z", "iopub.status.busy": "2021-01-14T10:43:43.995704Z", "iopub.status.idle": "2021-01-14T10:43:43.999857Z", "shell.execute_reply": "2021-01-14T10:43:44.000522Z" } }, "outputs": [ { "data": { "text/html": [ "\"Rick" ], "text/plain": [ "" ] }, "execution_count": 8, "metadata": {}, "output_type": "execute_result" } ], "source": [ "from IPython.display import HTML\n", "\n", "HTML('\"Rick')" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "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.6" } }, "nbformat": 4, "nbformat_minor": 4 }