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

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

@@ -26,9 +26,10 @@ jobs:
       matrix:
         command:
           [
-            "./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 :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:prefect-plugin:build -x :datahub-frontend:build -x :datahub-web-react:build --parallel",
             "./gradlew :datahub-frontend:build :datahub-web-react:build --parallel",
-            "./gradlew :metadata-ingestion-modules:airflow-plugin:build --parallel"
+            "./gradlew :metadata-ingestion-modules:airflow-plugin:build --parallel",
+            "./gradlew :metadata-ingestion-modules:prefect-plugin:build --parallel"
           ]
         timezone:
           [

.github/workflows/prefect-plugin.yml

@@ -0,0 +1,78 @@
+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:
+      - master
+    paths:
+      - ".github/**"
+      - "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:
+      - 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 and test  (extras ${{ matrix.extraPythonRequirement }})
+        run: ./gradlew -Pextra_pip_requirements='${{ matrix.extraPythonRequirement }}' :metadata-ingestion-modules:prefect-plugin:lint :metadata-ingestion-modules:airflow-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' && matrix.extraPythonRequirement == 'prefect>=2.0.0' }}
+        with:
+          name: Test Results (Prefect Plugin ${{ matrix.python-version}})
+          path: |
+            **/build/reports/tests/test/**
+            **/build/test-results/test/**
+            **/junit.*.xml
+      - 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-${{ matrix.python-version }}-${{ matrix.extraPythonRequirement }}
+          name: pytest-prefect
+          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

@@ -71,7 +71,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

@@ -573,6 +573,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

@@ -91,6 +91,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",
@@ -638,6 +643,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,49 @@
+# 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](https://prefecthq.github.io/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/2.10.13/cloud/cloud-quickstart/) to setup Prefect Cloud.
+3. Refer [Host Prefect server](https://docs.prefect.io/2.10.13/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
+
+For setup details please refer [prefect-datahub](https://prefecthq.github.io/prefect-datahub/).
+
+## 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.

metadata-ingestion-modules/prefect-plugin/.gitignore

@@ -0,0 +1,143 @@
+.envrc
+src/prefect_datahub/__init__.py.bak
+.vscode/
+output
+pvenv36/
+bq_credentials.json
+/tmp
+*.bak
+
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+pip-wheel-metadata/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+.python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# Generated classes
+src/datahub/metadata/
+wheels/
+junit.quick.xml