Repository: PrefectHQ/prefect

Issue:
Evaluate and cleanup examples
The `examples` folder could use some attention. Many of the examples are outdated or unclear in what they're demonstrating. We should go through and ensure that

- All examples run without error (preferably with no dependencies outside of prefect itself)
- All examples are up to date with current recommended practices
- Each example is clear in what it demonstrates, and is well commented.

Outdated or confusing examples should be removed.

We also might add running the examples as part of our CI system to ensure they remain functional.

Discussion:

Patch:
diff --git a/.gitignore b/.gitignore
index 95d78936e3..6abb149ce8 100644
--- a/.gitignore
+++ b/.gitignore
@@ -139,6 +139,9 @@ node_modules
 # docs
 docs/api/unreleased
 docs/api/latest
+docs/core/examples/
+!docs/core/examples/overview.md
+docs/.vuepress/public/examples.json
 docs/changelog.md
 docs/deploy/dist
 docs/.vuepress/public/prefect-coverage
diff --git a/changes/pr4295.yaml b/changes/pr4295.yaml
deleted file mode 100644
index 3f3a62faf1..0000000000
--- a/changes/pr4295.yaml
+++ /dev/null
@@ -1,3 +0,0 @@
-enhancement:
-  - "When manually set, `prefect.context.date` will be used to determine dependent values - [#4295](https://github.com/PrefectHQ/prefect/pull/4295)"
-  - "`prefect.context.date` will be cast to a `DateTime` object if given a parsable string - [#4295](https://github.com/PrefectHQ/prefect/pull/4295)"
\ No newline at end of file
diff --git a/changes/pr4309.yaml b/changes/pr4309.yaml
deleted file mode 100644
index a17116e036..0000000000
--- a/changes/pr4309.yaml
+++ /dev/null
@@ -1,5 +0,0 @@
-fix:
-  - "Fix `DatabricksRunNow` task attribute override behavior - [#4309](https://github.com/PrefectHQ/prefect/pull/4309)"
-
-contributor:
-  - "[Greg Lu](https://github.com/greglu)"
diff --git a/docs/.vuepress/config.js b/docs/.vuepress/config.js
index a815af8c14..b33f292cd7 100644
--- a/docs/.vuepress/config.js
+++ b/docs/.vuepress/config.js
@@ -338,7 +338,12 @@ module.exports = {
         {
           title: 'Examples',
           collapsable: true,
-          children: getChildren('docs/core', 'examples')
+          children: [
+            'examples/overview',
+            'examples/parameters',
+            'examples/mapping',
+            'examples/conditional'
+          ]
         },
         {
           title: 'PINs',
diff --git a/docs/.vuepress/styles/index.styl b/docs/.vuepress/styles/index.styl
index 2dbc224240..61abdcadbc 100644
--- a/docs/.vuepress/styles/index.styl
+++ b/docs/.vuepress/styles/index.styl
@@ -353,7 +353,6 @@ lang = '[class*="language-"]'
     box-decoration-break: clone;
     -webkit-box-decoration-break: clone;
     line-height: 2
-    white-space: pre-wrap
     tab-size: 4
 
 .prefect-block
diff --git a/docs/core/examples/README.md b/docs/core/examples/README.md
deleted file mode 100644
index 5d3d1778ca..0000000000
--- a/docs/core/examples/README.md
+++ /dev/null
@@ -1,7 +0,0 @@
----
-sidebarDepth: 0
----
-
-# Examples Overview
-
-This section contains a selection of easy to read, bite-sized examples of Prefect Flows intended to help you get a quick sense of the features and feel of Prefect.  This list is manually maintained, a more up-to-date list of examples can always be found in the [Prefect repository](https://github.com/PrefectHQ/prefect/tree/master/examples).
diff --git a/docs/core/examples/airflow_tutorial_dag.md b/docs/core/examples/airflow_tutorial_dag.md
deleted file mode 100644
index 434c93ff9e..0000000000
--- a/docs/core/examples/airflow_tutorial_dag.md
+++ /dev/null
@@ -1,63 +0,0 @@
----
-sidebarDepth: 0
----
-
-# Airflow Tutorial DAG
-
-A 1:1 rewrite of the [Airflow tutorial DAG](https://airflow.apache.org/tutorial.html)
-
-```python
-from datetime import datetime, timedelta
-
-import prefect
-from prefect import Parameter, task, Flow
-from prefect.schedules import IntervalSchedule
-from prefect.tasks.shell import ShellTask
-from prefect.tasks.templates.jinja2 import JinjaTemplate
-
-
-## default config settings such as this can generally be set in your
-## user config file
-retry_delay = timedelta(minutes=5)
-
-## create all relevant tasks
-t1 = ShellTask(
-    name="print_date", command="date", max_retries=1, retry_delay=retry_delay
-)
-t2 = ShellTask(name="sleep", command="sleep 5", max_retries=3, retry_delay=retry_delay)
-
-
-@task(max_retries=1, retry_delay=retry_delay)
-def add_7():
-    date = prefect.context.get("scheduled_start_time", datetime.utcnow())
-    return date + timedelta(days=7)
-
-
-## templated command; template vars will be read from both prefect.context as well as
-## any passed kwargs to the task
-command = """
-    {% for i in range(5) %}
-        echo "{{ scheduled_start_time }}"
-        echo "{{ scheduled_start_time_7 }}"
-        echo "{{ my_param }}"
-    {% endfor %}
-"""
-templated_command = JinjaTemplate(
-    template=command, max_retries=1, retry_delay=retry_delay
-)
-
-## create schedule for the Flow
-schedule = IntervalSchedule(start_date=datetime(2015, 6, 1), interval=timedelta(days=1))
-
-## create Flow and specify dependencies using functional API
-## we don't actually attach the schedule to this Flow so it only runs once
-with Flow("tutorial") as flow:
-    my_param = Parameter("my_param")
-    t2(upstream_tasks=[t1])
-    t3 = templated_command(
-        scheduled_start_time_7=add_7, my_param=my_param, upstream_tasks=[t1]
-    )
-
-
-flow.run(parameters={"my_param": "Parameter I passed in"})
-```
diff --git a/docs/core/examples/cached_task.md b/docs/core/examples/cached_task.md
deleted file mode 100644
index 08a870387a..0000000000
--- a/docs/core/examples/cached_task.md
+++ /dev/null
@@ -1,53 +0,0 @@
----
-sidebarDepth: 0
----
-
-# Task Caching
-
-This example demonstrates output caching; our first task returns a random number
-but requests to be cached for 1.5 minutes.  Our second task prints whatever input
-it is given.
-
-We run our flow on a 1 minute interval schedule and observe that the output of
-`return_random_number` only changes every other run, due to output caching.
-
-```python
-import datetime
-import random
-
-from prefect import task, Flow
-from prefect.schedules import IntervalSchedule
-
-
-@task(cache_for=datetime.timedelta(minutes=1, seconds=30))
-def return_random_number():
-    return random.random()
-
-
-@task
-def print_number(num):
-    print("=" * 50)
-    print("Value: {}".format(num))
-    print("=" * 50)
-
-
-schedule = IntervalSchedule(
-    start_date=datetime.datetime.utcnow(), interval=datetime.timedelta(minutes=1)
-)
-
-
-with Flow("cached-task", schedule=schedule) as flow:
-    result = print_number(return_random_number)
-
-
-flow.run()
-## ==================================================
-## Value: 0.8246312081499598
-## ==================================================
-## ==================================================
-## Value: 0.8246312081499598
-## ==================================================
-## ==================================================
-## Value: 0.36999574748592756
-## ==================================================
-```
diff --git a/docs/core/examples/daily_github_stats_to_airtable.md b/docs/core/examples/daily_github_stats_to_airtable.md
deleted file mode 100644
index 183fb7e505..0000000000
--- a/docs/core/examples/daily_github_stats_to_airtable.md
+++ /dev/null
@@ -1,56 +0,0 @@
----
-sidebarDepth: 0
----
-
-# Collect Daily Github Stats
-
-This simple ETL-style flow retrieves the number of GitHub "stargazers" and "watchers" for the Prefect repo and writes them to an
-Airtable document every day.
-
-
-```python
-import datetime
-import pendulum
-
-from prefect import Flow, task
-from prefect.triggers import any_failed
-from prefect.schedules import CronSchedule
-from prefect.tasks.airtable import WriteAirtableRow
-from prefect.tasks.github import GetRepoInfo
-
-
-repo_stats = GetRepoInfo(
-    name="Pull star counts",
-    repo="PrefectHQ/prefect",
-    info_keys=["stargazers_count", "subscribers_count"],
-    max_retries=1,
-    retry_delay=datetime.timedelta(minutes=1),
-)
-
-
-@task
-def process_stats(stats):
-    data = {
-        "Stars": stats["stargazers_count"],
-        "Watchers": stats["subscribers_count"],
-        "Date": pendulum.now("utc").isoformat(),
-    }
-    return data
-
-
-airtable = WriteAirtableRow(
-    base_key="XXXXXXX",
-    table_name="Stars",
-    max_retries=1,
-    retry_delay=datetime.timedelta(minutes=1),
-)
-daily_schedule = CronSchedule("0 8 * * *")
-
-
-with Flow("Collect Repo Stats", schedule=daily_schedule) as flow:
-    data = process_stats(repo_stats)
-    final = airtable(data)
-
-
-flow.run()
-```
diff --git a/docs/core/examples/etl.md b/docs/core/examples/etl.md
deleted file mode 100644
index 80545f192a..0000000000
--- a/docs/core/examples/etl.md
+++ /dev/null
@@ -1,34 +0,0 @@
----
-sidebarDepth: 0
----
-
-# ETL Flow
-
-A simple flow representing an ETL pipeline.
-
-```python
-from prefect import Flow, task
-
-
-@task
-def extract():
-    return [1, 2, 3]
-
-
-@task
-def transform(x):
-    return [i * 10 for i in x]
-
-
-@task
-def load(y):
-    print("Received y: {}".format(y))
-
-
-with Flow("ETL") as flow:
-    e = extract()
-    t = transform(e)
-    l = load(t)
-
-flow.run()
-```
diff --git a/docs/core/examples/flow_state_handler_viz.md b/docs/core/examples/flow_state_handler_viz.md
deleted file mode 100644
index fca3a1e0f2..0000000000
--- a/docs/core/examples/flow_state_handler_viz.md
+++ /dev/null
@@ -1,57 +0,0 @@
----
-sidebarDepth: 0
----
-
-# Visualization with a Flow State Handler
-
-A simple example Flow that stores the current flow visualization to a file
-each time the flow changes state, using a Flow-level state handler.
-
-Uses the same flow from "Retries w/ Mapping" on a minute schedule.
-
-```python
-import random
-from datetime import datetime, timedelta
-
-from prefect import Flow, task
-from prefect.schedules import IntervalSchedule
-
-
-def visualize(flow, old_state, new_state):
-    """
-    Flow state handler that stores the current flow
-    visualization to a known file location, for live viewing.
-    """
-    if isinstance(new_state.result, dict):
-        ## note that the relevant file will actually be called "known_location.pdf"
-        flow.visualize(flow_state=new_state, filename="known_location")
-    return new_state
-
-
-@task
-def generate_random_list():
-    n = random.randint(15, 25)
-    return list(range(n))
-
-
-@task(max_retries=3, retry_delay=timedelta(seconds=0))
-def randomly_fail():
-    x = random.random()
-    if x > 0.7:
-        raise ValueError("x is too large")
-
-
-schedule = IntervalSchedule(
-    start_date=datetime.utcnow(),
-    interval=timedelta(minutes=1),
-    end_date=datetime.utcnow() + timedelta(minutes=10),
-)
-
-with Flow(
-    "random-mapping-with-viz", schedule=schedule, state_handlers=[visualize]
-) as f:
-    randomly_fail.map(upstream_tasks=[generate_random_list])
-
-
-f.run()
-```
diff --git a/docs/core/examples/functional_docker.md b/docs/core/examples/functional_docker.md
deleted file mode 100644
index c88658668e..0000000000
--- a/docs/core/examples/functional_docker.md
+++ /dev/null
@@ -1,42 +0,0 @@
-# Docker Pipeline: Functional API
-
-This Prefect Flow creates a Container based on the latest prefect image, and
-executes an empty Flow inside that container.  Make sure to pull `prefecthq/prefect` prior
-to running this Flow.
-
-```python
-from prefect import Flow
-from prefect.tasks.docker import (
-    CreateContainer,
-    StartContainer,
-    GetContainerLogs,
-    WaitOnContainer,
-)
-from prefect.triggers import always_run
-
-
-## initialize tasks
-container = CreateContainer(
-    image_name="prefecthq/prefect",
-    command='''python -c "from prefect import Flow; f = Flow('empty'); f.run()"''',
-)
-start = StartContainer()
-logs = GetContainerLogs(trigger=always_run)
-status_code = WaitOnContainer()
-
-
-## set task dependencies via functional API
-
-with Flow("Run a Prefect Flow in Docker") as flow:
-    start_container = start(container_id=container)
-    code = status_code(container_id=container, upstream_tasks=[start_container])
-    collect_logs = logs(container_id=container, upstream_tasks=[code])
-
-## run flow and print logs
-flow_state = flow.run()
-
-print("=" * 30)
-print("Container Logs")
-print("=" * 30)
-print(flow_state.result[collect_logs].result)
-```
diff --git a/docs/core/examples/github_release_cycle.md b/docs/core/examples/github_release_cycle.md
deleted file mode 100644
index 747eec39ba..0000000000
--- a/docs/core/examples/github_release_cycle.md
+++ /dev/null
@@ -1,58 +0,0 @@
----
-sidebarDepth: 0
----
-
-# Github Biweekly Release Cycle
-
-This simple flow attempts to prompt a biweekly release by opening a PR from dev -> master every other Monday.
-
-If for any reason the PR fails to open, the Flow opens an issue alerting the team, with
-relevant debug information.
-
-```python
-import datetime
-import pendulum
-
-from prefect import Flow, task
-from prefect.triggers import any_failed
-from prefect.schedules import IntervalSchedule
-from prefect.tasks.github import CreateGitHubPR, OpenGitHubIssue
-
-
-pr_task = CreateGitHubPR(
-    name="Open dev->master PR",
-    repo="PrefectHQ/cloud",
-    base="master",
-    head="dev",
-    title="Bi-weekly Release",
-    max_retries=1,
-    retry_delay=datetime.timedelta(minutes=1),
-)
-
-
-@task(trigger=any_failed)
-def prepare_exception(exc):
-    return repr(exc)
-
-
-issue_task = OpenGitHubIssue(
-    name="Open Release Issue",
-    repo="PrefectHQ/cloud",
-    title="Release Cycle is Broken",
-    labels=["release", "bug"],
-)
-
-
-biweekly_schedule = IntervalSchedule(
-    start_date=pendulum.parse("2019-03-18"), interval=datetime.timedelta(days=14)
-)
-
-
-with Flow("Biweekly Cloud Release", schedule=biweekly_schedule) as flow:
-    exc = prepare_exception(pr_task)  # will only run if pr_task fails in some way
-    issue = issue_task(body=exc)
-
-
-flow.set_reference_tasks([pr_task])
-flow.run()
-```
diff --git a/docs/core/examples/imperative_docker.md b/docs/core/examples/imperative_docker.md
deleted file mode 100644
index 028ce908c9..0000000000
--- a/docs/core/examples/imperative_docker.md
+++ /dev/null
@@ -1,44 +0,0 @@
-# Docker Pipeline: Imperative API
-
-This Prefect Flow creates a Container based on the latest prefect image, and
-executes an empty Flow inside that container.  Make sure to pull `prefecthq/prefect` prior
-to running this Flow.
-
-```python
-from prefect import Flow
-from prefect.tasks.docker import (
-    CreateContainer,
-    StartContainer,
-    GetContainerLogs,
-    WaitOnContainer,
-)
-from prefect.triggers import always_run
-
-
-container = CreateContainer(
-    image_name="prefecthq/prefect",
-    command='''python -c "from prefect import Flow; f = Flow('empty'); f.run()"''',
-)
-start = StartContainer()
-logs = GetContainerLogs(trigger=always_run)
-status_code = WaitOnContainer()
-
-
-flow = Flow("Run a Prefect Flow in Docker")
-
-## set individual task dependencies using imperative API
-start.set_upstream(container, flow=flow, key="container_id")
-logs.set_upstream(container, flow=flow, key="container_id")
-status_code.set_upstream(container, flow=flow, key="container_id")
-
-status_code.set_upstream(start, flow=flow)
-logs.set_upstream(status_code, flow=flow)
-
-## run flow and print logs
-flow_state = flow.run()
-
-print("=" * 30)
-print("Container Logs")
-print("=" * 30)
-print(flow_state.result[logs].result)
-```
diff --git a/docs/core/examples/map_reduce.md b/docs/core/examples/map_reduce.md
deleted file mode 100644
index 2c4b2ed95f..0000000000
--- a/docs/core/examples/map_reduce.md
+++ /dev/null
@@ -1,46 +0,0 @@
----
-sidebarDepth: 0
----
-
-# Simple Map/Reduce
-
-A simple map-reduce style flow.
-
-```python
-from prefect import Flow, task
-
-# ------------------------------------
-# define some tasks
-
-
-@task
-def numbers_task():
-    return [1, 2, 3]
-
-
-@task
-def map_task(x):
-    return x + 1
-
-
-@task
-def reduce_task(x):
-    return sum(x)
-
-
-# ------------------------------------
-# build a flow
-
-with Flow("Map / Reduce 🤓") as flow:
-    numbers = numbers_task()
-    first_map = map_task.map(numbers)
-    second_map = map_task.map(first_map)
-    reduction = reduce_task(second_map)
-
-
-# ------------------------------------
-# run the flow
-
-state = flow.run()
-assert state.result[reduction].result == 12
-```
diff --git a/docs/core/examples/overview.md b/docs/core/examples/overview.md
new file mode 100644
index 0000000000..348bd8db23
--- /dev/null
+++ b/docs/core/examples/overview.md
@@ -0,0 +1,36 @@
+# Overview
+
+Prefect includes a number of examples covering different features. These can be
+viewed live in the docs, or accessed from the GitHub repo
+[here](https://github.com/PrefectHQ/prefect/tree/master/examples).
+
+## Running with Prefect Cloud or Server
+
+When running with Prefect Cloud or Prefect Server, you can register the
+examples in a new project with the Prefect CLI. You can either register all the
+examples at once, or select specific examples by name.
+
+```
+# Create a new project named "Prefect Examples"
+$ prefect create project "Prefect Examples"
+
+# Register all the examples into the "Prefect Examples" project
+$ prefect register --json https://docs.prefect.io/examples.json \
+    --project "Prefect Examples"
+
+# OR register only specific examples by specifying them by name
+$ prefect register --json https://docs.prefect.io/examples.json \
+    --project "Prefect Examples" \
+    --name "Example: Parameters" \
+    --name "Example: Mapping"
+```
+
+These can then be run using any agent with a ``prefect-examples`` label. For
+example, to start a local agent for running the examples:
+
+```
+$ prefect agent local start -l prefect-examples
+```
+
+If you haven't already, we recommend going through the [Orchestration
+Tutorial](/orchestration/tutorial/overview.md) beforehand.
diff --git a/docs/core/examples/parameterized_flow.md b/docs/core/examples/parameterized_flow.md
deleted file mode 100644
index a05d8980a0..0000000000
--- a/docs/core/examples/parameterized_flow.md
+++ /dev/null
@@ -1,48 +0,0 @@
----
-sidebarDepth: 0
----
-
-# Parameterized Conditions
-
-This is a simple flow that takes in a required parameter `value` and determines if it is even or odd.
-
-This utilizes one of the control flow tasks from the Prefect task library for evaluating
-the conditional returned from the `check_if_even` task.
-
-```python
-from prefect import task, Flow, Parameter
-from prefect.tasks.control_flow import ifelse
-
-
-@task
-def check_if_even(value):
-    return (value % 2 == 0)
-
-
-@task
-def print_odd(value):
-    print("{} is odd!".format(value))
-
-
-@task
-def print_even(value):
-    print("{} is even!".format(value))
-
-
-with Flow("Check Even/Odd") as f:
-    value = Parameter("value")
-    is_even = check_if_even(value)
-
-    even = print_even(value)
-    odd = print_odd(value)
-
-    ifelse(is_even, even, odd)
-
-
-# Prints '2 is even!'
-f.run(value=2)
-
-
-# Prints '1 is odd!'
-f.run(value=1)
-```
diff --git a/docs/core/examples/retries_with_mapping.md b/docs/core/examples/retries_with_mapping.md
deleted file mode 100644
index b7ac5a0ad9..0000000000
--- a/docs/core/examples/retries_with_mapping.md
+++ /dev/null
@@ -1,43 +0,0 @@
----
-sidebarDepth: 0
----
-
-# Retries and Mapping
-
-This example showcases a few basic Prefect concepts:
-
-- the ability to "map" tasks across the dynamic output of upstream tasks; additionally, it is not required that the
-    upstream task actually passes data to the downstream task, as this example demonstrates
-- task retries: `flow.run()` will perform retries, on schedule, for all tasks that require it,
-    including individual mapped tasks
-
-This flow first generates a list of random length, and then maps over that list to spawn a dynamic number of
-downstream tasks that randomly fail.  The takeaway here is that we don't have to know a-priori how many mapped tasks
-will be created prior to execution!  Additionally, each failed mapped task will retry on its own.
-
-```python
-import random
-from datetime import timedelta
-
-from prefect import task, Flow
-
-
-@task
-def generate_random_list():
-    n = random.randint(15, 25)
-    return list(range(n))
-
-
-@task(max_retries=3, retry_delay=timedelta(seconds=0))
-def randomly_fail():
-    x = random.random()
-    if x > 0.7:
-        raise ValueError("x is too large")
-
-
-with Flow("random-mapping") as f:
-    final = randomly_fail.map(upstream_tasks=[generate_random_list])
-
-# should see logs suggesting that some tasks are failing and retrying
-f.run()
-```
diff --git a/docs/core/examples/state_handler_logging.md b/docs/core/examples/state_handler_logging.md
deleted file mode 100644
index 8e73135d0d..0000000000
--- a/docs/core/examples/state_handler_logging.md
+++ /dev/null
@@ -1,45 +0,0 @@
----
-sidebarDepth: 0
----
-
-# Logging with a State Handler
-
-A simple Flow that logs the duration of task runs
-by using a task state handler.
-
-```python
-import pendulum
-import time
-
-from prefect import task, Flow
-
-
-def timestamper(task, old_state, new_state):
-    """
-    Task state handler that timestamps new states
-    and logs the duration between state changes using
-    the task's logger.
-    """
-    new_state.timestamp = pendulum.now("utc")
-    if hasattr(old_state, "timestamp"):
-        duration = (new_state.timestamp - old_state.timestamp).in_seconds()
-        task.logger.info(
-            "{} seconds passed in between state transitions".format(duration)
-        )
-    return new_state
-
-
-@task(state_handlers=[timestamper])
-def sleeper():
-    time.sleep(2)
-
-
-f = Flow("log-task-duration", tasks=[sleeper])
-f.run()
-# INFO - prefect.FlowRunner | Beginning Flow run for 'log-task-duration'
-# INFO - prefect.FlowRunner | Starting flow run.
-# INFO - prefect.TaskRunner | Task 'sleeper': Starting task run...
-# INFO - prefect.Task | 2 seconds passed in between state transitions
-# INFO - prefect.TaskRunner | Task 'sleeper': finished task run for task with final state: 'Success'
-# INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded
-```
diff --git a/docs/core/examples/task_looping.md b/docs/core/examples/task_looping.md
deleted file mode 100644
index 84dab68f95..0000000000
--- a/docs/core/examples/task_looping.md
+++ /dev/null
@@ -1,45 +0,0 @@
----
-sidebarDepth: 0
----
-
-# Task Looping
-
-This Flow uses Prefect's Task Looping feature for computing
-the largest Fibonacci number less than a given number M. Each
-Fibonacci number is computed using a black-box external API.
-
-```python
-import requests
-from datetime import timedelta
-
-import prefect
-from prefect import task, Flow, Parameter
-from prefect.engine.signals import LOOP
-
-
-@task(max_retries=5, retry_delay=timedelta(seconds=2))
-def compute_large_fibonacci(M):
-    # we extract the accumulated task loop result from context
-    loop_payload = prefect.context.get("task_loop_result", {})
-
-    n = loop_payload.get("n", 1)
-    fib = loop_payload.get("fib", 1)
-
-    next_fib = requests.post(
-        "https://nemo.api.stdlib.com/fibonacci@0.0.1/", data={"nth": n}
-    ).json()
-
-    if next_fib > M:
-        return fib  # return statements end the loop
-
-    raise LOOP(message=f"Fib {n}={next_fib}", result=dict(n=n + 1, fib=next_fib))
-
-
-with Flow("fibonacci") as flow:
-    M = Parameter("M")
-    fib_num = compute_large_fibonacci(M)
-
-
-flow_state = flow.run(M=100)
-print(flow_state.result[fib_num].result)  # 89
-```
diff --git a/docs/core/examples/twitter_to_airtable.md b/docs/core/examples/twitter_to_airtable.md
deleted file mode 100644
index aaa85c36b3..0000000000
--- a/docs/core/examples/twitter_to_airtable.md
+++ /dev/null
@@ -1,95 +0,0 @@
----
-sidebarDepth: 0
----
-
-# Send Tweets to Airtable
-
-
-This example is [based on work by Matt Dickenson](https://mattdickenson.com/2019/03/02/extract-replies-to-tweet/)
-and shows how to use Prefect to build a flow that loads replies to a tweet, then inserts those
-replies into Airtable.
-
-This flow relies heavily on local secrets to authenticate with both Twitter and Airtable; when used
-with Prefect Cloud, all credentials can be removed.
-
-```python
-import prefect
-from prefect import Flow, Parameter, task, unmapped
-from prefect.tasks.airtable import WriteAirtableRow
-from prefect.tasks.twitter import LoadTweetReplies
-
-# ------------------------------------------------------------
-# Secrets
-#
-# Fill out your credentials here for the local runs
-# ------------------------------------------------------------
-
-TWITTER_CREDS = {
-    "api_key": "🤫",
-    "api_secret": "🤫",
-    "access_token": "🤫",
-    "access_token_secret": "🤫",
-}
-
-AIRTABLE_API_KEY = "🤫"
-
-# ------------------------------------------------------------
-# Build the flow
-#
-# For this flow, we:
-#   - create parameters
-#   - create a small task to reformat tweets
-#   - wire up the flow with our parameters, reformatter, and two task library tasks
-# ------------------------------------------------------------
-
-twitter_user = Parameter("twitter_user")
-tweet_id = Parameter("tweet_id")
-airtable_base = Parameter("airtable_base")
-airtable_table = Parameter("airtable_table")
-
-
-@task
-def format_tweet(tweet) -> dict:
-    """
-    This task takes a tweet object and returns a dictionary containing its `user` and `text`
-    """
-    return dict(user=tweet.user.screen_name, text=tweet.text)
-
-
-with Flow("Tweets to Airtable") as flow:
-
-    # load tweets
-    replies = LoadTweetReplies()(user=twitter_user, tweet_id=tweet_id)
-
-    # map over each tweet to format it for Airtable
-    formatted_replies = format_tweet.map(replies)
-
-    # map over each formatted tweet and insert it to Airtable
-    WriteAirtableRow().map(
-        formatted_replies,
-        base_key=unmapped(airtable_base),
-        table_name=unmapped(airtable_table),
-    )
-
-
-# ------------------------------------------------------------
-# Run the flow
-#
-# Because we're running this flow locally, we create a secrets context to provide
-# our credentials, then call flow.run() with our parameter values
-# ------------------------------------------------------------
-# set up the local secrets
-with prefect.context(
-    secrets=dict(
-        TWITTER_API_CREDENTIALS=TWITTER_CREDS, AIRTABLE_API_KEY=AIRTABLE_API_KEY
-    )
-):
-
-    # run the flow
-    state = flow.run(
-        twitter_user="PrefectIO",
-        tweet_id="1046873748559814656",
-        airtable_base="<AN AIRTABLE BASE>",
-        airtable_table="<AN AIRTABLE TABLE>",
-    )
-```
diff --git a/docs/generate_docs.py b/docs/generate_docs.py
index 98affdc3e9..1727c4b10c 100644
--- a/docs/generate_docs.py
+++ b/docs/generate_docs.py
@@ -14,13 +14,18 @@ Each entry in `OUTLINE` is a dictionary with the following key/value pairs:
 
 On a development installation of Prefect, run `python generate_docs.py` from inside the `docs/` folder.
 """
+import ast
 import builtins
+import glob
 import html
 import importlib
 import inspect
+import json
 import os
 import re
 import shutil
+import subprocess
+import sys
 import textwrap
 from contextlib import contextmanager
 from functools import partial
@@ -36,6 +41,8 @@ from tokenizer import format_code
 OUTLINE_PATH = os.path.join(os.path.dirname(__file__), "outline.toml")
 outline_config = toml.load(OUTLINE_PATH)
 
+ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+
 
 @contextmanager
 def patch_imports():
@@ -233,7 +240,6 @@ def create_commands_table(commands):
 
 
 def format_command_doc(name, ctx, cmd):
-    table = f"### {name}\n"
     help_text = cmd.get_help(ctx).split("\n", 2)[2]
     # CLI commands with handwritten help sections will
     # contain two `Options` sections, drop one.
@@ -351,6 +357,129 @@ def get_class_methods(obj, methods=None):
         return [getattr(obj, m) for m in methods]
 
 
+EXAMPLE_TEMPLATE = """
+---
+editLink: false
+---
+
+{header}
+
+::: tip Registering with Prefect Cloud/Server
+
+This example can be registered in Prefect Cloud or Server by running:
+
+```
+{register_cmd}
+```
+
+(to register in a different project, replace `'Prefect Examples'` with your project name).
+:::
+
+```python
+{source}
+```
+
+::: details Output
+```
+$ python {relpath}
+{output}
+```
+:::
+
+*The flow source is available on GitHub [here](https://github.com/PrefectHQ/prefect/blob/{ref}/{relpath}).*
+"""
+
+
+def build_example(path):
+    """Build an example located at a specific path.
+
+    Args:
+        - path (str): the path to the example source file.
+
+    Returns:
+        - markdown (str): the rendered example in markdown
+        - flows (Dict[str, Flow]): the flows found in the example
+    """
+    from prefect import Flow
+    from prefect.storage import GitHub
+    from prefect.run_configs import UniversalRun
+
+    # Use the current commit (if specified in the environment)
+    ref = os.getenv("GIT_SHA", "master")
+
+    with open(path, "r", encoding="utf-8") as f:
+        contents = f.read()
+
+    namespace = {}
+    exec(contents, namespace)
+
+    try:
+        header = namespace["__doc__"]
+        tree = ast.parse(contents)
+        offset = tree.body[1].lineno - 1
+    except Exception as exc:
+        raise ValueError(f"No docstring header found for example at {path}") from exc
+
+    flows = {}
+    relpath = os.path.relpath(path, start=ROOT)
+    for f in namespace.values():
+        if isinstance(f, Flow):
+            f.storage = GitHub("PrefectHQ/prefect", path=relpath, ref=ref)
+            if not f.run_config:
+                f.run_config = UniversalRun()
+            f.run_config.labels.add("prefect-examples")
+            flows[f.name] = f.serialize()
+
+    source = "\n".join(contents.splitlines()[offset:]).strip()
+
+    res = subprocess.run(
+        [sys.executable, path],
+        capture_output=True,
+        check=True,
+        env={"PREFECT__LOGGING__FORMAT": "%(levelname)s | %(message)s"},
+    )
+    output = res.stdout.decode("utf-8").strip()
+
+    register_lines = [f"prefect register --json https://docs.prefect.io/examples.json"]
+    for name in sorted(flows):
+        register_lines.append(f"    --name {name!r}")
+    register_lines.append(f"    --project 'Prefect Examples'")
+
+    rendered = EXAMPLE_TEMPLATE.format(
+        header=header,
+        source=source,
+        output=output,
+        ref=ref,
+        relpath=relpath,
+        register_cmd=" \\\n".join(register_lines),
+    ).lstrip()
+
+    return rendered, flows
+
+
+def process_examples(footer=""):
+    """Build and render all examples found in the `examples/` directory"""
+    flows = {}
+    for path in glob.glob(os.path.join(ROOT, "examples", "*.py")):
+        filename = os.path.splitext(os.path.basename(path))[0]
+        output, new_flows = build_example(path)
+        conflicts = set(flows).intersection(new_flows)
+        if conflicts:
+            raise ValueError(
+                "Example flows must have unique names, found duplicate flows: {conflicts}"
+            )
+        flows.update(new_flows)
+        with open(
+            os.path.join("core", "examples", filename + ".md"), "w", encoding="utf-8"
+        ) as f:
+            f.write(output)
+            f.write(footer)
+
+    flows = [flows[k] for k in sorted(flows)]
+    with open(os.path.join(".vuepress", "public", "examples.json"), "wb") as f:
+        f.write(json.dumps({"version": 1, "flows": flows}).encode("utf-8"))
+
+
 def create_tutorial_notebooks(tutorial):
     """
     Utility that automagically creates an .ipynb notebook file from a markdown file consisting
@@ -465,6 +594,9 @@ if __name__ == "__main__":
                 f.write(changelog)
                 f.write(auto_generated_footer)
 
+        # Generate examples
+        process_examples(auto_generated_footer)
+
         for page in OUTLINE:
             # collect what to document
             fname, classes, fns, cmds = (
diff --git a/docs/orchestration/tutorial/next-steps.md b/docs/orchestration/tutorial/next-steps.md
index 84f20e7c39..1b829c1433 100644
--- a/docs/orchestration/tutorial/next-steps.md
+++ b/docs/orchestration/tutorial/next-steps.md
@@ -12,6 +12,25 @@ In this tutorial we covered:
 
 This is just the beginning; the Prefect API has many more features to explore!
 
+## Examples
+
+Prefect provides a number of [examples](/core/examples/overview.md) that illustrate
+different aspects of developing and running flows. These examples can all be run
+locally or through Prefect Cloud/Server. To create a new project and register all
+examples, run the following:
+
+You can register all the examples in a new project by running the following:
+
+```
+# Create a new "Prefect Examples" project
+$ prefect create project "Prefect Examples"
+
+# Register all the examples into the "Prefect Examples" project
+$ prefect register --json https://docs.prefect.io/examples.json --project "Prefect Examples"
+```
+
+See the [examples](/core/examples/overview.md) page for more information.
+
 ## Concepts
 
 Visit the [Concept](/orchestration/concepts/api.html) docs for actions such as
diff --git a/examples/conditional.py b/examples/conditional.py
new file mode 100644
index 0000000000..250a946347
--- /dev/null
+++ b/examples/conditional.py
@@ -0,0 +1,79 @@
+"""
+# Conditional Tasks
+
+Sometimes you have parts of a flow that you only want to run under certain
+conditions. To support this, Prefect provides several built-in [tasks for
+control-flow](/core/task_library/control_flow.md) that you can use to add
+conditional branches to your flow.
+
+Let's say you want to have a flow where different tasks are run based on the
+result of some conditional task.  In normal Python code this logic might look
+like:
+
+```python
+if check_condition():
+    val = action_if_true()
+else:
+    val = action_if_false()
+
+another_action(val)
+```
+
+To implement the same logic as a Flow, you can make use of Prefect's `case`
+blocks. Tasks added to a flow inside a `case` block are only run if the
+condition matches the case. Tasks in branches that aren't run will finish with
+a `Skipped` state.
+
+We'll also need Prefect's `merge` function to merge the branches together
+before the `another_action` task. This function takes one or more tasks
+conditional tasks, and returns the output of the first task that isn't
+`Skipped`. This can be used to "merge" multiple branches in a flow together. If
+the results of `action_if_true`/`action_if_false` weren't needed later on in
+the flow, we wouldn't need this step.
+
+The resulting flow looks like:
+
+![Flow with conditional branches](/idioms/conditional-branches-merge.png)
+"""
+from random import random
+
+from prefect import task, Flow, case
+from prefect.tasks.control_flow import merge
+
+
+@task
+def check_condition():
+    return random() < 0.5
+
+
+@task
+def action_if_true():
+    return "I am true!"
+
+
+@task
+def action_if_false():
+    return "I am false!"
+
+
+@task(log_stdout=True)
+def another_action(val):
+    print(val)
+
+
+with Flow("Example: Conditional Tasks") as flow:
+    cond = check_condition()
+
+    with case(cond, True):
+        val_if_true = action_if_true()
+
+    with case(cond, False):
+        val_if_false = action_if_false()
+
+    val = merge(val_if_true, val_if_false)
+
+    another_action(val)
+
+
+if __name__ == "__main__":
+    flow.run()
diff --git a/examples/mapping.py b/examples/mapping.py
new file mode 100644
index 0000000000..0a6654ed4a
--- /dev/null
+++ b/examples/mapping.py
@@ -0,0 +1,98 @@
+"""
+# Mapping
+
+[Mapping](/core/concepts/mapping.md) in Prefect can be used to apply the same
+task (or tasks) to multiple inputs. To map a task, use its `.map` method
+instead of calling the task itself.
+
+```python
+from prefect import task
+
+@task
+def add(x, y):
+    return x + y
+
+add(1, 2)  # 3
+add.map([1, 10], [2, 3])  # [3, 13]
+```
+
+By default all arguments to `.map` expect iterables that can be mapped over. If
+you want to pass in a non-iterable argument to be used by all branches in a
+mapped task, you can wrap that argument in `unmapped`
+([docs](/core/concepts/mapping.md#unmapped-inputs)).
+
+```python
+from prefect import unmapped
+
+add.map([1, 10], unmapped(2))  # [3, 12]
+```
+
+In this example we build a flow with 4 stages:
+
+- `get_numbers` returns a list of `n` numbers
+- `inc` maps across this list, running once per input
+- `add` maps across the output of `inc`, running once per input. `unmapped` is
+  used to pass `2` as a second "unmapped" argument.
+- The total list is then passed to `compute_sum` to calculate a total.
+
+```
+             ┌── inc ── add(2) ──┐
+             │                   │
+get_numbers ─┼── inc ── add(2) ──┼─ compute_sum
+             │                   │
+             ⋮                   ⋮
+             └── inc ── add(2) ──┘
+```
+
+Note that the number of inputs to a mapped task doesn't need to be known until
+runtime. By default this flow will run with `3` mapped branches - try
+providing different values for the parameter `n` to see how it affects
+execution.
+
+See the [mapping docs](/core/concepts/mapping.md) for more information.
+"""
+
+from prefect import Flow, Parameter, task, unmapped
+
+
+@task
+def get_numbers(n):
+    return range(1, n + 1)
+
+
+@task
+def inc(x):
+    return x + 1
+
+
+@task
+def add(x, y):
+    return x + y
+
+
+@task(log_stdout=True)
+def compute_sum(nums):
+    total = sum(nums)
+    print(f"total = {total}")
+    return total
+
+
+with Flow("Example: Mapping") as flow:
+    # The number of branches in the mapped pipeline
+    n = Parameter("n", default=3)
+
+    # Generate the initial items to map over
+    nums = get_numbers(n)  # [1, 2, 3]
+
+    # Apply `inc` to every item in `nums`
+    nums_2 = inc.map(nums)  # [2, 3, 4]
+
+    # Apply `add` to every item in `nums_2`, with `2` as the second argument.
+    nums_3 = add.map(nums_2, unmapped(2))  # [4, 5, 6]
+
+    # Compute the sum of all items in `nums_3`
+    total = compute_sum(nums_3)  # 15
+
+
+if __name__ == "__main__":
+    flow.run()
diff --git a/examples/airflow_tutorial_dag.py b/examples/old/airflow_tutorial_dag.py
similarity index 100%
rename from examples/airflow_tutorial_dag.py
rename to examples/old/airflow_tutorial_dag.py
diff --git a/examples/cached_task.py b/examples/old/cached_task.py
similarity index 100%
rename from examples/cached_task.py
rename to examples/old/cached_task.py
diff --git a/examples/daily_github_stats_to_airtable.py b/examples/old/daily_github_stats_to_airtable.py
similarity index 100%
rename from examples/daily_github_stats_to_airtable.py
rename to examples/old/daily_github_stats_to_airtable.py
diff --git a/examples/etl.py b/examples/old/etl.py
similarity index 100%
rename from examples/etl.py
rename to examples/old/etl.py
diff --git a/examples/feature_engineering.py b/examples/old/feature_engineering.py
similarity index 100%
rename from examples/feature_engineering.py
rename to examples/old/feature_engineering.py
diff --git a/examples/flow_state_handler_viz.py b/examples/old/flow_state_handler_viz.py
similarity index 100%
rename from examples/flow_state_handler_viz.py
rename to examples/old/flow_state_handler_viz.py
diff --git a/examples/functional_docker.py b/examples/old/functional_docker.py
similarity index 100%
rename from examples/functional_docker.py
rename to examples/old/functional_docker.py
diff --git a/examples/github_release_cycle.py b/examples/old/github_release_cycle.py
similarity index 100%
rename from examples/github_release_cycle.py
rename to examples/old/github_release_cycle.py
diff --git a/examples/imperative_docker.py b/examples/old/imperative_docker.py
similarity index 100%
rename from examples/imperative_docker.py
rename to examples/old/imperative_docker.py
diff --git a/examples/map_reduce.py b/examples/old/map_reduce.py
similarity index 100%
rename from examples/map_reduce.py
rename to examples/old/map_reduce.py
diff --git a/examples/parameterized_flow.py b/examples/old/parameterized_flow.py
similarity index 100%
rename from examples/parameterized_flow.py
rename to examples/old/parameterized_flow.py
diff --git a/examples/retries_with_mapping.py b/examples/old/retries_with_mapping.py
similarity index 100%
rename from examples/retries_with_mapping.py
rename to examples/old/retries_with_mapping.py
diff --git a/examples/state_handler_logging.py b/examples/old/state_handler_logging.py
similarity index 100%
rename from examples/state_handler_logging.py
rename to examples/old/state_handler_logging.py
diff --git a/examples/task_library/google/storage.py b/examples/old/task_library/google/storage.py
similarity index 100%
rename from examples/task_library/google/storage.py
rename to examples/old/task_library/google/storage.py
diff --git a/examples/task_library/great_expectations/data/npidata/npidata_pfile_20190902-20190908.csv b/examples/old/task_library/great_expectations/data/npidata/npidata_pfile_20190902-20190908.csv
similarity index 100%
rename from examples/task_library/great_expectations/data/npidata/npidata_pfile_20190902-20190908.csv
rename to examples/old/task_library/great_expectations/data/npidata/npidata_pfile_20190902-20190908.csv
diff --git a/examples/task_library/great_expectations/data/npidata/npidata_pfile_20190909-20190915.csv b/examples/old/task_library/great_expectations/data/npidata/npidata_pfile_20190909-20190915.csv
similarity index 100%
rename from examples/task_library/great_expectations/data/npidata/npidata_pfile_20190909-20190915.csv
rename to examples/old/task_library/great_expectations/data/npidata/npidata_pfile_20190909-20190915.csv
diff --git a/examples/task_library/great_expectations/great_expectations/.gitignore b/examples/old/task_library/great_expectations/great_expectations/.gitignore
similarity index 100%
rename from examples/task_library/great_expectations/great_expectations/.gitignore
rename to examples/old/task_library/great_expectations/great_expectations/.gitignore
diff --git a/examples/task_library/great_expectations/great_expectations/checkpoints/guaranteed_failure_npi_checkpoint.yml b/examples/old/task_library/great_expectations/great_expectations/checkpoints/guaranteed_failure_npi_checkpoint.yml
similarity index 100%
rename from examples/task_library/great_expectations/great_expectations/checkpoints/guaranteed_failure_npi_checkpoint.yml
rename to examples/old/task_library/great_expectations/great_expectations/checkpoints/guaranteed_failure_npi_checkpoint.yml
diff --git a/examples/task_library/great_expectations/great_expectations/checkpoints/guaranteed_failure_npi_checkpoint_circle.yml b/examples/old/task_library/great_expectations/great_expectations/checkpoints/guaranteed_failure_npi_checkpoint_circle.yml
similarity index 100%
rename from examples/task_library/great_expectations/great_expectations/checkpoints/guaranteed_failure_npi_checkpoint_circle.yml
rename to examples/old/task_library/great_expectations/great_expectations/checkpoints/guaranteed_failure_npi_checkpoint_circle.yml
diff --git a/examples/task_library/great_expectations/great_expectations/checkpoints/warning_npi_checkpoint.yml b/examples/old/task_library/great_expectations/great_expectations/checkpoints/warning_npi_checkpoint.yml
similarity index 100%
rename from examples/task_library/great_expectations/great_expectations/checkpoints/warning_npi_checkpoint.yml
rename to examples/old/task_library/great_expectations/great_expectations/checkpoints/warning_npi_checkpoint.yml
diff --git a/examples/task_library/great_expectations/great_expectations/checkpoints/warning_npi_checkpoint_circle.yml b/examples/old/task_library/great_expectations/great_expectations/checkpoints/warning_npi_checkpoint_circle.yml
similarity index 100%
rename from examples/task_library/great_expectations/great_expectations/checkpoints/warning_npi_checkpoint_circle.yml
rename to examples/old/task_library/great_expectations/great_expectations/checkpoints/warning_npi_checkpoint_circle.yml
diff --git a/examples/task_library/great_expectations/great_expectations/expectations/npidata_pfile_20190902-20190908/guaranteed_to_fail.json b/examples/old/task_library/great_expectations/great_expectations/expectations/npidata_pfile_20190902-20190908/guaranteed_to_fail.json
similarity index 100%
rename from examples/task_library/great_expectations/great_expectations/expectations/npidata_pfile_20190902-20190908/guaranteed_to_fail.json
rename to examples/old/task_library/great_expectations/great_expectations/expectations/npidata_pfile_20190902-20190908/guaranteed_to_fail.json
diff --git a/examples/task_library/great_expectations/great_expectations/expectations/npidata_pfile_20190902-20190908/warning.json b/examples/old/task_library/great_expectations/great_expectations/expectations/npidata_pfile_20190902-20190908/warning.json
similarity index 100%
rename from examples/task_library/great_expectations/great_expectations/expectations/npidata_pfile_20190902-20190908/warning.json
rename to examples/old/task_library/great_expectations/great_expectations/expectations/npidata_pfile_20190902-20190908/warning.json
diff --git a/examples/task_library/great_expectations/great_expectations/great_expectations.yml b/examples/old/task_library/great_expectations/great_expectations/great_expectations.yml
similarity index 100%
rename from examples/task_library/great_expectations/great_expectations/great_expectations.yml
rename to examples/old/task_library/great_expectations/great_expectations/great_expectations.yml
diff --git a/examples/task_library/great_expectations/great_expectations/plugins/custom_data_docs/styles/data_docs_custom_styles.css b/examples/old/task_library/great_expectations/great_expectations/plugins/custom_data_docs/styles/data_docs_custom_styles.css
similarity index 100%
rename from examples/task_library/great_expectations/great_expectations/plugins/custom_data_docs/styles/data_docs_custom_styles.css
rename to examples/old/task_library/great_expectations/great_expectations/plugins/custom_data_docs/styles/data_docs_custom_styles.css
diff --git a/examples/task_library/great_expectations/great_expectations_flow.py b/examples/old/task_library/great_expectations/great_expectations_flow.py
similarity index 100%
rename from examples/task_library/great_expectations/great_expectations_flow.py
rename to examples/old/task_library/great_expectations/great_expectations_flow.py
diff --git a/examples/task_library/mysql/mysql_flow.py b/examples/old/task_library/mysql/mysql_flow.py
similarity index 100%
rename from examples/task_library/mysql/mysql_flow.py
rename to examples/old/task_library/mysql/mysql_flow.py
diff --git a/examples/task_looping.py b/examples/old/task_looping.py
similarity index 100%
rename from examples/task_looping.py
rename to examples/old/task_looping.py
diff --git a/examples/write_tweets_to_airtable.py b/examples/old/write_tweets_to_airtable.py
similarity index 100%
rename from examples/write_tweets_to_airtable.py
rename to examples/old/write_tweets_to_airtable.py
diff --git a/examples/parameters.py b/examples/parameters.py
new file mode 100644
index 0000000000..c18277c863
--- /dev/null
+++ b/examples/parameters.py
@@ -0,0 +1,45 @@
+"""
+# Parameters
+
+A `Parameter` is a special type of `Task` representing an input that can vary
+per flow run. For example:
+
+```python
+x = Parameter("x", default=1)
+```
+
+Parameters have a name (`"x"` in this case), and may optionally include a
+default value. Parameters lacking a default value require an explicit value be
+configured for each flow run. Parameters with a default value may use the
+default, or optionally provide a different value at runtime. Parameters can be
+specified through the UI or CLI when running with Prefect Cloud/Server (see
+[here](/orchestration/tutorial/parameters.md)) or through the `parameters`
+kwarg when running locally with `flow.run`.
+
+For more information, see the [Parameter docs](/core/concepts/parameters.md).
+"""
+
+from prefect import Flow, Parameter, task
+
+
+@task(log_stdout=True)
+def print_total(x, y, total):
+    print(f"{x} + {y} = {total}")
+
+
+with Flow("Example: Parameters") as flow:
+    x = Parameter("x", default=1)
+    y = Parameter("y", default=2)
+
+    print_total(x, y, x + y)
+
+if __name__ == "__main__":
+    # When running a flow where all parameters have default values
+    # no parameters need to be specified at runtime.
+    print("Running with default values")
+    flow.run()
+
+    # One or more parameters can be specified at runtime through the use of the
+    # `parameters` argument. Here we set `x` to 8 and `y` to 9.
+    print("\nRunning with `x = 8` and `y = 9`")
+    flow.run(parameters={"x": 8, "y": 9})
diff --git a/src/prefect/engine/flow_runner.py b/src/prefect/engine/flow_runner.py
index 27717decd8..e741366341 100644
--- a/src/prefect/engine/flow_runner.py
+++ b/src/prefect/engine/flow_runner.py
@@ -155,29 +155,8 @@ class FlowRunner(Runner):
         context.update(flow_name=self.flow.name)
         context.setdefault("scheduled_start_time", pendulum.now("utc"))
 
-        # Determine the current time, allowing our formatted dates in the context
-        # to rely on a manually set value
-        now = context.get("date")
-        if isinstance(now, str):
-            # Attempt to parse into a `DateTime` object since it will often be passed
-            # as a serialized string from the UI we'll override the context on a
-            # successful parse so the type is consistent for users
-            try:
-                now = pendulum.parse(now)
-                context["date"] = now
-            except Exception:
-                pass
-        if not isinstance(now, pendulum.DateTime):
-            if now is not None:
-                self.logger.warning(
-                    "`date` was set in the context manually but could not be parsed "
-                    "into a pendulum `DateTime` object. Additional context variables "
-                    "that rely on the current date i.e `today` and `tomorrow` will be "
-                    "based on the current time instead of the `date` context variable."
-                )
-            now = pendulum.now("utc")
-
         # add various formatted dates to context
+        now = pendulum.now("utc")
         dates = {
             "date": now,
             "today": now.strftime("%Y-%m-%d"),
diff --git a/src/prefect/tasks/databricks/databricks_submitjob.py b/src/prefect/tasks/databricks/databricks_submitjob.py
index cdb00a6346..266bc24bf1 100644
--- a/src/prefect/tasks/databricks/databricks_submitjob.py
+++ b/src/prefect/tasks/databricks/databricks_submitjob.py
@@ -663,26 +663,25 @@ class DatabricksRunNow(Task):
         ), "`databricks_conn_secret` must be supplied as a valid dictionary."
         self.databricks_conn_secret = databricks_conn_secret
 
+        if json:
+            self.json = json
+
         # Initialize Databricks Connections
         hook = self.get_hook()
 
-        run_now_json = json or {}
-
-        if job_id is not None:
-            run_now_json["job_id"] = job_id
-        if notebook_params is not None:
-            merged = run_now_json.setdefault("notebook_params", {})
-            merged.update(notebook_params)
-            run_now_json["notebook_params"] = merged
-        if python_params is not None:
-            run_now_json["python_params"] = python_params
-        if spark_submit_params is not None:
-            run_now_json["spark_submit_params"] = spark_submit_params
-        if jar_params is not None:
-            run_now_json["jar_params"] = jar_params
+        if self.job_id is not None:
+            self.json["job_id"] = self.job_id
+        if self.notebook_params is not None:
+            self.json["notebook_params"] = self.notebook_params
+        if self.python_params is not None:
+            self.json["python_params"] = self.python_params
+        if self.spark_submit_params is not None:
+            self.json["spark_submit_params"] = self.spark_submit_params
+        if self.jar_params is not None:
+            self.json["jar_params"] = self.jar_params
 
         # Validate the dictionary to a valid JSON object
-        self.json = _deep_string_coerce(run_now_json)
+        self.json = _deep_string_coerce(self.json)
 
         # Submit the job
         self.run_id = hook.run_now(self.json)


Test patch:
diff --git a/docs/test_generate_docs.py b/docs/test_generate_docs.py
index 69a62b5222..75a13de5db 100644
--- a/docs/test_generate_docs.py
+++ b/docs/test_generate_docs.py
@@ -1,4 +1,6 @@
+import glob
 import inspect
+import os
 import re
 import sys
 import textwrap
@@ -23,6 +25,7 @@ try:
         get_call_signature,
         get_class_methods,
         patch_imports,
+        build_example,
         VALID_DOCSTRING_SECTIONS,
     )
 
@@ -36,6 +39,8 @@ except ImportError:
 
 pytest.mark.skipif(sys.version_info < (3, 6))
 
+ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+
 
 def consistency_check(obj, obj_name):
     patt = re.compile(r"(?<=>`)(.*?)(?=[\(|`:])")
@@ -481,3 +486,8 @@ def test_section_headers_are_properly_formatted(obj):
             assert (
                 False
             ), f"{obj.__module__}.{obj.__qualname__} has a poorly formatted '{section}' header"
+
+
+@pytest.mark.parametrize("path", glob.glob(os.path.join(ROOT, "examples", "*.py")))
+def test_example(path):
+    build_example(path)
diff --git a/tests/engine/test_flow_runner.py b/tests/engine/test_flow_runner.py
index 3f21a6b0ad..c07cc95fbf 100644
--- a/tests/engine/test_flow_runner.py
+++ b/tests/engine/test_flow_runner.py
@@ -1,6 +1,5 @@
 import collections
 import datetime
-import logging
 import queue
 import random
 import time
@@ -1324,67 +1323,6 @@ class TestContext:
         output = res.result[return_ctx_key].result
         assert isinstance(output, datetime.datetime)
 
-    @pytest.mark.parametrize("as_string", [True, False])
-    def test_context_derives_dates_from_custom_date(self, as_string):
-        @prefect.task
-        def return_ctx():
-            return prefect.context.copy()
-
-        custom_date = pendulum.now().add(months=1)
-
-        f = Flow(name="test", tasks=[return_ctx])
-        with prefect.context(
-            {"date": custom_date.to_datetime_string() if as_string else custom_date}
-        ):
-            res = f.run()
-
-        assert res.is_successful()
-
-        context = res.result[return_ctx].result
-
-        # Ensure the string is converted to a datetime object
-        assert isinstance(context["date"], pendulum.DateTime)
-
-        # Ensure dependent variables use the custom date
-        assert context["yesterday"] == custom_date.add(days=-1).strftime("%Y-%m-%d")
-        assert context["tomorrow"] == custom_date.add(days=1).strftime("%Y-%m-%d")
-        assert context["today_nodash"] == custom_date.strftime("%Y%m%d")
-        assert context["yesterday_nodash"] == custom_date.add(days=-1).strftime(
-            "%Y%m%d"
-        )
-        assert context["tomorrow_nodash"] == custom_date.add(days=1).strftime("%Y%m%d")
-
-    def test_context_warns_on_unparsable_custom_date(self, caplog, monkeypatch):
-
-        now = pendulum.now()
-        monkeypatch.setattr("pendulum.now", MagicMock(return_value=now))
-
-        caplog.set_level(logging.WARNING)
-
-        @prefect.task
-        def return_ctx_key():
-            return prefect.context.get("tomorrow")
-
-        f = Flow(name="test", tasks=[return_ctx_key])
-        with prefect.context({"date": "foobar"}):
-            res = f.run()
-
-        assert res.is_successful()
-
-        found_log = False
-        for record in caplog.records:
-            if (
-                record.levelno == logging.WARNING
-                and "could not be parsed into a pendulum `DateTime` object"
-                in record.message
-            ):
-                found_log = True
-        assert found_log
-
-        # Uses `now` to determine tomorrow
-        output = res.result[return_ctx_key].result
-        assert output == now.add(days=1).strftime("%Y-%m-%d")
-
     @pytest.mark.parametrize(
         "outer_context, inner_context, sol",
         [
diff --git a/tests/tasks/databricks/test_databricks.py b/tests/tasks/databricks/test_databricks.py
index d9525af003..05c539e990 100644
--- a/tests/tasks/databricks/test_databricks.py
+++ b/tests/tasks/databricks/test_databricks.py
@@ -2,44 +2,9 @@ import pytest
 
 from prefect.tasks.databricks import DatabricksSubmitRun
 from prefect.tasks.databricks import DatabricksRunNow
-from prefect.tasks.databricks.databricks_hook import DatabricksHook
 
 
-class DatabricksHookTestOverride(DatabricksHook):
-    """
-    Overrides `DatabricksHook` to avoid making actual API calls
-    and return mocked responses instead
-
-    Args:
-        - mocked_response (dict): JSON response of API call
-    """
-
-    def __init__(self, mocked_response={}, **kwargs) -> None:
-        self.mocked_response = mocked_response
-        super().__init__(self, **kwargs)
-
-    def _do_api_call(self, endpoint_info, json):
-        return self.mocked_response
-
-
-class DatabricksRunNowTestOverride(DatabricksRunNow):
-    """
-    Overrides `DatabricksRunNow` to allow mocked API responses
-    to be returned using `DatabricksHookTestOverride`
-
-    Args:
-        - mocked_response (dict): JSON response of API call
-    """
-
-    def __init__(self, mocked_response, **kwargs) -> None:
-        self.mocked_response = mocked_response
-        super().__init__(**kwargs)
-
-    def get_hook(self):
-        return DatabricksHookTestOverride(self.mocked_response)
-
-
-@pytest.fixture
+@pytest.fixture(scope="session")
 def job_config():
 
     config = {
@@ -58,7 +23,7 @@ def job_config():
     return config
 
 
-@pytest.fixture
+@pytest.fixture(scope="session")
 def notebook_job_config():
 
     config = {
@@ -72,52 +37,6 @@ def notebook_job_config():
     return config
 
 
-@pytest.fixture
-def notebook_job_config_full():
-
-    config = {
-        "job_id": 1,
-        "notebook_params": {
-            "dry-run": "true",
-            "oldest-time-to-consider": "1457570074236",
-        },
-        "python_params": ["python-param1", "python-param2"],
-        "spark_submit_params": ["spark-param1", "spark-param2"],
-        "jar_params": ["jar-param1", "jar-param2"],
-    }
-
-    return config
-
-
-@pytest.fixture
-def databricks_api_response_success():
-
-    response = {
-        "run_id": "test-run-id",
-        "run_page_url": "https://run_page_url",
-        "state": {
-            "life_cycle_state": "TERMINATED",
-            "result_state": "SUCCESS",
-            "state_message": "Completed",
-        },
-    }
-
-    return response
-
-
-@pytest.fixture
-def task_template(databricks_api_response_success, notebook_job_config_full):
-
-    return DatabricksRunNowTestOverride(
-        mocked_response=databricks_api_response_success,
-        databricks_conn_secret={
-            "host": "https://cloud.databricks.com",
-            "token": "databricks-token",
-        },
-        json=notebook_job_config_full,
-    )
-
-
 def test_raises_if_invalid_host_submitrun(job_config):
 
     # from prefect.tasks.secrets import PrefectSecret
@@ -142,64 +61,3 @@ def test_raises_if_invalid_host_runnow(notebook_job_config):
             databricks_conn_secret={"host": "", "token": ""}, json=notebook_job_config
         )
         task.run()
-
-
-class TestDatabricksRunNowAttributeOverrides:
-    """Test various expected attribute override behavior with `DatabricksRunNow.run`"""
-
-    def test_without_overrides(self, task_template):
-        task_template.run()
-
-        run_now_json = task_template.json
-        assert run_now_json.get("job_id") == "1"
-        assert run_now_json.get("notebook_params", {}).get("dry-run") == "true"
-        assert (
-            run_now_json.get("notebook_params", {}).get("oldest-time-to-consider")
-            == "1457570074236"
-        )
-        assert run_now_json.get("python_params") == ["python-param1", "python-param2"]
-        assert run_now_json.get("spark_submit_params") == [
-            "spark-param1",
-            "spark-param2",
-        ]
-        assert run_now_json.get("jar_params") == ["jar-param1", "jar-param2"]
-
-    def test_with_job_id_override(self, task_template):
-        task_template.run(job_id="42")
-        assert task_template.json.get("job_id") == "42"
-
-    def test_with_notebook_params_override(self, task_template):
-        task_template.run(notebook_params={"dry-run": "false"})
-
-        run_now_json = task_template.json
-        assert run_now_json.get("notebook_params", {}).get("dry-run") == "false"
-        assert (
-            run_now_json.get("notebook_params", {}).get("oldest-time-to-consider")
-            == "1457570074236"
-        )
-
-    def test_with_python_params_override(self, task_template):
-        task_template.run(python_params=["python-param3"])
-        assert task_template.json.get("python_params") == ["python-param3"]
-
-    def test_with_spark_submit_params_override(self, task_template):
-        task_template.run(spark_submit_params=["spark-param3"])
-        assert task_template.json.get("spark_submit_params") == ["spark-param3"]
-
-    def test_with_jar_params_override(self, task_template):
-        task_template.run(jar_params=["jar-param3"])
-        assert task_template.json.get("jar_params") == ["jar-param3"]
-
-    def test_with_json_override(self, task_template):
-        task_template.run(
-            json={"job_id": "123"},
-            notebook_params={"notebookparam1": "notebookvalue1"},
-        )
-
-        run_now_json = task_template.json
-        assert len(run_now_json) == 2
-        assert run_now_json.get("job_id") == "123"
-        assert (
-            run_now_json.get("notebook_params", {}).get("notebookparam1")
-            == "notebookvalue1"
-        )