Build an Asset Graph from Your Python Pipeline (Prefect Assets Demo)

Hi, I'm Kevin, a technical product manager at Prefect. I'm so excited to introduce assets for Prefect Cloud, a new first class way to catalog and visualize the data outcomes of your workflows and the lineage that connects them together.

Assets offer a structured interface for describing the data your workflows produce and rely on. Prefect uses this information to build a graph, giving you visibility into your data life cycle across flows, teams, and tools.

In Prefect, building your asset graph is easy. All you need to do is drop a few decorators and asset keys into your existing Python workflows. Let's take a look at how it works.

Here I have a pure Python data pipeline that loads raw data from an S3 bucket into a Snowflake table, then combines it with the selection from another table for a final aggregate view. I want to turn this pipeline into a prefix flow.

First, I'll import the flow and materialize decorators from the prefect Python package. I'll place the flow decorator on my pipeline function denoting it as the entry point of my workflow. Then I'll add the materialized decorator to the function that loads data from S3 into Snowflake. If you're already familiar with prefix tasks, the materialized decorator is an extension of the task decorator, meaning materializations have all the same retry and caching capabilities as basic tasks.

In prefect, the minimum definition for an asset is a key in the form of a URI. This task is responsible for materializing my snowflake table. So, I'll provide the key of my snowflake asset to the decorator. Since the file in S3 is a pre-existing asset that needs to load successfully for my workflow to proceed, I'll also include it in the decorator as an asset dependency. When this function runs, the S3 file will appear upstream of the snowflake table on the graph.

Finally, I'll add another materialized decorator, this time on the function that builds the final snowflake table. Because there's an input to this function containing data that was returned from the prior materialization. Prefect automatically detects a data dependency. This dependency implicitly describes a relationship between my two materialized assets. So the second snowflake table will appear downstream of the first one on the graph.

Now I'll run my flow. Once it's finished, I'll be able to see the resulting graph on the assets page in the UI. Each asset on the graph is built from its most recent materialization attempt. Asset state is determined by the outcome of the function that materialized it. A completed function creates a successful materialization. A raised exception creates a failed materialization.

Let's expand our asset definitions by introducing a more formal interface. I'll import asset and asset properties and use them to attach a human readable name, a description, the assets owners, and a URL linking to the asset in its source system. Prefect will automatically recognize owners that are team names, user handles, or user email addresses and your prefix cloud account, but any strings work here as well. Any place in the Python SDK that accepts asset keys also accepts full asset instances. Let's replace them.

Now, once I rerun my flow, we'll be able to see our more complete asset definitions in the UI. This is really useful for statically defining things I know about my assets in advance, but the stakeholders of my data ecosystem need to know a little bit more about what happened at the exact moment an asset was materialized.

I can attach arbitrary metadata to an asset by calling add asset metadata inside any function that materializes or depends on assets. I'm going to record the row count from each of my snowflake tables to add even more visibility to my workflow outcome. Now I don't have to leave the prefect UI to see how many rows are in my snowflake table.

This example only scratches the surface. Using the same simple interface we've already seen, I can express a huge variety of scenarios, including cross flow asset dependencies, multi-asset materializations, and even materializing single assets from multiple tasks that can be cached or retried. These examples and more can be found in our documentation.

Thanks for joining me and we look forward to seeing all the assets you materialize with Prefect.