feat(integration/prefect): prefect block to emit prefect pipeline

.github/workflows/build-and-test.yml

@@ -76,7 +76,7 @@ jobs:
  - name: Gradle build (and test) for NOT metadata ingestion
  if: ${{ matrix.command == 'except_metadata_ingestion' && needs.setup.outputs.backend_change == 'true' }}
  run: |
- ./gradlew build -x :metadata-ingestion:build -x :metadata-ingestion:check -x docs-website:build -x :metadata-integration:java:spark-lineage:test -x :metadata-io:test -x :metadata-ingestion-modules:airflow-plugin:build -x :metadata-ingestion-modules:airflow-plugin:check -x :datahub-frontend:build -x :datahub-web-react:build --parallel
+ ./gradlew build -x :metadata-ingestion:build -x :metadata-ingestion:check -x docs-website:build -x :metadata-integration:java:spark-lineage:test -x :metadata-io:test -x :metadata-ingestion-modules:airflow-plugin:build -x :metadata-ingestion-modules:airflow-plugin:check -x :metadata-ingestion-modules:prefect-plugin:build -x :metadata-ingestion-modules:prefect-plugin:check -x :datahub-frontend:build -x :datahub-web-react:build --parallel
  - name: Gradle build (and test) for frontend
  if: ${{ matrix.command == 'frontend' && needs.setup.outputs.frontend_change == 'true' }}
  run: |

.github/workflows/prefect-plugin.yml

