Use openlineage-airflow in airflow example and fix debugging step (#1635)

* Use openlineage-airflow in airflow example

Signed-off-by: wslulciuc <[email protected]>

* Fix alter column for step 5 in airflow example

Signed-off-by: wslulciuc <[email protected]>

Author: wslulciuc

.gitignore

@@ -20,11 +20,11 @@ bin/
 build/
 clients/java/out/
 dist/
-examples/airflow/marquez.env
-examples/airflow/requirements.txt
-examples/airflow/whl
 integrations/airflow/tests/integration/integration-requirements.txt
 integrations/airflow/tests/integration/requirements.txt
+examples/airflow/whl
+examples/airflow/requirements.txt
+examples/airflow/openlineage.env
 out/*
 venv
 
@@ -36,7 +36,3 @@ marquez.yml
 
 # Dependent Helm charts
 chart/charts/
-
-integrations/spark/src/test/resources/test_data/test_output
-integrations/spark/src/test/resources/test_data/rdd_to_csv_output
-integrations/spark/src/test/resources/test_data/rdd_to_table

examples/airflow/README.md

@@ -1,14 +1,14 @@
 # [Airflow](https://airflow.apache.org) Example
 
-In this example, we'll walk you through how to enable an **Airflow DAG** to send lineage metadata to **Marquez**. The example will help demonstrate some of the features of Marquez.
+In this example, we'll walk you through how to enable Airflow DAGs to send lineage metadata to Marquez using [OpenLineage](https://openlineage.io/). The example will help demonstrate some of the features of Marquez.
 
 ### What you’ll learn:
 
-* Enable Marquez in Airflow
-* Write your very first Marquez enabled DAG
+* Enable OpenLineage in Airflow
+* Write your very first OpenLineage enabled DAG
 * Troubleshoot a failing DAG using Marquez
 
-## Prerequisites
+# Prerequisites
 
 Before you begin, make sure you have installed:
 
@@ -17,29 +17,28 @@ Before you begin, make sure you have installed:
 
 > **Note:** We recommend that you have allocated at least **2 CPUs** and **8 GB** of memory to Docker.
 
-## Step 1: Prepare the Environment
+# Step 1: Setup
 
-First, if you haven't already, clone the Marquez repository and enter the `examples/airflow` directory.
+First, if you haven't already, clone the Marquez repository and change into the `examples/airflow` directory:
 
 ```bash
 $ git clone https://github.com/MarquezProject/marquez.git
 $ cd examples/airflow
 ```
 
-To make sure the latest [`marquez-airflow`](https://pypi.org/project/marquez-airflow) is downloaded when starting Airflow, you'll need to create a `requirements.txt` file with the following content:
+To make sure the latest [`openlineage-airflow`](https://pypi.org/project/openlineage-airflow) library is downloaded and installed when starting Airflow, you'll need to create a `requirements.txt` file with the following content:
 
 ```
-marquez-airflow
+openlineage-airflow
 ```
 
-Next, we'll need to specify where we want Airflow to send DAG metadata. To do so, create a config file named `marquez.env` with the following environment variables and values:
+Next, we'll need to specify where we want Airflow to send DAG metadata. To do so, create a config file named `openlineage.env` with the following environment variables and values:
 
 ```bash
-MARQUEZ_BACKEND=http             # Collect metadata using HTTP backend
-MARQUEZ_URL=http://marquez:5000  # The URL of the HTTP backend
-MARQUEZ_NAMESPACE=example        # The namespace associated with the collected metadata
+OPENLINEAGE_URL=http://marquez:5000  # The URL of the HTTP backend
+OPENLINEAGE_NAMESPACE=example        # The namespace associated with the DAG collected metadata
 ```
-> **Note:** The `marquez.env` config file will be used by the `airflow`, `airflow_scheduler`, and `airflow_worker` containers to send lineage metadata to Marquez.
+> **Note:** The `openlineage.env` config file will be used by the `airflow`, `airflow_scheduler`, and `airflow_worker` containers to send lineage metadata to Marquez.
 
 Your `examples/airflow/` directory should now contain the following:
 
@@ -49,36 +48,36 @@ Your `examples/airflow/` directory should now contain the following:
   ├── docker
   ├── docker-compose.yml
   ├── docs
-  ├── marquez.env
+  ├── openlineage.env
   └── requirements.txt
 
   ```
 
-## Step 2: Write Airflow DAGs using Marquez
+# Step 2: Write Airflow DAGs using OpenLineage
 
-In this step, we will create two new Airflow DAGs that perform simple tasks. The `counter` DAG will generate a random number every minute, while the `sum` DAG calculates a sum every five minutes. This will result in a simple pipeline containing two jobs and two datasets.
+In this step, we'll create two new Airflow DAGs that perform simple tasks. The `counter` DAG will generate a random number every minute, while the `sum` DAG calculates a sum every five minutes. This will result in a simple pipeline containing two jobs and two datasets.
 
 First, let's create the `dags/` folder where our example DAGs will be located:
 
 ```bash
 $ mkdir dags
 ```
 
-When writing our DAGs, we'll use [`marquez-airflow`](https://pypi.org/project/marquez-airflow), enabling Marquez to observe the DAG and automatically collect task-level metadata.  Notice that the only change required to begin collecting DAG metadata is to use `marquez-airflow` instead of `airflow`:
+When writing our DAGs, we'll use [`openlineage-airflow`](https://pypi.org/project/openlineage-airflow), enabling OpenLineage to observe the DAG and automatically collect task-level metadata.  Notice that the only change required to begin collecting DAG metadata is to use `openlineage.airflow` instead of `airflow`:
 
 ```diff
 - from airflow import DAG
-+ from marquez_airflow import DAG
++ from openlineage.airflow import DAG
 ```
 
-## Step 2.1: Create DAG `counter`
+# Step 2.1: Create DAG `counter`
 
 Under `dags/`, create a file named `counter.py` and add the following code:
 
 ```python
 import random
 
-from marquez_airflow import DAG
+from openlineage.airflow import DAG
 from airflow.operators.postgres_operator import PostgresOperator
 from airflow.utils.dates import days_ago
 
@@ -124,15 +123,14 @@ t2 = PostgresOperator(
 )
 
 t1 >> t2
-
 ```
 
-## Step 2.2: Create DAG `sum`
+# Step 2.2: Create DAG `sum`
 
 Under `dags/`, create a file named `sum.py` and add the following code:
 
 ```python
-from marquez_airflow import DAG
+from openlineage.airflow import DAG
 from airflow.operators.postgres_operator import PostgresOperator
 from airflow.utils.dates import days_ago
 
@@ -175,7 +173,6 @@ t2 = PostgresOperator(
 )
 
 t1 >> t2
-
 ```
 
 At this point, you should have the following under your `examples/airflow/` directory:
@@ -189,13 +186,13 @@ At this point, you should have the following under your `examples/airflow/` dire
 ├── docker/
 ├── docker-compose.yml
 ├── docs/
-├── marquez.env
+├── openlineage.env
 └── requirements.txt
 ```
 
-## Step 3: Start Airflow with Marquez
+# Step 3: Start Airflow with Marquez
 
-Now that we have our DAGs defined and Marquez is enabled in Airflow, we can run the example! To start Airflow, run:
+Now that we have our DAGs defined and OpenLineage is enabled in Airflow, we can run the example! To start Airflow, run:
 
 ```bash
 $ docker-compose up
@@ -205,20 +202,19 @@ $ docker-compose up
 
 **The above command will:**
 
-* Start Airflow and install `marquez-airflow`
+* Start Airflow and install `openlineage-airflow`
 * Start Marquez
 * Start Postgres
 
-To view the Airflow UI and verify it's running, open http://localhost:8080. Then, login using the username and password: `airflow` / `airflow`. You can also browse to http://localhost:3000 to view the Marquez UI.
-
+To view the Airflow UI and verify it's running, open [http://localhost:8080](http://localhost:8080). Then, login using the username and password: `airflow` / `airflow`. You can also browse to [http://localhost:3000](http://localhost:3000) to view the Marquez UI.
 
-## Step 4: View Collected Metadata
+# Step 4: View Collected Metadata
 
-To ensure that Airflow is executing `counter` and `sum`, navigate to the DAGs tab in Airflow and verify that they are both enabled and have a timestamp in the Last Run column.
+To ensure that Airflow is executing `counter` and `sum`, navigate to the DAGs tab in Airflow and verify that they are both enabled and are in a _running_ state:
 
 ![](./docs/airflow-view-dag.png)
 
-To view DAG metadata collected by Marquez from Airflow, browse to the Marquez UI by visiting http://localhost:3000. Then, use the _search_ bar in the upper right-side of the page and search for the `counter.inc` job. To view lineage metadata for `counter.inc`, click on the job from the drop-down list:
+To view DAG metadata collected by Marquez from Airflow, browse to the Marquez UI by visiting [http://localhost:3000](http://localhost:3000). Then, use the _search_ bar in the upper right-side of the page and search for the `counter.inc` job. To view lineage metadata for `counter.inc`, click on the job from the drop-down list:
 
 > **Note:** If the `counter.inc` job is not in the drop-down list, check to see if Airflow has successfully executed the DAG.
 
@@ -228,7 +224,7 @@ If you take a quick look at the lineage graph for `counter.inc`, you should see
 
 ![](./docs/lineage-view-job.png)
 
-## Step 5: Troubleshoot a Failing DAG with Marquez
+# Step 5: Troubleshoot a Failing DAG with Marquez
 
 In this step, let's quickly walk through a simple troubleshooting scenario where DAG `sum` begins to fail as the result of an upstream schema change for table `counts`. So, let's get to it!
 
@@ -241,13 +237,36 @@ t1 = PostgresOperator(
 -   task_id='if_not_exists',
 +   task_id='alter_name_of_column',
     postgres_conn_id='example_db',
--   sql='''
+    sql='''
 -   CREATE TABLE IF NOT EXISTS counts (
 -     value INTEGER
 -   );''',
-+   sql='''
-+   ALTER TABLE counts RENAME COLUMN value TO value_1_to_10;
-+   ''',
++    DO $$
++    BEGIN
++      IF EXISTS(SELECT *
++        FROM information_schema.columns
++        WHERE table_name='counts' and column_name='value')
++      THEN
++          ALTER TABLE "counts" RENAME COLUMN "value" TO "value_1_to_10";
++      END IF;
++    END $$;
+    ''',
+    dag=dag
+)
+```
+
+```diff
+t2 = PostgresOperator(
+    task_id='inc',
+    postgres_conn_id='example_db',
+    sql='''
+-    INSERT INTO counts (value)
++    INSERT INTO counts (value_1_to_10)
+         VALUES (%(value)s)
+    ''',
+    parameters={
+      'value': random.randint(1, 10)
+    },
     dag=dag
 )
 ```
@@ -281,11 +300,11 @@ With the code change, the DAG `sum` begins to run successfully:
 
 _Congrats_! You successfully step through a troubleshooting scenario of a failing DAG using metadata collected with Marquez! You can now add your own DAGs to `dags/` to build more complex data lineage graphs.
 
-## Next Steps
+# Next Steps
 
-* Review the Marquez [HTTP API](https://marquezproject.github.io/marquez/openapi.html) used to collect Airflow DAG metadata and learn how to build your own integrations
-* Take a look at our [`marquez-spark`](https://github.com/MarquezProject/marquez/tree/main/integrations/spark) integration that can be used with Airflow
+* Review the Marquez [HTTP API](https://marquezproject.github.io/marquez/openapi.html) used to collect Airflow DAG metadata and learn how to build your own integrations using OpenLineage
+* Take a look at [`openlineage-spark`](https://openlineage.io/integration/apache-spark) integration that can be used with Airflow
 
-## Feedback
+# Feedback
 
-What did you think of this example? You can reach out to us on [slack](http://bit.ly/MarquezSlack) and leave us feedback, or [open a pull request](https://github.com/MarquezProject/marquez/blob/main/CONTRIBUTING.md#submitting-a-pull-request) with your suggestions!  
+What did you think of this example? You can reach out to us on [slack](http://bit.ly/MarquezSlack) and leave us feedback, or [open a pull request](https://github.com/MarquezProject/marquez/blob/main/CONTRIBUTING.md#submitting-a-pull-request) with your suggestions!

examples/airflow/docker-compose.yml

@@ -5,7 +5,7 @@ services:
     ports:
       - "8080:8080"
     env_file:
-      - marquez.env
+      - openlineage.env
     environment:
       - AIRFLOW_USERNAME=airflow
       - AIRFLOW_PASSWORD=airflow
@@ -28,7 +28,7 @@ services:
   airflow_scheduler:
     image: bitnami/airflow-scheduler:1.10.13
     env_file:
-      - marquez.env
+      - openlineage.env
     environment:
       - AIRFLOW_FERNET_KEY=Z2uDm0ZL60fXNkEXG8LW99Ki2zf8wkmIltaTz1iQPDU=
       - AIRFLOW_DATABASE_HOST=postgres
@@ -48,7 +48,7 @@ services:
   airflow_worker:
     image: bitnami/airflow-worker:1.10.13
     env_file:
-      - marquez.env
+      - openlineage.env
     environment:
       - AIRFLOW_FERNET_KEY=Z2uDm0ZL60fXNkEXG8LW99Ki2zf8wkmIltaTz1iQPDU=
       - AIRFLOW_DATABASE_HOST=postgres