@@ -0,0 +1,85 @@
+name: Prefect Plugin
+on:
+ push:
+ branches:
+ - master
+ paths:
+ - ".github/workflows/prefect-plugin.yml"
+ - "metadata-ingestion-modules/prefect-plugin/**"
+ - "metadata-ingestion/**"
+ - "metadata-models/**"
+ pull_request:
+ branches:
+ - "**"
+ paths:
+ - ".github/workflows/prefect-plugin.yml"
+ - "metadata-ingestion-modules/prefect-plugin/**"
+ - "metadata-ingestion/**"
+ - "metadata-models/**"
+ release:
+ types: [published]
+
+concurrency:
+ group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+ cancel-in-progress: true
+
+jobs:
+ prefect-plugin:
+ runs-on: ubuntu-latest
+ env:
+ SPARK_VERSION: 3.0.3
+ DATAHUB_TELEMETRY_ENABLED: false
+ strategy:
+ matrix:
+ python-version: ["3.7", "3.10"]
+ include:
+ - python-version: "3.7"
+ - python-version: "3.10"
+ fail-fast: false
+ steps:
+ - name: Set up JDK 17
+ uses: actions/setup-java@v3
+ with:
+ distribution: "zulu"
+ java-version: 17
+ - uses: gradle/gradle-build-action@v2
+ - uses: actions/checkout@v3
+ - uses: actions/setup-python@v4
+ with:
+ python-version: ${{ matrix.python-version }}
+ cache: "pip"
+ - name: Install dependencies
+ run: ./metadata-ingestion/scripts/install_deps.sh
+ - name: Install prefect package
+ run: ./gradlew :metadata-ingestion-modules:prefect-plugin:lint :metadata-ingestion-modules:prefect-plugin:testQuick
+ - name: pip freeze show list installed
+ if: always()
+ run: source metadata-ingestion-modules/prefect-plugin/venv/bin/activate && pip freeze
+ - uses: actions/upload-artifact@v3
+ if: ${{ always() && matrix.python-version == '3.10'}}
+ with:
+ name: Test Results (Prefect Plugin ${{ matrix.python-version}})
+ path: |
+ **/build/reports/tests/test/**
+ **/build/test-results/test/**
+ **/junit.*.xml
+ !**/binary/**
+ - name: Upload coverage to Codecov
+ if: always()
+ uses: codecov/codecov-action@v3
+ with:
+ token: ${{ secrets.CODECOV_TOKEN }}
+ directory: .
+ fail_ci_if_error: false
+ flags: prefect,prefect-${{ matrix.extra_pip_extras }}
+ name: pytest-prefect-${{ matrix.python-version }}
+ verbose: true
+
+ event-file:
+ runs-on: ubuntu-latest
+ steps:
+ - name: Upload
+ uses: actions/upload-artifact@v3
+ with:
+ name: Event File
+ path: ${{ github.event_path }}

.github/workflows/test-results.yml

@@ -2,7 +2,7 @@ name: Test Results
 
 on:
  workflow_run:
- workflows: ["build & test", "metadata ingestion", "Airflow Plugin"]
+ workflows: ["build & test", "metadata ingestion", "Airflow Plugin", "Prefect Plugin"]
  types:
  - completed
 

docs-website/build.gradle

@@ -73,7 +73,8 @@ task yarnInstall(type: YarnTask) {
 task yarnGenerate(type: YarnTask, dependsOn: [yarnInstall,
  generateGraphQLSchema, generateJsonSchema,
  ':metadata-ingestion:modelDocGen', ':metadata-ingestion:docGen',
- ':metadata-ingestion:buildWheel', ':metadata-ingestion-modules:airflow-plugin:buildWheel'] ) {
+ ':metadata-ingestion:buildWheel', ':metadata-ingestion-modules:airflow-plugin:buildWheel',
+ ':metadata-ingestion-modules:prefect-plugin:buildWheel'] ) {
  inputs.files(projectMdFiles)
  outputs.cacheIf { true }
  args = ['run', 'generate']

docs-website/generateDocsDir.ts

@@ -572,6 +572,7 @@ function copy_python_wheels(): void {
  const wheel_dirs = [
  "../metadata-ingestion/dist",
  "../metadata-ingestion-modules/airflow-plugin/dist",
+ "../metadata-ingestion-modules/prefect-plugin/dist",
  ];
 
  const wheel_output_directory = path.join(STATIC_DIRECTORY, "wheels");

docs-website/sidebars.js

@@ -298,6 +298,11 @@ module.exports = {
  id: "docs/lineage/airflow",
  label: "Airflow",
  },
+ {
+ type: "doc",
+ id: "docs/lineage/prefect",
+ label: "Prefect",
+ },
  //"docker/airflow/local_airflow",
  "metadata-integration/java/spark-lineage/README",
  "metadata-ingestion/integration_docs/great-expectations",
@@ -750,6 +755,9 @@ module.exports = {
  //"docs/how/graph-onboarding",
  //"docs/demo/graph-onboarding",
  //"metadata-ingestion-modules/airflow-plugin/README"
+ //"metadata-ingestion-modules/prefect-plugin/README"
+ //"metadata-ingestion-modules/prefect-plugin/docs/concept_mapping"
+ //"metadata-ingestion-modules/prefect-plugin/docs/datahub_emitter"
  // "metadata-ingestion/schedule_docs/datahub", // we can delete this
  // TODO: change the titles of these, removing the "What is..." portion from the sidebar"
  // "docs/what/entity",

docs/lineage/prefect.md

@@ -0,0 +1,135 @@
+# Prefect Integration
+
+DataHub supports integration of
+
+- Prefect flow and task metadata
+- Flow run and Task run information as well as
+- Lineage information when present
+
+## What is Prefect Datahub Block?
+
+Blocks are primitive within Prefect that enable the storage of configuration and provide an interface for interacting with external systems. We integrated `prefect-datahub` block which use [Datahub Rest](../../metadata-ingestion/sink_docs/datahub.md#datahub-rest) emitter to emit metadata events while running prefect flow.
+
+## Prerequisites to use Prefect Datahub Block
+
+1. You need to use either Prefect Cloud (recommended) or the self hosted Prefect server.
+2. Refer [Cloud Quickstart](https://docs.prefect.io/latest/getting-started/quickstart/) to setup Prefect Cloud.
+3. Refer [Host Prefect server](https://docs.prefect.io/latest/guides/host/) to setup self hosted Prefect server.
+4. Make sure the Prefect api url is set correctly. You can check it by running below command:
+```shell
+prefect profile inspect
+```
+5. If you are using Prefect Cloud, the API URL should be set as `https://api.prefect.cloud/api/accounts/<account_id>/workspaces/<workspace_id>`.
+6. If you are using a self-hosted Prefect server, the API URL should be set as `http://<host>:<port>/api`.
+
+## Setup
+
+### Installation
+
+Install `prefect-datahub` with `pip`:
+
+```shell
+pip install 'prefect-datahub'
+```
+
+Requires an installation of Python 3.7+.
+
+### Saving configurations to a block
+
+This is a one-time activity, where you can save the configuration on the [Prefect block document store](https://docs.prefect.io/latest/concepts/blocks/#saving-blocks).
+While saving you can provide below configurations. Default value will get set if not provided while saving the configuration to block.
+
+Config | Type | Default | Description
+--- | --- | --- | ---
+datahub_rest_url | `str` | *http://localhost:8080* | DataHub GMS REST URL
+env | `str` | *PROD* | The environment that all assets produced by this orchestrator belong to. For more detail and possible values refer [here](https://datahubproject.io/docs/graphql/enums/#fabrictype).
+platform_instance | `str` | *None* | The instance of the platform that all assets produced by this recipe belong to. For more detail please refer [here](https://datahubproject.io/docs/platform-instances/).
+
+```python
+from prefect_datahub.datahub_emitter import DatahubEmitter
+DatahubEmitter(
+ datahub_rest_url="http://localhost:8080",
+ env="PROD",
+ platform_instance="local_prefect"
+).save("BLOCK-NAME-PLACEHOLDER")
+```
+
+Congrats! You can now load the saved block to use your configurations in your Flow code:
+
+```python
+from prefect_datahub.datahub_emitter import DatahubEmitter
+DatahubEmitter.load("BLOCK-NAME-PLACEHOLDER")
+```
+
+!!! info "Registering blocks"
+
+ Register blocks in this module to
+ [view and edit them](https://docs.prefect.io/ui/blocks/)
+ on Prefect Cloud:
+
+ ```bash
+ prefect block register -m prefect_datahub
+ ```
+
+### Load the saved block in prefect workflows
+
+After installing `prefect-datahub` and [saving the configution](#saving-configurations-to-a-block), you can easily use it within your prefect workflows to help you emit metadata event as show below!
+
+```python
+from prefect import flow, task
+from prefect_datahub.dataset import Dataset
+from prefect_datahub.datahub_emitter import DatahubEmitter
+
+datahub_emitter = DatahubEmitter.load("MY_BLOCK_NAME")
+
+@task(name="Transform", description="Transform the data")
+def transform(data):
+ data = data.split(" ")
+ datahub_emitter.add_task(
+ inputs=[Dataset("snowflake", "mydb.schema.tableA")],
+ outputs=[Dataset("snowflake", "mydb.schema.tableC")],
+ )
+ return data
+
+@flow(name="ETL flow", description="Extract transform load flow")
+def etl():
+ data = transform("This is data")
+ datahub_emitter.emit_flow()
+```
+
+**Note**: To emit the tasks, user compulsory need to emit flow. Otherwise nothing will get emit.
+
+## Concept mapping
+
+Prefect concepts are documented [here](https://docs.prefect.io/latest/concepts/), and datahub concepts are documented [here](https://datahubproject.io/docs/what-is-datahub/datahub-concepts).
+
+Prefect Concept | DataHub Concept
+--- | ---
+[Flow](https://docs.prefect.io/latest/concepts/flows/) | [DataFlow](https://datahubproject.io/docs/generated/metamodel/entities/dataflow/)
+[Flow Run](https://docs.prefect.io/latest/concepts/flows/#flow-runs) | [DataProcessInstance](https://datahubproject.io/docs/generated/metamodel/entities/dataprocessinstance)
+[Task](https://docs.prefect.io/latest/concepts/tasks/) | [DataJob](https://datahubproject.io/docs/generated/metamodel/entities/datajob/)
+[Task Run](https://docs.prefect.io/latest/concepts/tasks/#tasks) | [DataProcessInstance](https://datahubproject.io/docs/generated/metamodel/entities/dataprocessinstance)
+[Task Tag](https://docs.prefect.io/latest/concepts/tasks/#tags) | [Tag](https://datahubproject.io/docs/generated/metamodel/entities/tag/)
+
+
+## How to validate saved block and emit of metadata
+
+1. Go and check in Prefect UI at the Blocks menu if you can see the datahub emitter.
+2. Run a Prefect workflow. In the flow logs, you should see Datahub related log messages like:
+
+```
+Emitting flow to datahub...
+Emitting tasks to datahub...
+```
+## Debugging
+
+### Incorrect Prefect API URL
+
+If your Prefect API URL aren't being generated correctly or set incorrectly, then in that case you can set the Prefect API URL manually as show below:
+
+```shell
+prefect config set PREFECT_API_URL='http://127.0.0.1:4200/api'
+```
+
+### Connection error for Datahub Rest URL
+If you get ConnectionError: HTTPConnectionPool(host='localhost', port=8080), then in that case your GMS service is not up.