## 1. Project Overview & Quickstart (amundsen-io/amundsen)
## File: README.md
Amundsen is a *data discovery and metadata engine* for improving the productivity of data analysts, data scientists and engineers when interacting with data. It does that today by indexing data resources (tables, dashboards, streams, etc.) and powering a page-rank style search based on usage patterns (e.g. highly queried tables show up earlier than less queried tables). Think of it as **Google search for data**. The project is named after Norwegian explorer [Roald Amundsen](https://en.wikipedia.org/wiki/Roald_Amundsen), the first person to discover the South Pole.
Amundsen is hosted by the [LF AI & Data Foundation](https://lfaidata.foundation/). It includes three microservices, one data ingestion library and one common library.
- [amundsenfrontendlibrary](frontend): Frontend service which is a Flask application with a React frontend.
- [amundsensearchlibrary](search): Search service, which leverages Elasticsearch for search capabilities, is used to power frontend metadata searching.
- [amundsenmetadatalibrary](metadata): Metadata service, which leverages Neo4j or Apache Atlas as the persistent layer, to provide various metadata.
- [amundsendatabuilder](databuilder): Data ingestion library for building metadata graph and search index.
Users could either load the data with [a python script](https://github.com/amundsen-io/amundsen/blob/main/databuilder/example/scripts/sample_data_loader.py) with the library
or with an [Airflow DAG](https://github.com/amundsen-io/amundsen/tree/main/databuilder/example/dags) importing the library.
- [amundsencommon](common): Amundsen Common library holds common codes among microservices in Amundsen.
- [amundsengremlin](https://github.com/amundsen-io/amundsengremlin): Amundsen Gremlin library holds code used for converting model objects into vertices and edges in gremlin. It's used for loading data into an AWS Neptune backend.
- [amundsenrds](https://github.com/amundsen-io/amundsenrds): Amundsenrds contains ORM models to support relational database as metadata backend store in Amundsen. The schema in ORM models follows the logic of databuilder models. Amundsenrds will be used in databuilder and metadatalibrary for metadata storage and retrieval with relational databases.
## Documentation
* [Homepage](https://www.amundsen.io/)
* [Documentation](https://www.amundsen.io/amundsen/)
## Community Roadmap
We want your input about what is important, for that, add your votes using the đź‘Ť reaction:
* [Top Feature Requests](https://github.com/amundsen-io/amundsen/issues?q=is%3Aissue+is%3Aclosed+sort%3Areactions-%2B1-desc+label%3Atype%3Afeature+label%3Astatus%3Aneeds_votes+)
* [Documentation Requests](https://github.com/amundsen-io/amundsen/issues?q=is%3Aissue+is%3Aclosed+sort%3Areactions-%2B1-desc+label%3Atype%3Adocumentation+label%3Astatus%3Aneeds_votes+)
* [Top Bugs](https://github.com/amundsen-io/amundsen/issues?q=is%3Aissue+is%3Aclosed+sort%3Areactions-%2B1-desc+label%3Atype%3Abug+label%3Astatus%3Aneeds_votes+)
* [Top Questions](https://github.com/amundsen-io/amundsen/issues?q=is%3Aissue+is%3Aclosed+sort%3Areactions-%2B1-desc+label%3Atype%3Aquestion+label%3Astatus%3Aneeds_votes)
## Requirements
- Python >= 3.8
- Node v12
## User Interface
Please note that the mock images only served as demonstration purpose.
- **Landing Page**: The landing page for Amundsen including 1. search bars; 2. popular used tables;
- **Search Preview**: See inline search results as you type
- **Table Detail Page**: Visualization of a Hive / Redshift table
- **Column detail**: Visualization of columns of a Hive / Redshift table which includes an optional stats display
- **Data Preview Page**: Visualization of table data preview which could integrate with [Apache Superset](https://github.com/apache/incubator-superset) or other Data Visualization Tools.
## Getting Started and Installation
Please visit the Amundsen installation documentation for a [quick start](https://www.amundsen.io/amundsen/installation/) to bootstrap a default version of Amundsen with dummy data.
## Supported Entities
- Tables (from Databases)
- Dashboards
- ML Features
- People (from HR systems)
## Supported Integrations
### Table Connectors
- [Amazon Athena](https://aws.amazon.com/athena/)
- [Amazon EventBridge](https://aws.amazon.com/eventbridge/)
- [Amazon Glue](https://aws.amazon.com/glue/) and anything built over it
- [Amazon Redshift](https://aws.amazon.com/redshift/)
- [Apache Cassandra](https://cassandra.apache.org/)
- [Apache Druid](https://druid.apache.org/)
- [Apache Hive](https://hive.apache.org/)
- CSV
- [dbt](https://www.getdbt.com/)
- [Delta Lake](https://delta.io/)
- [Elasticsearch](https://www.elastic.co/)
- [Google BigQuery](https://cloud.google.com/bigquery)
- [IBM DB2](https://www.ibm.com/analytics/db2)
- [Kafka Schema Registry](https://docs.confluent.io/platform/current/schema-registry/index.html)
- [Microsoft SQL Server](https://www.microsoft.com/en-us/sql-server/default.aspx)
- [MySQL](https://www.mysql.com/)
- [Oracle](https://www.oracle.com/index.html) (through dbapi or sql_alchemy)
- [PostgreSQL](https://www.postgresql.org/)
- [PrestoDB](http://prestodb.io/)
- [Trino (formerly Presto SQL)](https://trino.io/)
- [Vertica](https://www.vertica.com/)
- [Snowflake](https://www.snowflake.com/)
Amundsen can also connect to any database that provides `dbapi` or `sql_alchemy` interface (which most DBs provide).
### Table Column Statistics
- [Pandas Profiling](https://pandas-profiling.github.io/pandas-profiling/docs/master/rtd/)
### Dashboard Connectors
- [Apache Superset](https://superset.apache.org/)
- [Mode Analytics](https://mode.com/)
- [Redash](https://redash.io/)
- [Tableau](https://tableau.com/)
- [Databricks SQL](https://databricks.com/product/databricks-sql)
### ETL Orchestration
- [Apache Airflow](https://airflow.apache.org/)
## Get Involved in the Community
Want help or want to help? Use the button in our [header](https://github.com/amundsen-io/amundsen#readme) to join our slack channel.
Contributions are also more than welcome! As explained in [CONTRIBUTING.md](https://github.com/amundsen-io/amundsen/blob/main/CONTRIBUTING.md) there are many ways to contribute, it does not all have to be code with new features and bug fixes, also documentation, like FAQ entries, bug reports, blog posts sharing experiences etc. all help move Amundsen forward. If you find a security vulnerability, [please follow this guide](https://github.com/amundsen-io/amundsen/blob/main/SECURITY.md).
## Architecture Overview
Please visit [Architecture](https://www.amundsen.io/amundsen/architecture/) for Amundsen architecture overview.
## Resources
### Blog Posts and Interviews
- [Amundsen - Lyft's data discovery & metadata engine](https://eng.lyft.com/amundsen-lyfts-data-discovery-metadata-engine-62d27254fbb9) (April 2019)
- [Software Engineering Daily podcast on Amundsen](https://softwareengineeringdaily.com/2019/04/16/lyft-data-discovery-with-tao-feng-and-mark-grover/) (April 2019)
- [How Lyft Drives Data Discovery](https://youtu.be/WVjss62XIG0) (July 2019)
- [Data Engineering podcast on Solving Data Discovery At Lyft](https://www.dataengineeringpodcast.com/amundsen-data-discovery-episode-92/) (Aug 2019)
- [Open Sourcing Amundsen: A Data Discovery And Metadata Platform](https://eng.lyft.com/open-sourcing-amundsen-a-data-discovery-and-metadata-platform-2282bb436234) (Oct 2019)
- [Adding Data Quality into Amundsen with Programmatic Descriptions](https://technology.edmunds.com/2020/05/27/Adding-Data-Quality-into-Amundsen-with-Programmatic-Descriptions/) by [Sam Shuster](https://github.com/samshuster) from [Edmunds.com](https://www.edmunds.com/) (May 2020)
- [Facilitating Data discovery with Apache Atlas and Amundsen](https://medium.com/wbaa/facilitating-data-discovery-with-apache-atlas-and-amundsen-631baa287c8b) by [Mariusz GĂłrski](https://github.com/mgorsk1) from [ING](https://www.ing.com/Home.htm) (June 2020)
- [Using Amundsen to Support User Privacy via Metadata Collection at Square](https://developer.squareup.com/blog/using-amundsen-to-support-user-privacy-via-metadata-collection-at-square/) by [Alyssa Ransbury](https://github.com/alran) from [Square](https://squareup.com/) (July 14, 2020)
- [Amundsen Joins LF AI as New Incubation Project](https://lfai.foundation/blog/2020/08/11/amundsen-joins-lf-ai-as-new-incubation-project/) (Aug 11, 2020)
- [Amundsen: one year later](https://eng.lyft.com/amundsen-1-year-later-7b60bf28602) (Oct 6, 2020)
### Talks
- Disrupting Data Discovery {[slides](https://www.slideshare.net/taofung/strata-sf-amundsen-presentation), [recording](https://www.youtube.com/watch?v=m1B-ptm0Rrw)} (Strata SF, March 2019)
- Amundsen: A Data Discovery Platform from Lyft {[slides](https://www.slideshare.net/taofung/data-council-sf-amundsen-presentation)} (Data Council SF, April 2019)
- Disrupting Data Discovery {[slides](https://www.slideshare.net/markgrover/disrupting-data-discovery)} (Strata London, May 2019)
- ING Data Analytics Platform (Amundsen is mentioned) {[slides](https://static.sched.com/hosted_files/kccnceu19/65/ING%20Data%20Analytics%20Platform.pdf), [recording](https://www.youtube.com/watch?v=8cE9ppbnDPs&t=465) } (Kubecon Barcelona, May 2019)
- Disrupting Data Discovery {[slides](https://www.slideshare.net/PhilippeMizrahi/meetup-sf-amundsen), [recording](https://www.youtube.com/watch?v=NgeCOVjSJ7A)} (Making Big Data Easy SF, May 2019)
- Disrupting Data Discovery {[slides](https://www.slideshare.net/TamikaTannis/neo4j-graphtour-santa-monica-2019-amundsen-presentation-173073727), [recording](https://www.youtube.com/watch?v=Gr3-RfWn49A)} (Neo4j Graph Tour Santa Monica, September 2019)
- Disrupting Data Discovery {[slides](https://www.slideshare.net/secret/56EPbcvswqyH90)} (IDEAS SoCal AI & Data Science Conference, Oct 2019)
- Data Discovery with Amundsen by [Gerard Toonstra](https://twitter.com/radialmind) from Coolblue {[slides](https://docs.google.com/presentation/d/1rkrP8ZobkLPZbwisrLWTdPN5I52SgVGM1eqAFDJXj2A/edit?usp=sharing)} and {[talk](https://www.youtube.com/watch?v=T54EO1MuE7I&list=PLqYhGsQ9iSEq7fDvXcd67iVzx5nsf9xnK&index=17)} (BigData Vilnius 2019)
- Towards Enterprise Grade Data Discovery and Data Lineage with Apache Atlas and Amundsen by [Verdan Mahmood](https://github.com/verdan) and Marek Wiewiorka from ING {[slides](https://docs.google.com/presentation/d/1FixTTNd1dt_f3PAKhL1KLOeOLsIQq0iFvQA6qlpjIg0/edit#slide=id.p1), [talk](https://bigdatatechwarsaw.eu/agenda/)} (Big Data Technology Warsaw Summit 2020)
- Airflow @ Lyft (which covers how we integrate Airflow and Amundsen) by [Tao Feng](https://github.com/feng-tao) {[slides](https://www.slideshare.net/taofung/airflow-at-lyft-airflow-summit2020) and [website](https://airflowsummit.org/sessions/how-airbnb-twitter-lyft-use-airflow/)} (Airflow Summit 2020)
- Data DAGs with lineage for fun and for profit by [Bolke de Bruin](https://github.com/bolkedebruin) {[website](https://airflowsummit.org/sessions/data-dags-with-lineage/)} (Airflow Summit 2020)
- Solving Data Discovery Challenges at Lyft with Amundsen, an Open-source Metadata Platform by [Tao Feng](https://github.com/feng-tao) ([Data+AI summit Europe 2020](https://databricks.com/session_eu20/solving-data-discovery-challenges-at-lyft-with-amundsen-an-open-source-metadata-platform))
- Data Discovery at Databricks with Amundsen by [Tao Feng](https://github.com/feng-tao) and [Tianru Zhou](https://www.linkedin.com/in/tianru-zhou-134868132/) ([Data+AI summit NA 2021](https://databricks.com/session_na21/data-discovery-at-databricks-with-amundsen))
### Related Articles
- [How LinkedIn, Uber, Lyft, Airbnb and Netflix are Solving Data Management and Discovery for Machine Learning Solutions](https://towardsdatascience.com/how-linkedin-uber-lyft-airbnb-and-netflix-are-solving-data-management-and-discovery-for-machine-9b79ee9184bb)
- [Data Discovery in 2020](https://medium.com/@torokyle/data-discovery-in-2020-3c907383caa0)
- [4 Data Trends to Watch in 2020](https://medium.com/memory-leak/4-data-trends-to-watch-in-2020-491707902c09)
- [Work-Bench Snapshot: The Evolution of Data Discovery & Catalog](https://medium.com/work-bench/work-bench-snapshot-the-evolution-of-data-discovery-catalog-2f6c0425616b)
- [Future of Data Engineering](https://www.infoq.com/presentations/data-engineering-pipelines-warehouses/)
- [Governance and Discovery](https://www.oreilly.com/radar/governance-and-discovery/)
- [A Data Engineer’s Perspective On Data Democratization](https://towardsdatascience.com/a-data-engineers-perspective-on-data-democratization-a8aed10f4253?source=friends_link&sk=63638570d03e4145265932c12af33f9d)
- [Graph Technology Landscape 2020](https://graphaware.com/graphaware/2020/02/17/graph-technology-landscape-2020.html)
- [In-house Data Discovery platforms](https://datastrategy.substack.com/p/in-house-data-discovery-platforms)
- [Linux Foundation AI Foundation Landscape](https://landscape.lfai.foundation/)
- [Lyft’s Amundsen: Data-Discovery with Built-In Trust](https://thenewstack.io/lyfts-amundsen-data-discovery-with-built-in-trust/)
- [How to find and organize your data from the command-line](https://towardsdatascience.com/how-to-find-and-organize-your-data-from-the-command-line-852a4042b2be)
- [Cataloging Tools for Data Teams](https://towardsdatascience.com/cataloging-tools-for-data-teams-8d62d7a4cd95)
- [An Overview of Data Discovery Platforms and Open Source Solutions](https://eugeneyan.com/writing/data-discovery-platforms/)
- [Hacking Data Discovery in AWS with Amundsen at SEEK](https://medium.com/seek-blog/hacking-data-discovery-with-amundsen-c55d262014f4)
- [A step-by-step guide deploying Amundsen on Google Cloud Platform](https://medium.com/talabat-tech/a-step-by-step-guide-deploying-amundsen-on-google-cloud-platform-ee4ef20d81f5)
- [Machine Learning Features discovery with Feast and Amundsen](https://getindata.com/blog/machine-learning-features-discovery-feast-amundsen)
- [Data discovery at REA group](https://www.rea-group.com/blog/data-discovery/)
- [Integrating Slack with Amundsen for Ease of Data Discovery](https://medium.com/convoy-tech/integrating-slack-with-amundsen-for-ease-of-data-discovery-ef3b54834da5)
- [Building a data discovery solution with Amundsen and Amazon Neptune](https://aws.amazon.com/blogs/database/building-a-data-discovery-solution-with-amundsen-and-amazon-neptune/)
- [Amundsen — Installing in an Istio-enabled environment](https://medium.com/@owenleung_89035/amundsen-installing-in-an-istio-enabled-environment-d0b9bdaeac49)
- [Amundsen — Integrate with Okta SingleSignOn](https://medium.com/@owenleung_89035/amundsen-integrate-with-okta-singlesignon-20ad0a22d3d9)
- [Data Engineering Skills](https://www.scaler.com/blog/data-engineer-skills/)
### Community meetings
Community meetings are held on the first Thursday of every month at 9 AM Pacific, Noon Eastern, 6 PM Central European Time. [Link to join](https://zoom.us/j/92594007352?pwd=Rmo4SGp4WnY3eGlzclgrWlp6bTNTQT09)
### Upcoming meetings & notes
You can the exact date for the next meeting and the agenda a few weeks before the meeting in [this doc](https://docs.google.com/document/d/1bsJWNt1GBFmV-aRbHFuYgMFnMgIIvAmbhsvDatb0Vis).
Notes from all past meetings are available [here](https://docs.google.com/document/d/1bsJWNt1GBFmV-aRbHFuYgMFnMgIIvAmbhsvDatb0Vis).
## Who uses Amundsen?
Here is the list of organizations that are **officially** using Amundsen today. If your organization uses Amundsen, please file a PR and update this list.
| [Asana](https://asana.com/)
[Bang & Olufsen](https://www.bang-olufsen.com/en)
[Brex](https://www.brex.com/)
[Cameo](https://www.cameo.com)
[Chan Zuckerberg Initiative](https://chanzuckerberg.com/)
[Cimpress Technology](https://cimpress.com)
[Coles Group](https://www.colesgroup.com.au/home/)
[Convoy](https://www.convoy.com)
[Data Sprints](https://datasprints.com/)
[Dcard](https://www.dcard.tw/)
[Delivery Hero](https://www.deliveryhero.com/) | [Devoted Health](https://www.devoted.com/)
[DHI Group](https://dhigroupinc.com/)
[Edmunds](https://www.edmunds.com/)
[Everfi](https://everfi.com/)
[Gusto](https://gusto.com/)
[Hurb](https://hurb.com)
[ING](https://www.ing.com/Home.htm)
[Instacart](https://www.instacart.com/)
[iRobot](https://www.irobot.com)
[Lett](https://lett.digital/)
[LMC](https://www.lmc.eu/cs/)
[Loft](https://loft.com.br)
[Lyft](https://www.lyft.com/) | [Merlin](https://merlinjobs.com)
[PicPay](https://picpay.com.br)
[Plarium Krasnodar](https://company.plarium.com/en/studio/russia-krasnodar/)
[PUBG](https://careers.pubg.com/)
[Rapido](https://rapido.bike/Careers?lan=en )
[REA Group](https://www.rea-group.com/)
[Remitly](https://www.remitly.com/)
[Snap](https://www.snap.com/en-US)
[Square](https://squareup.com/us/en)
[Tile](https://www.thetileapp.com)
[WePay](https://go.wepay.com/)
[WeTransfer](https://wetransfer.com)
[Workday](https://www.workday.com/en-us/homepage.html) |
| --- | --- | --- |
## Contributors ✨
Thanks goes to these incredible people:
---
## File: amundsen-kube-helm/templates/restore-backup/README.md
# Restoring neo4j Backups
The Amundsen Helm chart includes a Kubernetes CronJob that backs up the neo4j database to S3. If you need to restore from one of these backups, use the one-off pod in this directory.
## Create the Pod
You should have setup `kubectl` for the Kubernetes cluster you wish to restore in before running these commands.
Update the YAML file with the S3 Bucket for the backup you wish to restore and then apply the pod.
```shell
kubectl apply -n -f restore-neo4j-pod.yaml
```
Once the pod has been created, it will automatically run the restore. You can check the pod's logs to see whether it has succeeded for failed.
---
## File: amundsen-kube-helm/README.md
# Amundsen K8s Helm Charts
Source code can be found [here](https://github.com/amundsen-io/amundsen)
## What is this?
This is setup templates for deploying [amundsen](https://github.com/amundsen-io/amundsen) on [k8s (kubernetes)](https://kubernetes.io/), using [helm.](https://helm.sh/)
## How do I get started?
1. Make sure you have the following command line clients setup:
- k8s (kubectl)
- helm
2. Build out a cloud based k8s cluster, such as [Amazon EKS](https://aws.amazon.com/eks/)
3. Ensure you can connect to your cluster with cli tools in step 1.
## Prerequisites
1. Helm 2.14+
2. Kubernetes 1.14+
## Chart Requirements
Note we updated from elasticsearch 6 to elasticsearch 7
| Repository | Name | Version |
|------------|------|---------|
| https://helm.elastic.co | elasticsearch | 7.13.4 |
## Values
| Key | Type | Default | Description |
|-----|------|---------|-------------|
| LONG_RANDOM_STRING | int | `1234` | A long random string. You should probably provide your own. This is needed for OIDC. |
| affinity | object | `{}` | amundsen application wide configuration of affinity. This applies to search, metadata, frontend and neo4j. Elasticsearch has it's own configuation properties for this. [ref](https://kubernetes.io/docs/concepts/configuration/assign-pod-node/#affinity-and-anti-affinity) |
| clusterDomain | string | `"cluster.local"` | |
| dnsZone | string | `"teamname.company.com"` | **DEPRECATED - its not standard to pre construct urls this way.** The dns zone (e.g. group-qa.myaccount.company.com) the app is running in. Used to construct dns hostnames (on aws only). |
| dockerhubImagePath | string | `"amundsendev"` | **DEPRECATED - this is not useful, it would be better to just allow the whole image to be swapped instead.** The image path for dockerhub. |
| elasticsearch.enabled | bool | `true` | set this to false, if you want to provide your own ES instance. |
| elasticsearch.esJavaOpts | string | `"-Xmx8g -Xms8g"` | set init memory size (Xms) and maximum memory size (Xmx) for the es jvm. |
| elasticsearch.fullnameOverride | string | `"amundsen-elasticsearch-master"` | this is the service name of the amundsen elasticsearch master. Change it if you want to give a new name for the elasticsearch service |
| elasticsearch.image | string | `"elasticsearch"` | elasticsearch docker image name |
| elasticsearch.resources | object | `{"limits":{"memory":"15Gi"},"requests":{"memory":"10Gi"}}` | set the pod resources |
| elasticsearch.sysctlInitContainer | object | `{"enabled":false}` | If this set to true, the es pod will require some admin privilege, which is not allowed in most case. So set it to false |
| environment | string | `"dev"` | **DEPRECATED - its not standard to pre construct urls this way.** The environment the app is running in. Used to construct dns hostnames (on aws only) and ports. |
| flaskApp.class | string | `""` | The class name within the flaskApp.module |
| flaskApp.module | string | `""` | Any custom flask module you may need to implement as a wrapper |
| frontEnd.ALL_UNEDITABLE_SCHEMAS | string | `nil` | Environment variable for allowing/disallowing editing schemas via the UI. All schemas are allowed to be edited by default. Set to 'true' to disallow. See https://www.amundsen.io/amundsen/frontend/docs/flask_config/#uneditable-table-descriptions for more |
| frontEnd.affinity | object | `{}` | Frontend pod specific affinity. |
| frontEnd.annotations | object | `{}` | Frontend service specific tolerations. |
| frontEnd.baseUrl | string | `"http://localhost"` | used by notifications util to provide links to amundsen pages in emails. |
| frontEnd.config.class | string | `nil` | Optional Config class. |
| frontEnd.envVars | object | `{}` | |
| frontEnd.image | string | `"amundsendev/amundsen-frontend"` | The image of the frontend container. |
| frontEnd.imagePullSecrets | list | `[]` | Optional pod imagePullSecrets [ref](https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/) |
| frontEnd.imageTag | string | `"latest"` | The image tag of the frontend container. |
| frontEnd.nodeSelector | object | `{}` | Frontend pod specific nodeSelector. |
| frontEnd.podAnnotations | object | `{}` | Frontend pod specific annotations. |
| frontEnd.replicas | int | `1` | How many replicas of the frontend service to run. |
| frontEnd.resources | object | `{}` | See pod resourcing [ref](https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/) |
| frontEnd.serviceName | string | `"frontend"` | The frontend service name. |
| frontEnd.servicePort | int | `80` | The port the frontend service will be exposed on via the loadbalancer. |
| frontEnd.serviceType | string | `"ClusterIP"` | The frontend service type. See service types [ref](https://kubernetes.io/docs/concepts/services-networking/service/#publishing-services-service-types) |
| frontEnd.tolerations | list | `[]` | Frontend pod specific tolerations. |
| ingress.annotations | object | `{}` | |
| ingress.enabled | bool | `true` | set this to true, if you want a ingress that expose HTTP and HTTPS routes from outside the cluster to your amundsen services. Don't use this if you are in a public cloud such as AWS, GCP |
| ingress.hosts[0].host | string | `"amundsen-test.your-domain.com"` | |
| ingress.hosts[0].paths[0] | string | `"/"` | |
| ingress.tls[0].hosts[0] | string | `"amundsen-test.your-domain.com"` | |
| metadata.affinity | object | `{}` | Metadata pod specific affinity. |
| metadata.annotations | object | `{}` | Metadata service specific tolerations. |
| metadata.envVars | object | `{}` | |
| metadata.image | string | `"amundsendev/amundsen-metadata"` | The image of the metadata container. |
| metadata.imagePullSecrets | list | `[]` | Optional pod imagePullSecrets [ref](https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/) |
| metadata.imageTag | string | `"latest"` | The image tag of the metadata container. |
| metadata.nodeSelector | object | `{}` | Metadata pod specific nodeSelector. |
| metadata.podAnnotations | object | `{}` | Metadata pod specific annotations. |
| metadata.proxy.host | string | `nil` | host name / URI of your proxy |
| metadata.proxy.password | string | `nil` | Credentials - Password of the proxy |
| metadata.proxy.port | string | `nil` | Port on which the proxy is running |
| metadata.proxy.user | string | `nil` | Credentials - Username of the proxy |
| metadata.replicas | int | `1` | How many replicas of the metadata service to run. |
| metadata.resources | object | `{}` | See pod resourcing [ref](https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/) |
| metadata.serviceName | string | `"metadata"` | The metadata service name. |
| metadata.serviceType | string | `"ClusterIP"` | The metadata service type. See service types [ref](https://kubernetes.io/docs/concepts/services-networking/service/#publishing-services-service-types) |
| metadata.tolerations | list | `[]` | Metadata pod specific tolerations. |
| neo4j.affinity | object | `{}` | neo4j specific affinity. |
| neo4j.annotations | object | `{}` | neo4j service specific tolerations. |
| neo4j.backup | object | `{"enabled":false,"podAnnotations":{},"s3Path":"s3://dev/null","schedule":"0 * * * *"}` | If enabled is set to true, make sure and set the s3 path as well. |
| neo4j.backup.s3Path | string | `"s3://dev/null"` | The s3path to write to for backups. |
| neo4j.backup.schedule | string | `"0 * * * *"` | The schedule to run backups on. Defaults to hourly. |
| neo4j.config | object | `{"dbms":{"heap_initial_size":"1G","heap_max_size":"2G","pagecache_size":"2G"}}` | Neo4j application specific configuration. This type of configuration is why the charts/stable version is not used. See [ref](https://github.com/helm/charts/issues/21439) |
| neo4j.config.dbms | object | `{"heap_initial_size":"1G","heap_max_size":"2G","pagecache_size":"2G"}` | dbms config for neo4j |
| neo4j.config.dbms.heap_initial_size | string | `"1G"` | the initial java heap for neo4j |
| neo4j.config.dbms.heap_max_size | string | `"2G"` | the max java heap for neo4j |
| neo4j.config.dbms.pagecache_size | string | `"2G"` | the page cache size for neo4j |
| neo4j.enabled | bool | `true` | If neo4j is enabled as part of this chart, or not. Set this to false if you want to provide your own version. |
| neo4j.image | string | `"neo4j"` | The image of the neo4j container. |
| neo4j.imageTag | string | `"3.3.0"` | The image tag of the neo4j container. |
| neo4j.initPluginsContainer.image | string | `"appropriate/curl"` | The image of the init neo4j plugins container. |
| neo4j.initPluginsContainer.imageTag | string | `"latest"` | The image tag of the init neo4j plugins container. |
| neo4j.initPluginsContainer.command | list | _See values.yaml_ | The command to execute in the init neo4j plugins container. |
| neo4j.nodeSelector | object | `{}` | neo4j specific nodeSelector. |
| neo4j.persistence | object | `{}` | Neo4j persistence. Turn this on to keep your data between pod crashes, etc. This is also needed for backups. |
| neo4j.podAnnotations | object | `{}` | neo4j pod specific annotations. |
| neo4j.resources | object | `{}` | See pod resourcing [ref](https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/) |
| neo4j.serviceType | string | `"ClusterIP"` | The neo4j service type. See service types [ref](https://kubernetes.io/docs/concepts/services-networking/service/#publishing-services-service-types) |
| neo4j.tolerations | list | `[]` | neo4j specific tolerations. |
| neo4j.version | string | `"3.3.0"` | **DEPRECATED - Now using the neo4j.imageTag** The neo4j application version used by amundsen. |
| nodeSelector | object | `{}` | amundsen application wide configuration of nodeSelector. This applies to search, metadata, frontend and neo4j. Elasticsearch has it's own configuation properties for this. [ref](https://kubernetes.io/docs/concepts/configuration/assign-pod-node/#nodeselector) |
| oidc.configs.FLASK_OIDC_CONFIG_URL | string | `"https://accounts.google.com/.well-known/openid-configuration"` | |
| oidc.configs.FLASK_OIDC_PROVIDER_NAME | string | `"google"` | |
| oidc.configs.FLASK_OIDC_REDIRECT_URI | string | `"/auth"` | |
| oidc.configs.FLASK_OIDC_SCOPES | string | `"openid email profile"` | |
| oidc.configs.FLASK_OIDC_USER_ID_FIELD | string | `"email"` | |
| oidc.configs.FLASK_OIDC_WHITELISTED_ENDPOINTS | string | `"status,healthcheck,health"` | |
| oidc.enabled | bool | `false` | flag to enable/disable the OIDC. Once enabled, - everything under oidc.configs will be parsed - flaskApp.module will be set as 'flaskoidc' - flaskApp.class will be set as 'FlaskOIDC' |
| oidc.frontend.client_id | string | `""` | |
| oidc.frontend.client_secret | string | `""` | |
| oidc.metadata.client_id | string | `""` | |
| oidc.metadata.client_secret | string | `""` | |
| oidc.search.client_id | string | `""` | |
| oidc.search.client_secret | string | `""` | |
| podAnnotations | object | `{}` | amundsen application wide configuration of podAnnotations. This applies to search, metadata, frontend and neo4j. Elasticsearch has it's own configuation properties for this. [ref](https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/) |
| provider | string | `"aws"` | The cloud provider the app is running in. Used to construct dns hostnames (on aws only). |
| search.affinity | object | `{}` | Search pod specific affinity. |
| search.annotations | object | `{}` | Search service specific tolerations. |
| search.envVars | object | `{}` | |
| search.image | string | `"amundsendev/amundsen-search"` | The image of the search container. |
| search.imagePullSecrets | list | `[]` | Optional pod imagePullSecrets [ref](https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/) |
| search.imageTag | string | `"latest"` | The image tag of the search container. |
| search.nodeSelector | object | `{}` | Search pod specific nodeSelector. |
| search.podAnnotations | object | `{}` | Search pod specific annotations. |
| search.proxy.endpoint | string | `nil` | Endpoint of the search proxy (i.e., ES endpoint etc.) You should only need to change this, if you don't use the version in this chart. elasticsearch-master.user-pengfei.svc.cluster.local |
| search.proxy.password | string | `nil` | |
| search.proxy.user | string | `nil` | |
| search.replicas | int | `1` | How many replicas of the search service to run. |
| search.resources | object | `{}` | See pod resourcing [ref](https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/) |
| search.serviceName | string | `"search"` | The search service name. |
| search.serviceType | string | `"ClusterIP"` | The search service type. See service types [ref](https://kubernetes.io/docs/concepts/services-networking/service/#publishing-services-service-types) |
| search.tolerations | list | `[]` | Search pod specific tolerations. |
| tolerations | list | `[]` | amundsen application wide configuration of tolerations. This applies to search, metadata, frontend and neo4j. Elasticsearch has it's own configuation properties for this. [ref](https://kubernetes.io/docs/concepts/configuration/assign-pod-node/#taints-and-tolerations-beta-feature) |
----------------------------------------------
Autogenerated from chart metadata using [helm-docs v1.5.0](https://github.com/norwoodj/helm-docs/releases/v1.5.0)
## Ingress support
If you want to deploy Amundsen on a K8s cluster on premise. You can activate the ingress module. Do not active ingress if you are using a public cloud such as AWS, GCP, etc.
``` yaml
ingress:
enabled: true
annotations: {}
hosts:
- host: amundsen-test.your-domain.com
paths: [/]
tls:
- hosts:
- amundsen-test.your-domain.com
```
## Neo4j DBMS Config?
You may want to override the default memory usage for Neo4J. In particular, if you're just test-driving a deployment and your node exits with status 137, you should set the usage to smaller values:
``` yaml
config:
dbms:
heap_initial_size: 1G
heap_max_size: 2G
pagecache_size: 2G
```
With this values file, you can then install Amundsen using Helm 2 with:
``` shell
helm install ./templates/helm --values impl/helm/dev/values.yaml
```
For Helm 3 it's now mandatory to specify a [chart reference name](https://helm.sh/docs/intro/using_helm/#helm-install-installing-a-package) e.g. `my-amundsen`:
``` shell
helm install my-amundsen ./templates/helm --values impl/helm/dev/values.yaml
```
## Other Notes
- For aws setup, you will also need to setup the [external-dns plugin](https://github.com/kubernetes-incubator/external-dns)
- There is an existing helm chart for neo4j, but, it is missing some features necessary to for use such as:
- [\[stable/neo4j\] make neo4j service definition more extensible](https://github.com/helm/charts/issues/21441); without this, it is not possible to setup external load balancers, external-dns, etc
- [\[stable/neo4j\] allow custom configuration of neo4j](https://github.com/helm/charts/issues/21439); without this, custom configuration is not possible which includes setting configmap based settings, which also includes turning on apoc.
---
## File: common/README.md
# Amundsen Common
[](https://badge.fury.io/py/amundsen-common)
[](LICENSE)
[](https://img.shields.io/badge/PRs-welcome-brightgreen.svg)
[](https://amundsenworkspace.slack.com/join/shared_invite/enQtNTk2ODQ1NDU1NDI0LTc3MzQyZmM0ZGFjNzg5MzY1MzJlZTg4YjQ4YTU0ZmMxYWU2MmVlMzhhY2MzMTc1MDg0MzRjNTA4MzRkMGE0Nzk)
Amundsen Common library holds common codes among micro services in Amundsen.
For information about Amundsen and our other services, visit the [main repository](https://github.com/amundsen-io/amundsen). Please also see our instructions for a [quick start](https://github.com/amundsen-io/amundsen/blob/master/docs/installation.md#bootstrap-a-default-version-of-amundsen-using-docker) setup of Amundsen with dummy data, and an [overview of the architecture](https://github.com/amundsen-io/amundsen/blob/master/docs/architecture.md#architecture).
## Requirements
- Python >= 3.8
## Doc
- https://www.amundsen.io/amundsen/
---
## File: databuilder/docs/dashboard_ingestion_guide.md
# Dashboard Ingestion guidance
(Currently this guidance is about using Databuilder to ingest Dashboard metadata into Neo4j and Elasticsearch)
Dashboard ingestion consists of multiple Databuilder jobs and it can be described in four steps:
1. Ingest base data to Neo4j.
2. Ingest additional data and decorate Neo4j over base data.
3. Update Elasticsearch index using Neo4j data
4. Remove stale data
Note that Databuilder jobs need to be sequenced as 1 -> 2 -> 3 -> 4. To sequencing these jobs, Lyft uses Airflow to orchestrate the job, but Databuilder is not limited to Airflow and you can also simply use Python script to sequence it -- not recommended for production though.
Also, step 1, 3, 4 is expected to have one Databuilder job where Step 2 is expected to have **multiple** Databuilder jobs and number of Databuilder jobs in step 2 is expected to grow as we add more metadata into Dashboard. To improve performance, it is recommended, but not required, to execute Databuilder jobs in step 2 concurrently.
Once finished step 1 and 2, you will have Graph like this:
Here this documentation will be using [Mode Dashboard](https://app.mode.com/) as concrete example to show how to ingest Dashboard metadata. However, this ingestion process not limited to Mode Dashboard and any other Dashboard can follow this flow.
### 1. Ingest base data to Neo4j.
Using [ModeDashboardExtractor](../README.md#modedashboardextractor) along with [FsNeo4jCSVLoader](../README.md#fsneo4jcsvloader) and [Neo4jCsvPublisher](../README.md#neo4jcsvpublisher) to add base information such as Dashboard group name, Dashboard group id, Dashboard group description, Dashboard name, Dashboard id, Dashboard description to Neo4j. Use [this job configuration](../README.md#modedashboardextractor) example to configure the job.
### 2. Ingest additional data and decorate Neo4j over base data.
Use other Mode dashboard's extractors in create & launch multiple Databuilder jobs. Note that it all Databuilder job here will use [FsNeo4jCSVLoader](../README.md#fsneo4jcsvloader) and [Neo4jCsvPublisher](../README.md#neo4jcsvpublisher) where their configuration should be almost the same except the `NODE_FILES_DIR` and `RELATION_FILES_DIR` that is being used for temporary location to hold data.
List of other Extractors can be found [here](../README.md#mode-dashboard-extractor)
#### 2.1. Ingest Dashboard usage data and decorate Neo4j over base data.
Mode provide usage data (view count) per Dashboard, but this is accumulated usage data. The main use case of usage is search ranking and `accumulated usage` is not that much useful for Amundsen as we don't want to show certain Dashboard that was popular years ago and potentially deprecated.
To bring recent usage information, we can `snapshot` accumulated usage per report daily and extract recent usage information (past 30 days, 60 days, 90 days that fits our view of recency).
##### 2.1.1. Ingest `accumulated usage` into Data warehouse (e.g: Hive, BigQuery, Redshift, Postgres, etc)
In this step, you can use ModeDashboardUsageExtractor to extract `accumulated_view_count` and load into Data warehouse of your choice by using GenericLoader.
Note that GenericLoader just takes a callback function, and you need to provide a function that `INSERT` record into your Dataware house.
```python
extractor = ModeDashboardUsageExtractor()
loader = GenericLoader()
task = DefaultTask(extractor=extractor,
loader=loader)
job_config = ConfigFactory.from_dict({
'{}.{}'.format(extractor.get_scope(), ORGANIZATION): organization,
'{}.{}'.format(extractor.get_scope(), MODE_ACCESS_TOKEN): mode_token,
'{}.{}'.format(extractor.get_scope(), MODE_PASSWORD_TOKEN): mode_password,
'{}.{}'.format(loader.get_scope(), 'callback_function'): mode_dashboard_usage_loader_callback_function,
})
job = DefaultJob(conf=job_config, task=task)
job.launch()
```
Step 2. Extract past ? days usage data from your Data warehouse and publish it to Neo4j.
You could use [existing extractors](../README.md#list-of-extractors) to achieve this with [DashboardUsage model](./models.md#dashboardusage) along with [FsNeo4jCSVLoader](../README.md#fsneo4jcsvloader) and [Neo4jCsvPublisher](../README.md#neo4jcsvpublisher).
### 3. Update Elasticsearch index using Neo4j data
Once data is ready in Neo4j, extract Neo4j data and push it to Elasticsearch using [Neo4jSearchDataExtractor](../databuilder/extractor/neo4j_search_data_extractor.py) and [ElasticsearchPublisher](../databuilder/publisher/elasticsearch_publisher.py)
```
/* Detailed source-code truncated for AI context efficiency. */
```
*Note that `DASHBOARD_ELASTICSEARCH_INDEX_MAPPING` is defined [here](../../common/amundsen_common/models/index_map.py).
### 4. Remove stale data
Dashboard ingestion, like Table ingestion, is UPSERT (CREATE OR UPDATE) operation and there could be some data deleted on source. Not removing it in Neo4j basically leaving a stale data in Amundsen.
You can use [Neo4jStalenessRemovalTask](../README.md#removing-stale-data-in-neo4j----neo4jstalenessremovaltask) to remove stale data.
There are two strategies to remove stale data. One is to use `job_publish_tag` and the other one is to use `milliseconds_to_expire`.
For example, you could use `job_publish_tag` to remove stale `DashboardGroup`, `Dashboard`, and `Query` nodes. And you could use `milliseconds_to_expire` on `Timestamp` node, `READ` relation, and `READ_BY`. One of the main reasons to use `milliseconds_to_expire` is to avoid race condition and it is explained more [here](./README.md#using-publisher_last_updated_epoch_ms-to-remove-stale-data)
---
## File: databuilder/docs/models.md
# Amundsen Models
## Overview
These are the python classes that live in [databuilder/models/](../databuilder/models/).
Models represent the data structures that live in either neo4j (if the model extends Neo4jSerializable) or in elasticsearch.
Models that extend Neo4jSerializable have methods to create:
- the nodes
- the relationships
In this way, amundsendatabuilder pipelines can create python objects that can then be loaded into neo4j / elastic search
without developers needing to know the internals of the neo4j schema.
-----
## The Models
### [TableMetadata](../databuilder/models/table_metadata.py)
*What datasets does my org have?*
#### Description
This corresponds to a dataset in amundsen and is the core building block.
In addition to ColumnMetadata, tableMetadata is one of the first datasets you should extract as
almost everything else depends on these being populated.
#### Extraction
In general, for Table and Column Metadata, you should be able to use one of the pre-made extractors
in the [extractor package](../databuilder/extractor)
### [Watermark](../databuilder/models/watermark.py)
*What is the earliest data that this table has? What is the latest data?*
This is NOT the same as when the data was last updated.
#### Description
Corresponds to the earliest and latest date that a dataset has. Only makes
sense if the dataset is timeseries data.
For example, a given table may have data from 2019/01/01 -> 2020/01/01
In that case the low watermark is 2019/01/01 and the high watermark is 2020/01/01.
#### Extraction
Depending on the datastore of your dataset, you would extract this by:
- a query on the minimum and maximum partition (hive)
- a query for the minimum and maximum record of a given timestamp column
### [TableColumnUsage](../databuilder/models/table_column_usage.py)
#### Description
*How many queries is a given column getting? By which users?*
Has query counts per a given table per a user. This can help identify
who uses given datasets so people can contact them if they have questions
on how to use a given dataset or if a dataset is changing. It is also used as a
search boost so that the most used tables are put to the top of the search results.
This model also populates the Popular Resources section on the homepage.
A table must have at least 10 unique users to appear on the homepage.
This configuration is defined [here](https://github.com/amundsen-io/amundsen/blob/main/metadata/metadata_service/config.py#L81)
and can be changed if your usage is less.
#### Extraction
For more traditional databases, there should be system tables where you can obtain
these sorts of usage statistics.
In other cases, you may need to use audit logs which could require a custom solution.
Finally, for non-traditional data lakes, getting this information exactly maybe difficult and you may need to rely
on a heuristic.
### [User](../databuilder/models/user.py)
*What users are there out there? Which team is this user on?*
#### Description
Represents all of the metadata for a user at your company.
This is required if you are going to be having authentication turned on.
#### Extraction
TODO
### [TableColumnStats](../databuilder/models/table_stats.py)
* What are the min/max values for this column? How many nulls are in this column? *
#### Description
This represents statistics on the column level (this is not for table level metrics).
The idea is that different companies will want to track different things about different columns, so this is highly
customizable.
It also will probably require a distributed cluster in order to calculate these regularly and in general is
probably the least accessible metrics to get at without a custom solution.
#### Extraction
The idea here would be to implement something that does the following:
For each table you care about:
For each column you care about:
Calculate statistics that you care about such as min/max/average etc.
### [Application](../databuilder/models/application.py)
* What job/application is writing to this table? *
#### Description
This is used to provide users a way to find out what job/application is responsible for writing to this dataset.
#### Extraction
TODO
### [Table Owner](../databuilder/models/table_owner.py)
* What team or user owns this dataset? *
#### Description
A dataset can have one or more owners. These owners are used when requesting table descriptions or could be just a useful
point of contact for a user inquiring about how to use a dataset.
#### Extraction
Although the main point of entry for owners is through the WebUI, you could in theory
extract this information based on who created a given table.
### [Table Source](../databuilder/models/table_source.py)
* Where is the source code for the application that writes to this dataset? *
#### Description
Generally there is going to be code that your company owns that describes how a dataset is created.
This model is what represents the link and type of repository to this source code so it is available to users.
#### Extraction
You will need a github/gitlab/your repository crawler in order to populate this automatically.
The idea there would be to search for a given table name or something else that is a unique identifier such that you can be confident
that the source correctly matches to this table.
### [TableLastUpdated](../databuilder/models/table_last_updated.py)
* When was the last time this data was updated? Is this table stale or deprecated? *
#### Description
This value is used to describe the last time the table had datapoints inserted into it.
It is a very useful value as it can help users identify if there are tables that are no longer being updated.
#### Extraction
There are some extractors available for this like [hive_table_last_updated_extractor](../databuilder/extractor/hive_table_last_updated_extractor.py)
that you can refer to. But you will need access to history that provides information on when the last data write happened on a given table.
If this data isn't available for your data source, you maybe able to approximate it by looking at the max of some timestamp column.
## Dashboard models
Dashboard models are normalized which means that the model is separated so that it can be easily decoupled with how data is extracted. (If model is denormalized, all metadata is in model, then one extraction needs to able to pull all the data which makes extraction hard and complex) There's trade off in this decision of normalized design where it can be inefficient in the case that some ingestion can be done in one job for metadata source happen to provide all data it need. However, to make model flexible for most of metadata, it is normalized.
### [DashboardMetadata](../databuilder/models/dashboard/dashboard_metadata.py)
#### Description
A baseline of Dashboard metadata that consists of dashboard group name, dashboard group description, dashboard description, etc. This model needs to be ingested first as other model builds relation to this.
#### Extraction
[ModeDashboardExtractor](../databuilder/extractor/dashboard/mode_analytics/mode_dashboard_extractor.py)
#### [DashboardOwner](../databuilder/models/dashboard/dashboard_owner.py)
#### Description
A model that encapsulate Dashboard's owner. Note that it does not create new user as it has insufficient information about user but it builds relation between User and Dashboard
#### Extraction
[ModeDashboardOwnerExtractor](../databuilder/extractor/dashboard/mode_analytics/mode_dashboard_owner_extractor.py)
#### [DashboardTable](../databuilder/models/dashboard/dashboard_table.py)
A model that link Dashboard with the tables used in various charts of the dashboard. Note that it does not create new dashboard, table as it has insufficient information but it builds relation between Tables and Dashboard.
Supporting extractor: Currently there's no open sourced extractor for this. In Lyft, there's audit table that records SQL query, where it came from with identifier, along with tables that is used in SQL query. We basically query this table via [DBAPIExtractor](../databuilder/extractor/db_api_extractor.py)
#### [DashboardUsage](../databuilder/models/dashboard/dashboard_usage.py)
#### Description
A model that encapsulate Dashboard usage between Dashboard and User
#### Extraction
You can use [ModeDashboardUsageExtractor](../databuilder/extractor/dashboard/mode_analytics/mode_dashboard_usage_extractor.py) . However, currently Mode only provides accumulated view count where we need recent view counts (past 30, 60, or 90 days). To get recent view count, in Lyft, we use [ModeDashboardUsageExtractor](../databuilder/extractor/dashboard/mode_analytics/mode_dashboard_usage_extractor.py) to extract accumulated view count and [GenericLoader](https://github.com/amundsen-io/amundsendatabuilder/blob/master/databuilder/loader/generic_loader.py) to load its record (no publisher here and publisher is not mandatory in DefaultJob) as a event where event materialized as daily snapshot. Once it captures daily accumulated view count, ingest recent view count by querying the datastore. In Lyft, we query via [DBAPIExtractor](../databuilder/extractor/db_api_extractor.py) through Presto.
#### [DashboardLastModifiedTimestamp](../databuilder/models/dashboard/dashboard_last_modified.py)
#### Description
A model that encapsulate Dashboard's last modified timestamp in epoch
#### Extraction
[ModeDashboardLastModifiedTimestampExtractor](../databuilder/extractor/dashboard/mode_analytics/mode_dashboard_last_modified_timestamp_extractor.py)
#### [DashboardExecution](../databuilder/models/dashboard/dashboard_execution.py)
A model that encapsulate Dashboard's execution timestamp in epoch and execution state. Note that this model supports last_execution and last_successful_execution by using [different identifier](../databuilder/models/dashboard/dashboard_execution.py#L23) in the URI.
#### Extraction
[ModeDashboardExecutionsExtractor](../databuilder/extractor/dashboard/mode_analytics/mode_dashboard_executions_extractor.py) which extracts last_execution.
[ModeDashboardLastSuccessfulExecutionExtractor](../databuilder/extractor/dashboard/mode_analytics/mode_dashboard_last_successful_executions_extractor.py)
#### [DashboardQuery](../databuilder/models/dashboard/dashboard_query.py)
#### Description
A model that encapsulate Dashboard's query information.
Supporting extractor: [ModeDashboardQueriesExtractor](../databuilder/extractor/dashboard/mode_analytics/mode_dashboard_queries_extractor.py)
#### [DashboardChart](../databuilder/models/dashboard/dashboard_chart.py)
#### Description
A model that encapsulate Dashboard's charts where chart is associated with query.
#### Extraction
[ModeDashboardChartsExtractor](../databuilder/extractor/dashboard/mode_analytics/mode_dashboard_charts_extractor.py)
## Feature models
Feature models include [FeatureMetadata](../databuilder/models/feature/feature_metadata.py), which encapsulates the basic feature details,
and supplemental models [Feature_Generation_Code](../databuilder/models/feature/feature_generation_code.py) and
[Feature_Watermark](../databuilder/models/feature/feature_watermark.py) for adding extra metadata.
In addition, the Tag, Badge, Owner, and Programmatic_Description models work with features.
### [FeatureMetadata](../databuilder/models/feature/feature_metadata.py)
#### Description
A baseline of Feature metadata. This model needs to be ingested first as other models build relations to it.
#### Extraction
No specific extractors are provided at this time. We expect users will either write custom extractors,
or use generic extractors (e.g. SQLAlchemyExtractor).
### [Feature_Generation_Code](../databuilder/models/feature/feature_generation_code.py)
#### Description
Allows ingesting the text of the generation code (SQL or otherwise) which was used to create a feature.
#### Extraction
No specific extractors are provided at this time. We expect users will either write custom extractors,
or use generic extractors (e.g. SQLAlchemyExtractor).
### [Feature_Watermark](../databuilder/models/feature/feature_watermark.py)
#### Description
Allows ingesting the high and low data range of a feature. Unlike [Watermark](../databuilder/models/watermark.py),
which is specific to tables (requires a partition, for example), Feature_Watermark is more general and does not
care about how the feature is stored.
#### Extraction
No specific extractors are provided at this time. We expect users will either write custom extractors,
or use generic extractors (e.g. SQLAlchemyExtractor).
---
## File: databuilder/docs/query_metadata_guide.md
# Query Metadata Guide
This document provides guidance on how to ingest query metadata and query composition metadata into Amundsen.
Query metadata and query composition metadata consists of four parts:
1. `QueryMetadata`: This represents a query
2. `QueryExecutionsMetadata`: This is an aggregation, representing the number of times a given query was executed wtihin an hour, day, week, etc.
3. `QueryJoinMetadata`: Represents a join between two columns
4. `QueryWhereMetadata`: Represents a whereclause used in a query, this may be associated to one or more columns and tables
The `QueryExecutionsMetadata`, `QueryJoinMetadata`, `QueryWhereMetadata` and `QueryMetadata` can be seen here:
Amundsen uses `QueryExecutionsMetadata` to determine time-sensitive relevance. As new `QueryExecutionsMetadata` are added and old ones are removed, Amundsen is able to continue to keep the most recent queries and the related joins and wheres relevant.
Since `QueryMetadata` can be related to `QueryExecutionsMetadata`, `QueryJoinMetadata` and `QueryWhereMetadata`, a single extractor is used to ingest these objects into Amundsen.
---
## File: databuilder/README.md
# Amundsen Databuilder
[](https://badge.fury.io/py/amundsen-databuilder)
[](LICENSE)
[](https://pypi.org/project/amundsen-databuilder/)
[](https://img.shields.io/badge/PRs-welcome-brightgreen.svg)
[](https://amundsenworkspace.slack.com/join/shared_invite/enQtNTk2ODQ1NDU1NDI0LTc3MzQyZmM0ZGFjNzg5MzY1MzJlZTg4YjQ4YTU0ZmMxYWU2MmVlMzhhY2MzMTc1MDg0MzRjNTA4MzRkMGE0Nzk)
Amundsen Databuilder is a data ingestion library, which is inspired by [Apache Gobblin](https://gobblin.apache.org/). It could be used in an orchestration framework(e.g. Apache Airflow) to build data from Amundsen. You could use the library either with an adhoc python script([example](https://github.com/amundsen-io/amundsen/blob/main/databuilder/example/scripts/sample_data_loader.py)) or inside an Apache Airflow DAG([example](https://github.com/amundsen-io/amundsen/blob/main/databuilder/example/dags/hive_sample_dag.py)).
For information about Amundsen and our other services, visit the [main repository](https://github.com/amundsen-io/amundsen#amundsen) `README.md` . Please also see our instructions for a [quick start](https://github.com/amundsen-io/amundsen/blob/master/docs/installation.md#bootstrap-a-default-version-of-amundsen-using-docker) setup of Amundsen with dummy data, and an [overview of the architecture](https://github.com/amundsen-io/amundsen/blob/master/docs/architecture.md#architecture).
## Requirements
- Python >= 3.8.x
- elasticsearch 7.x
## Doc
- https://www.amundsen.io/amundsen/
## Concept
ETL job consists of extraction of records from the source, transform records, if necessary, and load records into the sink. Amundsen Databuilder is a ETL framework for Amundsen and there are corresponding components for ETL called Extractor, Transformer, and Loader that deals with record level operation. A component called task controls all these three components.
Job is the highest level component in Databuilder that controls task and publisher and is the one that client use to launch the ETL job.
In Databuilder, each components are highly modularized and each components are using namespace based config, HOCON config, which makes it highly reusable and pluggable. (e.g: transformer can be reused within extractor, or extractor can be reused within extractor)
(Note that concept on components are highly motivated by [Apache Gobblin](https://gobblin.apache.org/ "Apache Gobblin"))
### [Extractor](https://github.com/amundsen-io/amundsen/tree/main/databuilder/databuilder/extractor "Extractor")
An extractor extracts records from the source. This does not necessarily mean that it only supports [pull pattern](https://blogs.sap.com/2013/12/09/to-push-or-pull-that-is-the-question/ "pull pattern") in ETL. For example, extracting records from messaging bus makes it a push pattern in ETL.
### [Transformer](https://github.com/amundsen-io/amundsen/tree/main/databuilder/databuilder/transformer "Transformer")
A transformer takes a record from either an extractor or from other transformers (via ChainedTransformer) to transform the record.
### [Loader](https://github.com/amundsen-io/amundsen/tree/main/databuilder/databuilder/loader "Loader")
A loader takes a record from a transformer or from an extractor directly and loads it to a sink, or a staging area. As the loading operates at a record-level, it's not capable of supporting atomicity.
### [Task](https://github.com/amundsen-io/amundsen/tree/main/databuilder/databuilder/task "Task")
A task orchestrates an extractor, a transformer, and a loader to perform a record-level operation.
### [Record](https://github.com/amundsen-io/amundsen/tree/main/databuilder/databuilder/models "Record")
A record is represented by one of [models](https://github.com/amundsen-io/amundsen/tree/main/databuilder/databuilder/models "models").
### [Publisher](https://github.com/amundsen-io/amundsen/tree/main/databuilder/databuilder/publisher "Publisher")
A publisher is an optional component. Its common usage is to support atomicity in job level and/or to easily support bulk load into the sink.
### [Job](https://github.com/amundsen-io/amundsen/tree/main/databuilder/databuilder/job "Job")
A job is the highest level component in Databuilder, and it orchestrates a task and, if any, a publisher.
## [Model](docs/models.md)
Models are abstractions representing the domain.
## List of extractors
#### [DBAPIExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/db_api_extractor.py "DBAPIExtractor")
An extractor that uses [Python Database API](https://www.python.org/dev/peps/pep-0249/ "Python Database API") interface. DBAPI requires three information, connection object that conforms DBAPI spec, a SELECT SQL statement, and a [model class](https://github.com/amundsen-io/amundsen/tree/main/databuilder/databuilder/models "model class") that correspond to the output of each row of SQL statement.
```python
job_config = ConfigFactory.from_dict({
'extractor.dbapi{}'.format(DBAPIExtractor.CONNECTION_CONFIG_KEY): db_api_conn,
'extractor.dbapi.{}'.format(DBAPIExtractor.SQL_CONFIG_KEY ): select_sql_stmt,
'extractor.dbapi.model_class': 'package.module_name.class_name'
})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=DBAPIExtractor(),
loader=AnyLoader()))
job.launch()
```
#### [GenericExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/generic_extractor.py "GenericExtractor")
An extractor that takes list of dict from user through config.
#### [HiveTableLastUpdatedExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/hive_table_last_updated_extractor.py "HiveTableLastUpdatedExtractor")
An extractor that extracts last updated time from Hive metastore and underlying file system. Although, hive metastore has a parameter called "last_modified_time", but it cannot be used as it provides DDL timestamp not DML timestamp.
For this reason, HiveTableLastUpdatedExtractor is utilizing underlying file of Hive to fetch latest updated date. However, it is not efficient to poke all files in Hive, and it only pokes underlying storage for non-partitioned table. For partitioned table, it will fetch partition created timestamp, and it's close enough for last updated timestamp.
As getting metadata from files could be time consuming there're several features to increase performance.
1. Support of multithreading to parallelize metadata fetching. Although, cpython's multithreading is not true multithreading as it's bounded by single core, getting metadata of file is mostly IO bound operation. Note that number of threads should be less or equal to number of connections.
1. User can pass where clause to only include certain schema and also remove certain tables. For example, by adding something like `TBL_NAME NOT REGEXP '(tmp|temp)` would eliminate unncecessary computation.
```python
job_config = ConfigFactory.from_dict({
'extractor.hive_table_last_updated.partitioned_table_where_clause_suffix': partitioned_table_where_clause,
'extractor.hive_table_last_updated.non_partitioned_table_where_clause_suffix'): non_partitioned_table_where_clause,
'extractor.hive_table_last_updated.extractor.sqlalchemy.{}'.format(
SQLAlchemyExtractor.CONN_STRING): connection_string,
'extractor.hive_table_last_updated.extractor.fs_worker_pool_size': pool_size,
'extractor.hive_table_last_updated.filesystem.{}'.format(FileSystem.DASK_FILE_SYSTEM): s3fs.S3FileSystem(
anon=False,
config_kwargs={'max_pool_connections': pool_size})})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=HiveTableLastUpdatedExtractor(),
loader=AnyLoader()))
job.launch()
```
#### [HiveTableMetadataExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/hive_table_metadata_extractor.py "HiveTableMetadataExtractor")
An extractor that extracts table and column metadata including database, schema, table name, table description, column name and column description from Hive metastore database.
```python
job_config = ConfigFactory.from_dict({
'extractor.hive_table_metadata.{}'.format(HiveTableMetadataExtractor.WHERE_CLAUSE_SUFFIX_KEY): where_clause_suffix,
'extractor.hive_table_metadata.extractor.sqlalchemy.{}'.format(SQLAlchemyExtractor.CONN_STRING): connection_string()})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=HiveTableMetadataExtractor(),
loader=AnyLoader()))
job.launch()
```
#### [CassandraExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/cassandra_extractor.py "CassandraExtractor")
An extractor that extracts table and column metadata including keyspace, table name, column name and column type from Apache Cassandra databases
```python
job_config = ConfigFactory.from_dict({
'extractor.cassandra.{}'.format(CassandraExtractor.CLUSTER_KEY): cluster_identifier_string,
'extractor.cassandra.{}'.format(CassandraExtractor.IPS_KEY): [127.0.0.1],
'extractor.cassandra.{}'.format(CassandraExtractor.KWARGS_KEY): {},
'extractor.cassandra.{}'.format(CassandraExtractor.FILTER_FUNCTION_KEY): my_filter_function,
})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=CassandraExtractor(),
loader=AnyLoader()))
job.launch()
```
If using the function filter options here is the function description
```python
def filter(keytab, table):
# return False if you don't want to add that table and True if you want to add
return True
```
If needed to define more args on the cassandra cluster you can pass through kwargs args
```python
config = ConfigFactory.from_dict({
'extractor.cassandra.{}'.format(CassandraExtractor.IPS_KEY): [127.0.0.1],
'extractor.cassandra.{}'.format(CassandraExtractor.KWARGS_KEY): {'port': 9042}
})
# it will call the cluster constructor like this
Cluster([127.0.0.1], **kwargs)
```
#### [GlueExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/glue_extractor.py "GlueExtractor")
An extractor that extracts table and column metadata including database, schema, table name, table description, column name and column description from AWS Glue metastore.
Before running make sure you have a working AWS profile configured and have access to search tables on Glue
```python
job_config = ConfigFactory.from_dict({
'extractor.glue.{}'.format(GlueExtractor.CLUSTER_KEY): cluster_identifier_string,
'extractor.glue.{}'.format(GlueExtractor.FILTER_KEY): [],
'extractor.glue.{}'.format(GlueExtractor.PARTITION_BADGE_LABEL_KEY): label_string,
})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=GlueExtractor(),
loader=AnyLoader()))
job.launch()
```
Optionally, you may add a partition badge label to the configuration. This will apply that label to all columns that are identified as partition keys in Glue.
If using the filters option here is the input format. For more information on filters visit [link](https://docs.aws.amazon.com/glue/latest/webapi/API_PropertyPredicate.html)
```
[
{
"Key": "string",
"Value": "string",
"Comparator": "EQUALS"|"GREATER_THAN"|"LESS_THAN"|"GREATER_THAN_EQUALS"|"LESS_THAN_EQUALS"
}
...
]
```
Example filtering on database and table. Note that Comparator can only apply to time fields.
```
[
{
"Key": "DatabaseName",
"Value": "my_database"
},
{
"Key": "Name",
"Value": "my_table"
}
]
```
#### [Delta-Lake-MetadataExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/delta_lake_metadata_extractor.py)
An extractor that runs on a spark cluster and obtains delta-lake metadata using spark sql commands.
This custom solution is currently necessary because the hive metastore does not contain all metadata information for delta-lake tables.
For simplicity, this extractor can also be used for all hive tables as well.
Because it must run on a spark cluster,
it is required that you have an operator (for example a [databricks submit run operator](https://airflow.apache.org/docs/stable/_modules/airflow/contrib/operators/databricks_operator.html))
that calls the configuration code on a spark cluster.
```python
spark = SparkSession.builder.appName("Amundsen Delta Lake Metadata Extraction").getOrCreate()
job_config = create_delta_lake_job_config()
dExtractor = DeltaLakeMetadataExtractor()
dExtractor.set_spark(spark)
job = DefaultJob(conf=job_config,
task=DefaultTask(extractor=dExtractor, loader=FsNeo4jCSVLoader()),
publisher=Neo4jCsvPublisher())
job.launch()
```
The delta lake extractor supports extraction of complex data types to be indexed and searchable.
```
struct>,f:map>>
Will be extracted as:
a int
b string
c array>
c.d int
c.e string
f map>
f.g int
f.h string
```
This functionality is behind a configuration value. Simply set EXTRACT_NESTED_COLUMNS to True in the job config.
You can check out the sample deltalake metadata script for a full example.
#### [DremioMetadataExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/dremio_metadata_extractor.py)
An extractor that extracts table and column metadata including database, schema, table name, table description, column name and column description from [Dremio](https://www.dremio.com).
Before running make sure that you have the Dremio ODBC driver installed. Default config values assume the default driver name for the [MacBook install](https://docs.dremio.com/drivers/mac-odbc.html).
```python
job_config = ConfigFactory.from_dict({
'extractor.dremio.{}'.format(DremioMetadataExtractor.DREMIO_USER_KEY): DREMIO_USER,
'extractor.dremio.{}'.format(DremioMetadataExtractor.DREMIO_PASSWORD_KEY): DREMIO_PASSWORD,
'extractor.dremio.{}'.format(DremioMetadataExtractor.DREMIO_HOST_KEY): DREMIO_HOST})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=DremioMetadataExtractor(),
loader=AnyLoader()))
job.launch()
```
#### [DruidMetadataExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/druid_metadata_extractor.py)
An extractor that extracts table and column metadata including database, schema, table name, table description, column name and column description from a [Druid](https://druid.apache.org/) DB.
The `where_clause_suffix` could be defined, normally you would like to filter out the in `INFORMATION_SCHEMA`.
You could specify the following job config
```python
conn_string = "druid+https://{host}:{port}/druid/v2/sql/".format(
host=druid_broker_host,
port=443
)
job_config = ConfigFactory.from_dict({
'extractor.druid_metadata.{}'.format(PostgresMetadataExtractor.WHERE_CLAUSE_SUFFIX_KEY): where_clause_suffix,
'extractor.druid_metadata.extractor.sqlalchemy.{}'.format(SQLAlchemyExtractor.CONN_STRING): conn_string()})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=DruidMetadataExtractor(),
loader=AnyLoader()))
job.launch()
```
#### [OracleMetadataExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/oracle_metadata_extractor.py "OracleMetadataExtractor")
An extractor that extracts table and column metadata including database, schema, table name, table description, column name and column description from the Oracle database.
By default, the Oracle database name is 'oracle'. To override this, set `CLUSTER_KEY` to what you wish to use as the cluster name.
The `where_clause_suffix` below should define which schemas you'd like to query. The SQL query driving the extraction is defined [here](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/oracle_metadata_extractor.py)
```python
job_config = ConfigFactory.from_dict({
'extractor.oracle_metadata.{}'.format(OracleMetadataExtractor.WHERE_CLAUSE_SUFFIX_KEY): where_clause_suffix,
'extractor.oracle_metadata.extractor.sqlalchemy.{}'.format(SQLAlchemyExtractor.CONN_STRING): connection_string()})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=OracleMetadataExtractor(),
loader=AnyLoader()))
job.launch()
```
#### [PostgresMetadataExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/postgres_metadata_extractor.py "PostgresMetadataExtractor")
An extractor that extracts table and column metadata including database, schema, table name, table description, column name and column description from a Postgres or Redshift database.
By default, the Postgres/Redshift database name is used as the cluster name. To override this, set `USE_CATALOG_AS_CLUSTER_NAME`
to `False`, and `CLUSTER_KEY` to what you wish to use as the cluster name.
The `where_clause_suffix` below should define which schemas you'd like to query (see [the sample dag](https://github.com/amundsen-io/amundsen/blob/main/databuilder/example/dags/postgres_sample_dag.py) for an example).
The SQL query driving the extraction is defined [here](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/postgres_metadata_extractor.py)
```python
job_config = ConfigFactory.from_dict({
'extractor.postgres_metadata.{}'.format(PostgresMetadataExtractor.WHERE_CLAUSE_SUFFIX_KEY): where_clause_suffix,
'extractor.postgres_metadata.{}'.format(PostgresMetadataExtractor.USE_CATALOG_AS_CLUSTER_NAME): True,
'extractor.postgres_metadata.extractor.sqlalchemy.{}'.format(SQLAlchemyExtractor.CONN_STRING): connection_string()})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=PostgresMetadataExtractor(),
loader=AnyLoader()))
job.launch()
```
#### [MSSQLMetadataExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/mssql_metadata_extractor.py "PostgresMetadataExtractor")
An extractor that extracts table and column metadata including database, schema, table name, table description, column name and column description from a Microsoft SQL database.
By default, the Microsoft SQL Server Database name is used as the cluster name. To override this, set `USE_CATALOG_AS_CLUSTER_NAME`
to `False`, and `CLUSTER_KEY` to what you wish to use as the cluster name.
The `where_clause_suffix` below should define which schemas you'd like to query (`"('dbo','sys')"`).
The SQL query driving the extraction is defined [here](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/mssql_metadata_extractor.py)
This extractor is highly derived from [PostgresMetadataExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/postgres_metadata_extractor.py "PostgresMetadataExtractor").
```python
job_config = ConfigFactory.from_dict({
'extractor.mssql_metadata.{}'.format(MSSQLMetadataExtractor.WHERE_CLAUSE_SUFFIX_KEY): where_clause_suffix,
'extractor.mssql_metadata.{}'.format(MSSQLMetadataExtractor.USE_CATALOG_AS_CLUSTER_NAME): True,
'extractor.mssql_metadata.extractor.sqlalchemy.{}'.format(SQLAlchemyExtractor.CONN_STRING): connection_string()})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=MSSQLMetadataExtractor(),
loader=AnyLoader()))
job.launch()
```
#### [MysqlMetadataExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/mysql_metadata_extractor.py "MysqlMetadataExtractor")
An extractor that extracts table and column metadata including database, schema, table name, table description, column name and column description from a MYSQL database.
By default, the MYSQL database name is used as the cluster name. To override this, set `USE_CATALOG_AS_CLUSTER_NAME`
to `False`, and `CLUSTER_KEY` to what you wish to use as the cluster name.
The `where_clause_suffix` below should define which schemas you'd like to query.
The SQL query driving the extraction is defined [here](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/mysql_metadata_extractor.py)
```python
job_config = ConfigFactory.from_dict({
'extractor.mysql_metadata.{}'.format(MysqlMetadataExtractor.WHERE_CLAUSE_SUFFIX_KEY): where_clause_suffix,
'extractor.mysql_metadata.{}'.format(MysqlMetadataExtractor.USE_CATALOG_AS_CLUSTER_NAME): True,
'extractor.mysql_metadata.extractor.sqlalchemy.{}'.format(SQLAlchemyExtractor.CONN_STRING): connection_string()})
job = DefaultJob(conf=job_config,
task=DefaultTask(extractor=MysqlMetadataExtractor(), loader=FsNeo4jCSVLoader()),
publisher=Neo4jCsvPublisher())
job.launch()
```
#### [Db2MetadataExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/db2_metadata_extractor.py "Db2MetadataExtractor")
An extractor that extracts table and column metadata including database, schema, table name, table description, column name and column description from a Unix, Windows or Linux Db2 database or BigSQL.
The `where_clause_suffix` below should define which schemas you'd like to query or those that you would not (see [the sample data loader](https://github.com/amundsen-io/amundsen/blob/main/databuilder/example/sample_db2_data_loader.py) for an example).
The SQL query driving the extraction is defined [here](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/db2_metadata_extractor.py)
```python
job_config = ConfigFactory.from_dict({
'extractor.db2_metadata.{}'.format(Db2MetadataExtractor.WHERE_CLAUSE_SUFFIX_KEY): where_clause_suffix,
'extractor.db2_metadata.extractor.sqlalchemy.{}'.format(SQLAlchemyExtractor.CONN_STRING): connection_string()})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=Db2MetadataExtractor(),
loader=AnyLoader()))
job.launch()
```
#### [SnowflakeMetadataExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/snowflake_metadata_extractor.py "SnowflakeMetadataExtractor")
An extractor that extracts table and column metadata including database, schema, table name, table description, column name and column description from a Snowflake database.
By default, the Snowflake database name is used as the cluster name. To override this, set `USE_CATALOG_AS_CLUSTER_NAME`
to `False`, and `CLUSTER_KEY` to what you wish to use as the cluster name.
By default, the Snowflake database is set to `PROD`. To override this, set `DATABASE_KEY`
to `WhateverNameOfYourDb`.
By default, the Snowflake schema is set to `INFORMATION_SCHEMA`. To override this, set `SCHEMA_KEY`
to `WhateverNameOfYourSchema`.
Note that `ACCOUNT_USAGE` is a separate schema which allows users to query a wider set of data at the cost of latency.
Differences are defined [here](https://docs.snowflake.com/en/sql-reference/account-usage.html#differences-between-account-usage-and-information-schema)
The `where_clause_suffix` should define which schemas you'd like to query (see [the sample dag](https://github.com/amundsen-io/amundsen/blob/main/databuilder/example/scripts/sample_snowflake_data_loader.py) for an example).
The SQL query driving the extraction is defined [here](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/snowflake_metadata_extractor.py)
```python
job_config = ConfigFactory.from_dict({
'extractor.snowflake.{}'.format(SnowflakeMetadataExtractor.SNOWFLAKE_DATABASE_KEY): 'YourDbName',
'extractor.snowflake.{}'.format(SnowflakeMetadataExtractor.WHERE_CLAUSE_SUFFIX_KEY): where_clause_suffix,
'extractor.snowflake.{}'.format(SnowflakeMetadataExtractor.USE_CATALOG_AS_CLUSTER_NAME): True,
'extractor.snowflake.extractor.sqlalchemy.{}'.format(SQLAlchemyExtractor.CONN_STRING): connection_string()})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=SnowflakeMetadataExtractor(),
loader=AnyLoader()))
job.launch()
```
#### [GenericUsageExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/generic_usage_extractor.py "GenericUsageExtractor")
An extractor that extracts table popularity metadata from a custom created Snowflake table (created by a script that may look like [this scala script](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/example/scripts/sample_snowflake_table_usage.scala "sample_snowflake_table_usage")). You can create a DAG using the [Databricks Operator](https://github.com/apache/airflow/blob/main/airflow/providers/databricks/operators/databricks.py) and run this script within Databricks or wherever you are able to run Scala.
By default, `snowflake` is used as the database name. `ColumnReader` has the datasource as its `database` input, and database as its `cluster` input.
The following inputs are related to where you create your Snowflake popularity table.
By default, the Snowflake popularity database is set to `PROD`. To override this, set `POPULARITY_TABLE_DATABASE`
to `WhateverNameOfYourDb`.
By default, the Snowflake popularity schema is set to `SCHEMA`. To override this, set `POPULARTIY_TABLE_SCHEMA`
to `WhateverNameOfYourSchema`.
By default, the Snowflake popularity table is set to `TABLE`. To override this, set `POPULARITY_TABLE_NAME`
to `WhateverNameOfYourTable`.
The `where_clause_suffix` should define any filtering you'd like to include in your query. For example, this may include `user_email`s that you don't want to include in your popularity definition.
```python
job_config = ConfigFactory.from_dict({
f'extractor.generic_usage.extractor.sqlalchemy.{SQLAlchemyExtractor.CONN_STRING}': connection_string(),
f'extractor.generic_usage.{GenericUsageExtractor.WHERE_CLAUSE_SUFFIX_KEY}': where_clause_suffix,
f'extractor.generic_usage.{GenericUsageExtractor.POPULARITY_TABLE_DATABASE}': 'WhateverNameOfYourDb',
f'extractor.generic_usage.{GenericUsageExtractor.POPULARTIY_TABLE_SCHEMA}': 'WhateverNameOfYourSchema',
f'extractor.generic_usage.{GenericUsageExtractor.POPULARITY_TABLE_NAME}': 'WhateverNameOfYourTable',
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=GenericUsageExtractor(),
loader=AnyLoader()))
job.launch()
```
#### [SnowflakeTableLastUpdatedExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/snowflake_table_last_updated_extractor.py "SnowflakeTableLastUpdatedExtractor")
An extractor that extracts table last updated timestamp from a Snowflake database.
It uses same configs as the `SnowflakeMetadataExtractor` described above.
The SQL query driving the extraction is defined [here](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/snowflake_table_last_updated_extractor.py)
```python
job_config = ConfigFactory.from_dict({
'extractor.snowflake_table_last_updated.{}'.format(SnowflakeTableLastUpdatedExtractor.SNOWFLAKE_DATABASE_KEY): 'YourDbName',
'extractor.snowflake_table_last_updated.{}'.format(SnowflakeTableLastUpdatedExtractor.WHERE_CLAUSE_SUFFIX_KEY): where_clause_suffix,
'extractor.snowflake_table_last_updated.{}'.format(SnowflakeTableLastUpdatedExtractor.USE_CATALOG_AS_CLUSTER_NAME): True,
'extractor.snowflake_table_last_updated.extractor.sqlalchemy.{}'.format(SQLAlchemyExtractor.CONN_STRING): connection_string()})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=SnowflakeTableLastUpdatedExtractor(),
loader=AnyLoader()))
job.launch()
```
#### [BigQueryMetadataExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/bigquery_metadata_extractor.py "BigQuery Metdata Extractor")
An extractor that extracts table and column metadata including database, schema, table name, table description, column name and column description from a Bigquery database.
The API calls driving the extraction is defined [here](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/bigquery_metadata_extractor.py)
You will need to create a service account for reading metadata and grant it "BigQuery Metadata Viewer" access to all of your datasets. This can all be done via the bigquery ui.
Download the credentials file and store it securely. Set the `GOOGLE_APPLICATION_CREDENTIALS` environment varible to the location of your credtials files and your code should have access to everything it needs.
You can configure bigquery like this. You can optionally set a label filter if you only want to pull tables with a certain label.
```python
job_config = {
'extractor.bigquery_table_metadata.{}'.format(
BigQueryMetadataExtractor.PROJECT_ID_KEY
): gcloud_project
}
if label_filter:
job_config[
'extractor.bigquery_table_metadata.{}'
.format(BigQueryMetadataExtractor.FILTER_KEY)
] = label_filter
task = DefaultTask(extractor=BigQueryMetadataExtractor(),
loader=csv_loader,
transformer=NoopTransformer())
job = DefaultJob(conf=ConfigFactory.from_dict(job_config),
task=task,
publisher=Neo4jCsvPublisher())
job.launch()
```
#### [Neo4jEsLastUpdatedExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/neo4j_es_last_updated_extractor.py "Neo4jEsLastUpdatedExtractor")
An extractor that basically get current timestamp and passes it GenericExtractor. This extractor is basically being used to create timestamp for "Amundsen was last indexed on ..." in Amundsen web page's footer.
#### [Neo4jExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/neo4j_extractor.py "Neo4jExtractor")
An extractor that extracts records from Neo4j based on provided [Cypher query](https://neo4j.com/developer/cypher/ "Cypher query"). One example is to extract data from Neo4j so that it can transform and publish to Elasticsearch.
```python
job_config = ConfigFactory.from_dict({
'extractor.neo4j.{}'.format(Neo4jExtractor.CYPHER_QUERY_CONFIG_KEY): cypher_query,
'extractor.neo4j.{}'.format(Neo4jExtractor.GRAPH_URL_CONFIG_KEY): neo4j_endpoint,
'extractor.neo4j.{}'.format(Neo4jExtractor.MODEL_CLASS_CONFIG_KEY): 'package.module.class_name',
'extractor.neo4j.{}'.format(Neo4jExtractor.NEO4J_AUTH_USER): neo4j_user,
'extractor.neo4j.{}'.format(Neo4jExtractor.NEO4J_AUTH_PW): neo4j_password},
'extractor.neo4j.{}'.format(Neo4jExtractor.NEO4J_ENCRYPTED): True})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=Neo4jExtractor(),
loader=AnyLoader()))
job.launch()
```
#### [Neo4jSearchDataExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/neo4j_search_data_extractor.py "Neo4jSearchDataExtractor")
An extractor that is extracting Neo4j utilizing Neo4jExtractor where CYPHER query is already embedded in it.
```python
job_config = ConfigFactory.from_dict({
'extractor.search_data.extractor.neo4j.{}'.format(Neo4jExtractor.GRAPH_URL_CONFIG_KEY): neo4j_endpoint,
'extractor.search_data.extractor.neo4j.{}'.format(Neo4jExtractor.MODEL_CLASS_CONFIG_KEY): 'databuilder.models.neo4j_data.Neo4jDataResult',
'extractor.search_data.extractor.neo4j.{}'.format(Neo4jExtractor.NEO4J_AUTH_USER): neo4j_user,
'extractor.search_data.extractor.neo4j.{}'.format(Neo4jExtractor.NEO4J_AUTH_PW): neo4j_password},
'extractor.search_data.extractor.neo4j.{}'.format(Neo4jExtractor.NEO4J_ENCRYPTED): False})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=Neo4jSearchDataExtractor(),
loader=AnyLoader()))
job.launch()
```
#### [AtlasSearchDataExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/atlas_search_data_extractor.py "AtlasSearchDataExtractor")
An extractor that is extracting Atlas Data to index compatible with Elasticsearch Search Proxy.
```
/* Detailed source-code truncated for AI context efficiency. */
```
#### [VerticaMetadataExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/vertica_metadata_extractor.py "MysqlMetadataExtractor")
An extractor that extracts table and column metadata including database, schema, table name, column name and column datatype from a Vertica database.
A sample loading script for Vertica is provided [here](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/databuilder/example/scripts/sample_vertica_loader.py)
By default, the Vertica database name is used as the cluster name. The `where_clause_suffix` in the example can be used to define which schemas you would like to query.
#### [SQLAlchemyExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/sql_alchemy_extractor.py "SQLAlchemyExtractor")
An extractor utilizes [SQLAlchemy](https://www.sqlalchemy.org/ "SQLAlchemy") to extract record from any database that support SQL Alchemy.
```python
job_config = ConfigFactory.from_dict({
'extractor.sqlalchemy.{}'.format(SQLAlchemyExtractor.CONN_STRING): connection_string(),
'extractor.sqlalchemy.{}'.format(SQLAlchemyExtractor.EXTRACT_SQL): sql,
'extractor.sqlalchemy.model_class': 'package.module.class_name'})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=SQLAlchemyExtractor(),
loader=AnyLoader()))
job.launch()
```
#### [DbtExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/dbt_extractor.py "SQLAlchemyExtractor")
This extractor utilizes the [dbt](https://www.getdbt.com/ "dbt") output files `catalog.json` and `manifest.json` to extract metadata and ingest it into Amundsen. The `catalog.json` and `manifest.json` can both be generated by running `dbt docs generate` in your dbt project. Visit the [dbt artifacts page](https://docs.getdbt.com/reference/artifacts/dbt-artifacts "dbt artifacts") for more information.
The `DbtExtractor` can currently create the following:
- Tables and their definitions
- Columns and their definitions
- Table level lineage
- dbt tags (as Amundsen badges or tags)
- Table Sources (e.g. link to GitHib where the dbt template resides)
```python
job_config = ConfigFactory.from_dict({
# Required args
f'extractor.dbt.{DbtExtractor.DATABASE_NAME}': 'snowflake',
f'extractor.dbt.{DbtExtractor.MANIFEST_JSON}': catalog_file_loc, # File location
f'extractor.dbt.{DbtExtractor.DATABASE_NAME}': json.dumps(manifest_data), # JSON Dumped object
# Optional args
f'extractor.dbt.{DbtExtractor.SOURCE_URL}': 'https://github.com/your-company/your-repo/tree/main',
f'extractor.dbt.{DbtExtractor.EXTRACT_TABLES}': True,
f'extractor.dbt.{DbtExtractor.EXTRACT_DESCRIPTIONS}': True,
f'extractor.dbt.{DbtExtractor.EXTRACT_TAGS}': True,
f'extractor.dbt.{DbtExtractor.IMPORT_TAGS_AS}': 'badges',
f'extractor.dbt.{DbtExtractor.EXTRACT_LINEAGE}': True,
})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=DbtExtractor(),
loader=AnyLoader()))
job.launch()
```
### [RestAPIExtractor](./databuilder/extractor/restapi/rest_api_extractor.py)
A extractor that utilizes [RestAPIQuery](#rest-api-query) to extract data. RestAPIQuery needs to be constructed ([example](./databuilder/extractor/dashboard/mode_analytics/mode_dashboard_extractor.py#L40)) and needs to be injected to RestAPIExtractor.
### Mode Dashboard Extractor
Here are extractors that extracts metadata information from Mode via Mode's REST API.
Prerequisite:
1. You will need to [create API access token](https://mode.com/developer/api-reference/authentication/) that has admin privilege.
2. You will need organization code. This is something you can easily get by looking at one of Mode report's URL.
`https://app.mode.com//reports/report_token`
#### [ModeDashboardExtractor](./databuilder/extractor/dashboard/mode_analytics/mode_dashboard_extractor.py)
A Extractor that extracts core metadata on Mode dashboard. https://app.mode.com/
It extracts list of reports that consists of:
Dashboard group name (Space name)
Dashboard group id (Space token)
Dashboard group description (Space description)
Dashboard name (Report name)
Dashboard id (Report token)
Dashboard description (Report description)
Other information such as report run, owner, chart name, query name is in separate extractor.
It calls two APIs ([spaces API](https://mode.com/developer/discovery-api/analytics/spaces/) and [reports API](https://mode.com/developer/discovery-api/analytics/reports/)) joining together.
You can create Databuilder job config like this.
```
/* Detailed source-code truncated for AI context efficiency. */
```
#### [ModeDashboardOwnerExtractor](./databuilder/extractor/dashboard/mode_analytics/mode_dashboard_owner_extractor.py)
An Extractor that extracts Dashboard owner. Mode itself does not have concept of owner and it will use creator as owner. Note that if user left the organization, it would skip the dashboard.
You can create Databuilder job config like this. (configuration related to loader and publisher is omitted as it is mostly the same. Please take a look at this [example](#ModeDashboardExtractor) for the configuration that holds loader and publisher.
```python
extractor = ModeDashboardOwnerExtractor()
task = DefaultTask(extractor=extractor,
loader=FsNeo4jCSVLoader(), )
job_config = ConfigFactory.from_dict({
'{}.{}'.format(extractor.get_scope(), ORGANIZATION): organization,
'{}.{}'.format(extractor.get_scope(), MODE_BEARER_TOKEN): mode_bearer_token,
})
job = DefaultJob(conf=job_config,
task=task,
publisher=Neo4jCsvPublisher())
job.launch()
```
#### [ModeDashboardLastSuccessfulExecutionExtractor](./databuilder/extractor/dashboard/mode_analytics/mode_dashboard_last_successful_executions_extractor.py)
A Extractor that extracts Mode dashboard's last successful run (execution) timestamp.
You can create Databuilder job config like this. (configuration related to loader and publisher is omitted as it is mostly the same. Please take a look at this [example](#ModeDashboardExtractor) for the configuration that holds loader and publisher.
```python
extractor = ModeDashboardLastSuccessfulExecutionExtractor()
task = DefaultTask(extractor=extractor, loader=FsNeo4jCSVLoader())
job_config = ConfigFactory.from_dict({
'{}.{}'.format(extractor.get_scope(), ORGANIZATION): organization,
'{}.{}'.format(extractor.get_scope(), MODE_BEARER_TOKEN): mode_bearer_token,
})
job = DefaultJob(conf=job_config,
task=task,
publisher=Neo4jCsvPublisher())
job.launch()
```
#### [ModeDashboardExecutionsExtractor](./databuilder/extractor/dashboard/mode_analytics/mode_dashboard_executions_extractor.py)
A Extractor that extracts last run (execution) status and timestamp.
You can create Databuilder job config like this. (configuration related to loader and publisher is omitted as it is mostly the same. Please take a look at this [example](#ModeDashboardExtractor) for the configuration that holds loader and publisher.
```python
extractor = ModeDashboardExecutionsExtractor()
task = DefaultTask(extractor=extractor, loader=FsNeo4jCSVLoader())
job_config = ConfigFactory.from_dict({
'{}.{}'.format(extractor.get_scope(), ORGANIZATION): organization,
'{}.{}'.format(extractor.get_scope(), MODE_BEARER_TOKEN): mode_bearer_token,
})
job = DefaultJob(conf=job_config,
task=task,
publisher=Neo4jCsvPublisher())
job.launch()
```
#### [ModeDashboardLastModifiedTimestampExtractor](./databuilder/extractor/dashboard/mode_analytics/mode_dashboard_last_modified_timestamp_extractor.py)
A Extractor that extracts Mode dashboard's last modified timestamp.
You can create Databuilder job config like this. (configuration related to loader and publisher is omitted as it is mostly the same. Please take a look at this [example](#ModeDashboardExtractor) for the configuration that holds loader and publisher.
```python
extractor = ModeDashboardLastModifiedTimestampExtractor()
task = DefaultTask(extractor=extractor, loader=FsNeo4jCSVLoader())
job_config = ConfigFactory.from_dict({
'{}.{}'.format(extractor.get_scope(), ORGANIZATION): organization,
'{}.{}'.format(extractor.get_scope(), MODE_BEARER_TOKEN): mode_bearer_token,
})
job = DefaultJob(conf=job_config,
task=task,
publisher=Neo4jCsvPublisher())
job.launch()
```
#### [ModeDashboardQueriesExtractor](./databuilder/extractor/dashboard/mode_analytics/mode_dashboard_queries_extractor.py)
A Extractor that extracts Mode's query information.
You can create Databuilder job config like this. (configuration related to loader and publisher is omitted as it is mostly the same. Please take a look at this [example](#ModeDashboardExtractor) for the configuration that holds loader and publisher.
```python
extractor = ModeDashboardQueriesExtractor()
task = DefaultTask(extractor=extractor, loader=FsNeo4jCSVLoader())
job_config = ConfigFactory.from_dict({
'{}.{}'.format(extractor.get_scope(), ORGANIZATION): organization,
'{}.{}'.format(extractor.get_scope(), MODE_BEARER_TOKEN): mode_bearer_token,
})
job = DefaultJob(conf=job_config,
task=task,
publisher=Neo4jCsvPublisher())
job.launch()
```
#### [ModeDashboardChartsBatchExtractor](./databuilder/extractor/dashboard/mode_analytics/mode_dashboard_charts_batch_extractor.py)
A Extractor that extracts Mode Dashboard charts metadata.
You can create Databuilder job config like this. (configuration related to loader and publisher is omitted as it is mostly the same. Please take a look at this [example](#ModeDashboardExtractor) for the configuration that holds loader and publisher.
```python
extractor = ModeDashboardChartsBatchExtractor()
task = DefaultTask(extractor=extractor, loader=FsNeo4jCSVLoader())
job_config = ConfigFactory.from_dict({
'{}.{}'.format(extractor.get_scope(), ORGANIZATION): organization,
'{}.{}'.format(extractor.get_scope(), MODE_BEARER_TOKEN): mode_bearer_token,
})
job = DefaultJob(conf=job_config,
task=task,
publisher=Neo4jCsvPublisher())
job.launch()
```
#### [ModeDashboardUserExtractor](./databuilder/extractor/dashboard/mode_analytics/mode_dashboard_user_extractor.py)
A Extractor that extracts Mode user_id and then update User node.
You can create Databuilder job config like this. (configuration related to loader and publisher is omitted as it is mostly the same. Please take a look at this [example](#ModeDashboardExtractor) for the configuration that holds loader and publisher.
```python
extractor = ModeDashboardUserExtractor()
task = DefaultTask(extractor=extractor, loader=FsNeo4jCSVLoader())
job_config = ConfigFactory.from_dict({
'{}.{}'.format(extractor.get_scope(), ORGANIZATION): organization,
'{}.{}'.format(extractor.get_scope(), MODE_ACCESS_TOKEN): mode_token,
'{}.{}'.format(extractor.get_scope(), MODE_PASSWORD_TOKEN): mode_password,
})
job = DefaultJob(conf=job_config,
task=task,
publisher=Neo4jCsvPublisher())
job.launch()
```
#### [ModeDashboardUsageExtractor](./databuilder/extractor/dashboard/mode_analytics/mode_dashboard_usage_extractor.py)
A Extractor that extracts Mode dashboard's accumulated view count.
Note that this provides accumulated view count which does [not effectively show relevancy](./docs/dashboard_ingestion_guide.md#21-ingest-dashboard-usage-data-and-decorate-neo4j-over-base-data). Thus, fields from this extractor is not directly compatible with [DashboardUsage](./docs/models.md#dashboardusage) model.
If you are fine with `accumulated usage`, you could use TemplateVariableSubstitutionTransformer to transform Dict payload from [ModeDashboardUsageExtractor](./databuilder/extractor/dashboard/mode_analytics/mode_dashboard_usage_extractor.py) to fit [DashboardUsage](./docs/models.md#dashboardusage) and transform Dict to [DashboardUsage](./docs/models.md#dashboardusage) by [TemplateVariableSubstitutionTransformer](./databuilder/transformer/template_variable_substitution_transformer.py), and [DictToModel](./databuilder/transformer/dict_to_model.py) transformers. ([Example](./databuilder/extractor/dashboard/mode_analytics/mode_dashboard_queries_extractor.py#L36) on how to combining these two transformers)
#### [OpenLineageTableLineageExtractor](./databuilder/extractor/openlineage_extractor.py)
A Extractor that extracts table lineage information from [OpenLineage](https://github.com/OpenLineage/OpenLineage) events.
> :warning: Extractor expects input data in the form of **openLineage events in ndjson format**
Custom Openlineage json extraction keys may be set by passing those values:
* OpenLineageTableLineageExtractor.OL_INPUTS_KEY - json key for inputs list
* OpenLineageTableLineageExtractor.OL_OUTPUTS_KEY- json key for output list
* OpenLineageTableLineageExtractor.OL_DATASET_NAMESPACE_KEY - json key for namespace name (inputs/outputs scope)
* OpenLineageTableLineageExtractor.OL_DATASET_DATABASE_KEY - json key for database name (inputs/outputs scope)
* OpenLineageTableLineageExtractor.OL_DATASET_NAME_KEY - json key for dataset name (inputs/outputs scope)
```python
tmp_folder = f'/tmp/amundsen/lineage'
dict_config = {
f'loader.filesystem_csv_atlas.{FsAtlasCSVLoader.ENTITY_DIR_PATH}': f'{tmp_folder}/entities',
f'loader.filesystem_csv_atlas.{FsAtlasCSVLoader.RELATIONSHIP_DIR_PATH}': f'{tmp_folder}/relationships',
f'loader.filesystem_csv_atlas.{FsAtlasCSVLoader.SHOULD_DELETE_CREATED_DIR}': False,
f'publisher.atlas_csv_publisher.{AtlasCSVPublisher.ATLAS_CLIENT}': AtlasClient('http://localhost:21000', ('admin', 'admin')),
f'publisher.atlas_csv_publisher.{AtlasCSVPublisher.ENTITY_DIR_PATH}': f'{tmp_folder}/entities',
f'publisher.atlas_csv_publisher.{AtlasCSVPublisher.RELATIONSHIP_DIR_PATH}': f'{tmp_folder}/relationships',
f'publisher.atlas_csv_publisher.{AtlasCSVPublisher.ATLAS_ENTITY_CREATE_BATCH_SIZE}': 10,
f'extractor.openlineage_tablelineage.{OpenLineageTableLineageExtractor.CLUSTER_NAME}': 'datalab',
f'extractor.openlineage_tablelineage.{OpenLineageTableLineageExtractor.OL_DATASET_NAMESPACE_OVERRIDE}': 'hive_table',
f'extractor.openlineage_tablelineage.{OpenLineageTableLineageExtractor.TABLE_LINEAGE_FILE_LOCATION}': 'input_dir/openlineage_nd.json',
}
job_config = ConfigFactory.from_dict(dict_config)
task = DefaultTask(extractor=OpenLineageTableLineageExtractor(), loader=FsAtlasCSVLoader())
job = DefaultJob(conf=job_config,
task=task,
publisher=AtlasCSVPublisher())
job.launch()
```
### [RedashDashboardExtractor](./databuilder/extractor/dashboard/redash/redash_dashboard_extractor.py)
The included `RedashDashboardExtractor` provides support for extracting basic metadata for Redash dashboards (dashboard name, owner, URL, created/updated timestamps, and a generated description) and their associated queries (query name, URL, and raw query). It can be extended with a configurable table parser function to also support extraction of `DashboardTable` metadata. (See below for example usage.)
Note: `DashboardUsage` and `DashboardExecution` metadata are not supported in this extractor, as these concepts are not supported by the Redash API.
The `RedashDashboardExtractor` depends on the following Redash API endpoints: `GET /api/dashboards`, `GET /api/dashboards/`. It has been tested against Redash 8 and is also expected to work with Redash 9.
```python
extractor = RedashDashboardExtractor()
task = DefaultTask(extractor=extractor, loader=FsNeo4jCSVLoader())
job_config = ConfigFactory.from_dict({
'extractor.redash_dashboard.redash_base_url': redash_base_url, # ex: https://redash.example.org
'extractor.redash_dashboard.api_base_url': api_base_url, # ex: https://redash.example.org/api
'extractor.redash_dashboard.api_key': api_key, # ex: abc1234
'extractor.redash_dashboard.table_parser': table_parser, # ex: my_library.module.parse_tables
'extractor.redash_dashboard.redash_version': redash_version # ex: 8. optional, default=9
})
job = DefaultJob(conf=job_config,
task=task,
publisher=Neo4jCsvPublisher())
job.launch()
```
#### RedashDashboardExtractor: table_parser
The `RedashDashboardExtractor` extracts raw queries from each dashboard. You may optionally use these queries to parse out relations to tables in Amundsen. A table parser can be provided in the configuration for the `RedashDashboardExtractor`, as seen above. This function should have type signature `(RedashVisualizationWidget) -> Iterator[TableRelationData]`. For example:
```python
def parse_tables(viz_widget: RedashVisualizationWidget) -> Iterator[TableRelationData]:
# Each viz_widget corresponds to one query.
# viz_widget.data_source_id is the ID of the target DB in Redash.
# viz_widget.raw_query is the raw query (e.g., SQL).
if viz_widget.data_source_id == 123:
table_names = some_sql_parser(viz_widget.raw_query)
return [TableRelationData('some_db', 'prod', 'some_schema', tbl) for tbl in table_names]
return []
```
### [TableauDashboardExtractor](./databuilder/extractor/dashboard/tableau/tableau_dashboard_extractor.py)
The included `TableauDashboardExtractor` provides support for extracting basic metadata for Tableau workbooks. All Tableau extractors including this one use the [Tableau Metadata GraphQL API](https://help.tableau.com/current/api/metadata_api/en-us/index.html) to gather the metadata. Tableau "workbooks" are mapped to Amundsen dashboards, and the top-level project in which these workbooks preside is the dashboard group. The metadata it gathers is as follows:
- Dashboard name (Workbook name)
- Dashboard description (Workbook description)
- Dashboard creation timestamp (Workbook creation timestamp)
- Dashboard group name (Workbook top-level folder name)
- Dashboard and dashboard group URL
If you wish to exclude top-level projects from being loaded, specify their names in the `tableau_excluded_projects` list and workbooks from any of those projects will not be indexed.
Tableau's concept of "owners" does not map cleanly into Amundsen's understanding of owners, as the owner of a Tableau workbook is simply whoever updated it last, even if they made a very small change. This can prove problematic in determining the true point of contact for a workbook, so it's simply omitted for now. Similarly, the hierachy of `dashboard/query/chart` in Amundsen does not map into Tableau, where `charts` have only an optional relation to queries and vice versa. For these reasons, there are not extractors for either entity.
The Tableau Metadata API also does not support usage or execution statistics, so there are no extractors for these entities either.
Sample job config:
```python
extractor = TableauDashboardExtractor()
task = DefaultTask(extractor=extractor, loader=FsNeo4jCSVLoader())
job_config = ConfigFactory.from_dict({
'extractor.tableau_dashboard_metadata.tableau_host': tableau_host,
'extractor.tableau_dashboard_metadata.api_version': tableau_api_version,
'extractor.tableau_dashboard_metadata.site_name': tableau_site_name,
'extractor.tableau_dashboard_metadata.tableau_personal_access_token_name': tableau_personal_access_token_name,
'extractor.tableau_dashboard_metadata.tableau_personal_access_token_secret': tableau_personal_access_token_secret,
'extractor.tableau_dashboard_metadata.excluded_projects': tableau_excluded_projects,
'extractor.tableau_dashboard_metadata.cluster': tableau_dashboard_cluster,
'extractor.tableau_dashboard_metadata.database': tableau_dashboard_database,
'extractor.tableau_dashboard_metadata.transformer.timestamp_str_to_epoch.timestamp_format': "%Y-%m-%dT%H:%M:%SZ",
})
job = DefaultJob(conf=job_config,
task=task,
publisher=Neo4jCsvPublisher())
job.launch()
```
### [TableauDashboardTableExtractor](./databuilder/extractor/dashboard/tableau/tableau_dashboard_table_extractor.py)
The included `TableauDashboardTableExtractor` provides support for extracting table metadata from Tableau workbooks. The extractor assumes all the table entities have already been created; if you are interested in using the provided `TableauExternalTableExtractor`, make sure that job runs before this one, as it will create the tables required by this job. It also assumes that the dashboards are using their names as the primary ID.
A sample job config is shown below. Configuration related to the loader and publisher is omitted as it is mostly the same. Please take a look at this [example](#TableauDashboardExtractor) for the configuration that holds loader and publisher.
```python
extractor = TableauDashboardTableExtractor()
task = DefaultTask(extractor=extractor, loader=FsNeo4jCSVLoader())
job_config = ConfigFactory.from_dict({
'extractor.tableau_dashboard_table.tableau_host': tableau_host,
'extractor.tableau_dashboard_table.api_version': tableau_api_version,
'extractor.tableau_dashboard_table.site_name': tableau_site_name,
'extractor.tableau_dashboard_table.tableau_personal_access_token_name': tableau_personal_access_token_name,
'extractor.tableau_dashboard_table.tableau_personal_access_token_secret': tableau_personal_access_token_secret,
'extractor.tableau_dashboard_table.excluded_projects': tableau_excluded_projects,
'extractor.tableau_dashboard_table.cluster': tableau_dashboard_cluster,
'extractor.tableau_dashboard_table.database': tableau_dashboard_database,
'extractor.tableau_dashboard_table.external_cluster_name': tableau_external_table_cluster,
'extractor.tableau_dashboard_table.external_schema_name': tableau_external_table_schema,
})
job = DefaultJob(conf=job_config,
task=task,
publisher=Neo4jCsvPublisher())
job.launch()
```
### [TableauDashboardQueryExtractor](./databuilder/extractor/dashboard/tableau/tableau_dashboard_query_extractor.py)
The included `TableauDashboardQueryExtractor` provides support for extracting query metadata from Tableau workbooks. It retrives the name and query text for each custom SQL query.
A sample job config is shown below. Configuration related to the loader and publisher is omitted as it is mostly the same. Please take a look at this [example](#TableauDashboardExtractor) for the configuration that holds loader and publisher.
```python
extractor = TableauDashboardQueryExtractor()
task = DefaultTask(extractor=extractor, loader=FsNeo4jCSVLoader())
job_config = ConfigFactory.from_dict({
'extractor.tableau_dashboard_query.tableau_host': tableau_host,
'extractor.tableau_dashboard_query.api_version': tableau_api_version,
'extractor.tableau_dashboard_query.site_name': tableau_site_name,
'extractor.tableau_dashboard_query.tableau_personal_access_token_name': tableau_personal_access_token_name,
'extractor.tableau_dashboard_query.tableau_personal_access_token_secret': tableau_personal_access_token_secret,
'extractor.tableau_dashboard_query.excluded_projects': tableau_excluded_projects,
'extractor.tableau_dashboard_query.cluster': tableau_dashboard_cluster,
'extractor.tableau_dashboard_query.database': tableau_dashboard_database,
})
job = DefaultJob(conf=job_config,
task=task,
publisher=Neo4jCsvPublisher())
job.launch()
```
### [TableauDashboardLastModifiedExtractor](./databuilder/extractor/dashboard/tableau/tableau_dashboard_last_modified_extractor.py)
The included `TableauDashboardLastModifiedExtractor` provides support for extracting the last updated timestamp for Tableau workbooks.
A sample job config is shown below. Configuration related to the loader and publisher is omitted as it is mostly the same. Please take a look at this [example](#TableauDashboardExtractor) for the configuration that holds loader and publisher.
```python
extractor = TableauDashboardQueryExtractor()
task = DefaultTask(extractor=extractor, loader=FsNeo4jCSVLoader())
job_config = ConfigFactory.from_dict({
'extractor.tableau_dashboard_last_modified.tableau_host': tableau_host,
'extractor.tableau_dashboard_last_modified.api_version': tableau_api_version,
'extractor.tableau_dashboard_last_modified.site_name': tableau_site_name,
'extractor.tableau_dashboard_last_modified.tableau_personal_access_token_name': tableau_personal_access_token_name,
'extractor.tableau_dashboard_last_modified.tableau_personal_access_token_secret': tableau_personal_access_token_secret,
'extractor.tableau_dashboard_last_modified.excluded_projects': tableau_excluded_projects,
'extractor.tableau_dashboard_last_modified.cluster': tableau_dashboard_cluster,
'extractor.tableau_dashboard_last_modified.database': tableau_dashboard_database,
'extractor.tableau_dashboard_last_modified.transformer.timestamp_str_to_epoch.timestamp_format': "%Y-%m-%dT%H:%M:%SZ",
})
job = DefaultJob(conf=job_config,
task=task,
publisher=Neo4jCsvPublisher())
job.launch()
```
### [TableauExternalTableExtractor](./databuilder/extractor/dashboard/tableau/tableau_external_table_extractor.py)
The included `TableauExternalTableExtractor` provides support for extracting external table entities referenced by Tableau workbooks. In this context, "external" tables are "tables" that are not from a typical database, and are loaded using some other data format, like CSV files.
This extractor has been tested with the following types of external tables; feel free to add others, but it's recommended
to test them in a non-production instance first to be safe.
- Excel spreadsheets
- Text files (including CSV files)
- Salesforce connections
- Google Sheets connections
Use the `external_table_types` list config option to specify which external connection types you would like to index;
refer to your Tableau instance for the exact formatting of each connection type string.
Excel spreadsheets, Salesforce connections, and Google Sheets connections are all classified as
"databases" in terms of Tableau's Metadata API, with their "subsheets" forming their "tables" when
present. However, these tables are not assigned a schema, this extractor chooses to use the name
of the parent sheet as the schema, and assign a new table to each subsheet. The connection type is
always used as the database, and for text files, the schema is set using the `external_schema_name`
config option. Since these external tables are usually named for human consumption only and often
contain a wider range of characters, all inputs are sanitized to remove any problematic
occurences before they are inserted: see the `sanitize` methods `TableauDashboardUtils` for specifics.
A more concrete example: if one had a Google Sheet titled "Growth by Region" with 2 subsheets called
"FY19 Report" and "FY20 Report", two tables would be generated with the following keys:
`googlesheets://external.growth_by_region/FY_19_Report`
`googlesheets://external.growth_by_region/FY_20_Report`
A sample job config is shown below. Configuration related to the loader and publisher is omitted as it is mostly the same. Please take a look at this [example](#TableauDashboardExtractor) for the configuration that holds loader and publisher.
```python
extractor = TableauExternalTableExtractor()
task = DefaultTask(extractor=extractor, loader=FsNeo4jCSVLoader())
job_config = ConfigFactory.from_dict({
'extractor.tableau_external_table.tableau_host': tableau_host,
'extractor.tableau_external_table.api_version': tableau_api_version,
'extractor.tableau_external_table.site_name': tableau_site_name,
'extractor.tableau_external_table.tableau_personal_access_token_name': tableau_personal_access_token_name,
'extractor.tableau_external_table.tableau_personal_access_token_secret': tableau_personal_access_token_secret,
'extractor.tableau_external_table.excluded_projects': tableau_excluded_projects,
'extractor.tableau_external_table.cluster': tableau_dashboard_cluster,
'extractor.tableau_external_table.database': tableau_dashboard_database,
'extractor.tableau_external_table.external_cluster_name': tableau_external_table_cluster,
'extractor.tableau_external_table.external_schema_name': tableau_external_table_schema,
'extractor.tableau_external_table.external_table_types': tableau_external_table_types
})
job = DefaultJob(conf=job_config,
task=task,
publisher=Neo4jCsvPublisher())
job.launch()
```
### [DatabricksSQLDashboardExtractor](./databuilder/extractor/dashboard/databricks_sql/databricks_sql_dashboard_extractor.py)
The `DatabricksSQLDashboardExtractor` extracts metadata about dashboards created in [Databricks SQL](https://databricks.com/product/databricks-sql)
The only configuration you need is a Databricks Host Name (i.e `https://my-company.cloud.databricks.com`) and a valid Databricks API Token. Make sure that the user that generated this token has permissions to read dashboards.
Example:
```python
extractor = DatabricksSQLDashboardExtractor()
task = DefaultTask(extractor=extractor, loader=FsNeo4jCSVLoader())
job_config = ConfigFactory.from_dict({
f"extractor.databricks_sql_extractor.{DatabricksSQLDashboardExtractor.DATABRICKS_HOST_KEY}": "MY-DATABRICKS-API-TOKEN",
f"extractor.databricks_sql_extractor.{DatabricksSQLDashboardExtractor.DATABRICKS_API_TOKEN_KEY}": "https://my-company.cloud.databricks.com",
# ...plus nessescary configs for neo4j...
})
job = DefaultJob(
conf=job_config,
task=task,
publisher=Neo4jCsvPublisher(),
)
job.launch()
```
### [ApacheSupersetMetadataExtractor](./databuilder/extractor/dashboard/apache_superset/apache_superset_metadata_extractor.py)
The included `ApacheSupersetMetadataExtractor` provides support for extracting basic metadata for Apache Superset dashboards.
All Apache Superset extractors including this one use Apache Superset REST API (`/api/v1`) and were developed based on Apache Superset version `1.1`.
##### Caution!
Apache Superset does not contain metadata fulfilling the concept of `DashboardGroup`. For that reasons, when configuring extractor following parameters must be provided:
- dashboard_group_id (required)
- dashboard_group_name (required)
- cluster (required)
- dashboard_group_description (optional)
#### DashboardMetadata
`ApacheSupersetMetadataExtractor` extracts metadata into `DashboardMetadata` model.
##### Metadata available in REST API
- Dashboard id (id)
- Dashboard name (dashboard_title)
- Dashboard URL (url)
##### Metadata not available in Apache Superset REST API
- Dashboard description
- Dashboard creation timestamp
#### DashboardLastModifiedTimestamp
`ApacheSupersetLastModifiedTimestampExtractor` extracts metadata into `DashboardLastModifiedTimestamp` model.
##### Available in REST API
- Dashboard last modified timestamp (changed_on property of dashboard)
###### Caution!
`changed_on` value does not provide timezone info so we assume it's UTC.
#### Sample job config
```python
tmp_folder = f'/tmp/amundsen/dashboard'
dict_config = {
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.NODE_DIR_PATH}': f'{tmp_folder}/nodes',
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.RELATION_DIR_PATH}': f'{tmp_folder}/relationships',
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.SHOULD_DELETE_CREATED_DIR}': True,
f'extractor.apache_superset.{ApacheSupersetBaseExtractor.DASHBOARD_GROUP_ID}': '1',
f'extractor.apache_superset.{ApacheSupersetBaseExtractor.DASHBOARD_GROUP_NAME}': 'dashboard group',
f'extractor.apache_superset.{ApacheSupersetBaseExtractor.DASHBOARD_GROUP_DESCRIPTION}': 'dashboard group description',
f'extractor.apache_superset.{ApacheSupersetBaseExtractor.CLUSTER}': 'gold',
f'extractor.apache_superset.{ApacheSupersetBaseExtractor.APACHE_SUPERSET_SECURITY_SETTINGS_DICT}': dict(
username='admin',
password='admin',
provider='db')
}
job_config = ConfigFactory.from_dict(dict_config)
task = DefaultTask(extractor=ApacheSupersetMetadataExtractor(), loader=FsNeo4jCSVLoader())
job = DefaultJob(conf=job_config,
task=task)
job.launch()
```
### [ApacheSupersetTableExtractor](./databuilder/extractor/dashboard/apache_superset/apache_superset_table_extractor.py)
The included `ApacheSupersetTableExtractor` provides support for extracting relationships between dashboards and tables. All Apache Superset extractors including this one use Apache Superset REST API (`api/v1`).
##### Caution!
As table information in Apache Superset is minimal, following configuration options enable parametrization required to achieve proper relationship information:
- `driver_to_database_mapping` - mapping between sqlalchemy `drivername` and actual `database` property of `TableMetadata` model.
- `database_to_cluster_mapping` - mapping between Apache Superset Database ID and `cluster` from `TableMedata` model (defaults to `cluster` config of `extractor.apache_superset`)
#### DashboardTable
##### Metadata available in REST API
- Table keys
#### Sample job config
```python
tmp_folder = f'/tmp/amundsen/dashboard'
dict_config = {
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.NODE_DIR_PATH}': f'{tmp_folder}/nodes',
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.RELATION_DIR_PATH}': f'{tmp_folder}/relationships',
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.SHOULD_DELETE_CREATED_DIR}': True,
f'extractor.apache_superset.{ApacheSupersetBaseExtractor.DASHBOARD_GROUP_ID}': '1',
f'extractor.apache_superset.{ApacheSupersetBaseExtractor.DASHBOARD_GROUP_NAME}': 'dashboard group',
f'extractor.apache_superset.{ApacheSupersetBaseExtractor.DASHBOARD_GROUP_DESCRIPTION}': 'dashboard group description',
f'extractor.apache_superset.{ApacheSupersetBaseExtractor.CLUSTER}': 'gold',
f'extractor.apache_superset.{ApacheSupersetBaseExtractor.APACHE_SUPERSET_SECURITY_SETTINGS_DICT}': dict(
username='admin',
password='admin',
provider='db')
}
job_config = ConfigFactory.from_dict(dict_config)
task = DefaultTask(extractor=ApacheSupersetTableExtractor(), loader=FsNeo4jCSVLoader())
job = DefaultJob(conf=job_config,
task=task)
job.launch()
```
### [ApacheSupersetChartExtractor](./databuilder/extractor/dashboard/apache_superset/apache_superset_chart_extractor.py)
The included `ApacheSupersetChartExtractor` provides support for extracting information on charts connected to given dashboard.
##### Caution!
Currently there is no way to connect Apache Superset `Query` model to neither `Chart` nor `Dashboard` model. For that reason, to comply with Amundsen
Databuilder data model, we register single `DashboardQuery` node serving as a bridge to which all the `DashboardChart` nodes are connected.
#### DashboardChart
##### Metadata available in REST API
- Chart id (id)
- Chart name (chart_name)
- Chart type (viz_type)
##### Metadata not available in REST API
- Chart url
#### Sample job config
```python
tmp_folder = f'/tmp/amundsen/dashboard'
dict_config = {
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.NODE_DIR_PATH}': f'{tmp_folder}/nodes',
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.RELATION_DIR_PATH}': f'{tmp_folder}/relationships',
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.SHOULD_DELETE_CREATED_DIR}': True,
f'extractor.apache_superset.{ApacheSupersetBaseExtractor.DASHBOARD_GROUP_ID}': '1',
f'extractor.apache_superset.{ApacheSupersetBaseExtractor.DASHBOARD_GROUP_NAME}': 'dashboard group',
f'extractor.apache_superset.{ApacheSupersetBaseExtractor.DASHBOARD_GROUP_DESCRIPTION}': 'dashboard group description',
f'extractor.apache_superset.{ApacheSupersetBaseExtractor.CLUSTER}': 'gold',
f'extractor.apache_superset.{ApacheSupersetBaseExtractor.APACHE_SUPERSET_SECURITY_SETTINGS_DICT}': dict(
username='admin',
password='admin',
provider='db')
}
job_config = ConfigFactory.from_dict(dict_config)
task = DefaultTask(extractor=ApacheSupersetChartExtractor(), loader=FsNeo4jCSVLoader())
job = DefaultJob(conf=job_config,
task=task)
job.launch()
```
### [PandasProfilingColumnStatsExtractor](./databuilder/extractor/pandas_profiling_column_stats_extractor.py)
[Pandas profiling](https://github.com/pandas-profiling/pandas-profiling) is a library commonly used by Data Engineer and Scientists to calculate advanced data profiles on data.
It is run on pandas dataframe and results in json file containing (amongst other things) descriptive and quantile statistics on columns.
#### Required input parameters
- `FILE_PATH` - file path to pandas-profiling **json** report
- `TABLE_NAME` - name of the table for which report was calculated
- `SCHEMA_NAME` - name of the schema from which table originates
- `DATABASE_NAME` - name of database technology from which table originates
- `CLUSTER_NAME` - name of the cluster from which table originates
#### Optional input parameters
- `PRECISION` - precision for metrics of `float` type. Defaults to `3` meaning up to 3 digits after decimal point.
- `STAT_MAPPINGS` - if you wish to collect only selected set of metrics configure this option with dictionary of following format:
- key - raw name of the stat in pandas-profiling
- value - tuple of 2 elements:
- first value of the tuple - full name of the stat (this influences what will be rendered for user in UI)
- second value of the tuple - function modifying the stat (by default we just do type casting)
Such dictionary should in that case contain only keys of metrics you wish to collect.
For example - if you want only min and max value of a column, provide extractor with configuration option:
```python
PandasProfilingColumnStatsExtractor.STAT_MAPPINGS = {'max': ('Maximum', float), 'min': ('Minimum', float)}
```
Complete set of available metrics is defined as DEFAULT_STAT_MAPPINGS attribute of PandasProfilingColumnStatsExtractor.
#### Common usage patterns
As pandas profiling is executed on top of pandas dataframe, it is up to the user to populate the dataframe before running
the report calculation (and subsequently the extractor). While doing so remember that it might not be a good idea to run the
report on a complete set of rows if your tables are very sparse. In such case it is recommended to dump a subset of rows
to pandas dataframe beforehand and calculate the report on just a sample of original data.
##### Spark support
Support for native execution of pandas-profiling on Spark Dataframe is currently worked on and should come in the future.
#### Sample job config
```python
import pandas as pd
import pandas_profiling
from pyhocon import ConfigFactory
from sqlalchemy import create_engine
from databuilder.extractor.pandas_profiling_column_stats_extractor import PandasProfilingColumnStatsExtractor
from databuilder.job.job import DefaultJob
from databuilder.loader.file_system_neo4j_csv_loader import FsNeo4jCSVLoader
from databuilder.task.task import DefaultTask
table_name = 'video_game_sales'
schema_name = 'superset'
# Load table contents to pandas dataframe
db_uri = f'postgresql://superset:superset@localhost:5432/{schema_name}'
engine = create_engine(db_uri, echo=True)
df = pd.read_sql_table(
table_name,
con=engine
)
# Calculate pandas-profiling report on a table
report_file = '/tmp/table_report.json'
report = df.profile_report(sort=None)
report.to_file(report_file)
# Run PandasProfilingColumnStatsExtractor on calculated report
tmp_folder = f'/tmp/amundsen/column_stats'
dict_config = {
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.NODE_DIR_PATH}': f'{tmp_folder}/nodes',
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.RELATION_DIR_PATH}': f'{tmp_folder}/relationships',
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.SHOULD_DELETE_CREATED_DIR}': False,
'extractor.pandas_profiling.table_name': table_name,
'extractor.pandas_profiling.schema_name': schema_name,
'extractor.pandas_profiling.database_name': 'postgres',
'extractor.pandas_profiling.cluster_name': 'dev',
'extractor.pandas_profiling.file_path': report_file
}
job_config = ConfigFactory.from_dict(dict_config)
task = DefaultTask(extractor=PandasProfilingColumnStatsExtractor(), loader=FsNeo4jCSVLoader())
job = DefaultJob(conf=job_config,
task=task)
job.launch()
```
### [ElasticsearchMetadataExtractor](./databuilder/extractor/es_metadata_extractor.py)
The included `ElasticsearchMetadataExtractor` provides support for extracting basic metadata for Elasticsearch indexes.
It extracts index metadata into `TableMetadata` model so the results are retrievable the same way as table metadata.
Index properties (fields) are treated as `ColumnMetadata`.
#### Technical indexes
This extractor will collect metadata for all indexes of your Elasticsearch instance except for technical indices (which names start with `.`)
#### Configuration
Following configuration options are supported under `extractor.es_metadata` scope:
- `cluster` (required) - name of the cluster of Elasticsearch instance we are extracting metadata from.
- `schema` (required) - name of the schema of Elasticsearch instance we are extracting metadata from.
- `client` (required) - object containing `Elasticsearch` class instance for connecting to Elasticsearch.
- `extract_technical_details` (defaults to `False`) - if `True` index `aliases` and `settings` will be extracted as `Programmatic Descriptions`.
- `correct_sort_order` (defaults to `False`) - if `True` column sort order will match Elasticsearch mapping order.
#### Sample job config
```python3
import os
from elasticsearch import Elasticsearch
from pyhocon import ConfigFactory
from databuilder.extractor.es_metadata_extractor import ElasticsearchMetadataExtractor
from databuilder.job.job import DefaultJob
from databuilder.loader.file_system_neo4j_csv_loader import FsNeo4jCSVLoader
from databuilder.task.task import DefaultTask
tmp_folder = '/tmp/es_metadata'
node_files_folder = f'{tmp_folder}/nodes'
relationship_files_folder = f'{tmp_folder}/relationships'
dict_config = {
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.NODE_DIR_PATH}': node_files_folder,
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.RELATION_DIR_PATH}': relationship_files_folder,
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.SHOULD_DELETE_CREATED_DIR}': True,
f'extractor.es_metadata.{ElasticsearchMetadataExtractor.CLUSTER}': 'demo',
f'extractor.es_metadata.{ElasticsearchMetadataExtractor.SCHEMA}': 'dev',
f'extractor.es_metadata.{ElasticsearchMetadataExtractor.ELASTICSEARCH_CLIENT_CONFIG_KEY}': Elasticsearch()
}
job_config = ConfigFactory.from_dict(dict_config)
task = DefaultTask(extractor=ElasticsearchMetadataExtractor(), loader=FsNeo4jCSVLoader())
job = DefaultJob(conf=job_config,
task=task)
```
### [ElasticsearchColumnStatsExtractor](./databuilder/extractor/es_column_stats_extractor.py)
The included `ElasticsearchColumnStatsExtractor` provides support for extracting basic statistics on numerical properties of Elasticsearch indexes.
It extracts statistics using [Elasticsearch aggregation `matrix_stats`](https://www.elastic.co/guide/en/elasticsearch/reference/master/search-aggregations-matrix-stats-aggregation.html). It disregards statistics named `covariance` and `correlation`.
#### Technical indexes
This extractor will collect metadata for all indexes of your Elasticsearch instance except for technical indices (which names start with `.`)
#### Configuration
Following configuration options are supported under `extractor.es_column_stats` scope:
- `cluster` (required) - name of the cluster of Elasticsearch instance we are extracting metadata from.
- `schema` (required) - name of the schema of Elasticsearch instance we are extracting metadata from.
- `client` (required) - object containing `Elasticsearch` class instance for connecting to Elasticsearch.
- `extract_technical_details` (defaults to `False`) - if `True` index `aliases` and `settings` will be extracted as `Programmatic Descriptions`.
#### Sample job config
```python3
import os
from elasticsearch import Elasticsearch
from pyhocon import ConfigFactory
from databuilder.extractor.es_column_stats_extractor import ElasticsearchColumnStatsExtractor
from databuilder.job.job import DefaultJob
from databuilder.loader.file_system_neo4j_csv_loader import FsNeo4jCSVLoader
from databuilder.task.task import DefaultTask
tmp_folder = '/tmp/es_column_stats'
node_files_folder = f'{tmp_folder}/nodes'
relationship_files_folder = f'{tmp_folder}/relationships'
dict_config = {
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.NODE_DIR_PATH}': node_files_folder,
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.RELATION_DIR_PATH}': relationship_files_folder,
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.SHOULD_DELETE_CREATED_DIR}': True,
f'extractor.es_column_stats.{ElasticsearchColumnStatsExtractor.CLUSTER}': 'demo',
f'extractor.es_column_stats.{ElasticsearchColumnStatsExtractor.SCHEMA}': 'dev',
f'extractor.es_column_stats.{ElasticsearchColumnStatsExtractor.ELASTICSEARCH_CLIENT_CONFIG_KEY}': Elasticsearch()
}
job_config = ConfigFactory.from_dict(dict_config)
task = DefaultTask(extractor=ElasticsearchColumnStatsExtractor(), loader=FsNeo4jCSVLoader())
job = DefaultJob(conf=job_config,
task=task)
```
### [ElasticsearchWatermarkExtractor](./databuilder/extractor/es_watermark_extractor.py)
The included `ElasticsearchWatermarkExtractor` provides support for extracting watermarks for Elasticsearch indexes.
#### Technical indexes
This extractor will collect metadata for all indexes of your Elasticsearch instance except for technical indices (which names start with `.`)
#### Configuration
Following configuration options are supported under `extractor.es_watermark` scope:
- `cluster` (required) - name of the cluster of Elasticsearch instance we are extracting metadata from.
- `schema` (required) - name of the schema of Elasticsearch instance we are extracting metadata from.
- `client` (required) - object containing `Elasticsearch` class instance for connecting to Elasticsearch.
- `time_field` (defaults to `@timestamp`) - name of the field representing time.
#### Sample job config
```python3
from elasticsearch import Elasticsearch
from pyhocon import ConfigFactory
from databuilder.extractor.es_watermark_extractor import ElasticsearchWatermarkExtractor
from databuilder.job.job import DefaultJob
from databuilder.loader.file_system_neo4j_csv_loader import FsNeo4jCSVLoader
from databuilder.task.task import DefaultTask
tmp_folder = '/tmp/es_watermark'
node_files_folder = f'{tmp_folder}/nodes'
relationship_files_folder = f'{tmp_folder}/relationships'
dict_config = {
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.NODE_DIR_PATH}': node_files_folder,
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.RELATION_DIR_PATH}': relationship_files_folder,
f'loader.filesystem_csv_neo4j.{FsNeo4jCSVLoader.SHOULD_DELETE_CREATED_DIR}': True,
f'extractor.es_watermark.{ElasticsearchWatermarkExtractor.CLUSTER}': 'demo',
f'extractor.es_watermark.{ElasticsearchWatermarkExtractor.SCHEMA}': 'dev',
f'extractor.es_watermark.{ElasticsearchWatermarkExtractor.ELASTICSEARCH_TIME_FIELD}': 'time',
f'extractor.es_watermark.{ElasticsearchWatermarkExtractor.ELASTICSEARCH_CLIENT_CONFIG_KEY}': Elasticsearch()
}
job_config = ConfigFactory.from_dict(dict_config)
task = DefaultTask(extractor=ElasticsearchWatermarkExtractor(), loader=FsNeo4jCSVLoader())
job = DefaultJob(conf=job_config,
task=task)
```
### [BamboohrUserExtractor](./databuilder/extractor/user/bamboohr/bamboohr_user_extractor.py)
The included `BamboohrUserExtractor` provides support for extracting basic user metadata from [BambooHR](https://www.bamboohr.com/). For companies and organizations that use BambooHR to store employee information such as email addresses, first names, last names, titles, and departments, use the `BamboohrUserExtractor` to populate Amundsen user data.
A sample job config is shown below.
```python
extractor = BamboohrUserExtractor()
task = DefaultTask(extractor=extractor, loader=FsNeo4jCSVLoader())
job_config = ConfigFactory.from_dict({
'extractor.bamboohr_user.api_key': api_key,
'extractor.bamboohr_user.subdomain': subdomain,
})
job = DefaultJob(conf=job_config,
task=task,
publisher=Neo4jCsvPublisher())
job.launch()
```
### [SalesForceExtractor](./databuilder/extractor/salesforce_extractor.py)
The included `SalesForceExtractor` provides support for extracting basic SalesForce object metadata
from [SalesForce](https://developer.salesforce.com/).
This extractor depends on the Python client [simple-salesforce](https://github.com/simple-salesforce/simple-salesforce).
A sample job config is shown below. Some notes about the configuration keys.
This extractor currently only supports connecting to SalesForce with a username, password, and security token.
You pass these values in as configuration keys.
The extractor will by default pull all SalesForce metadata objects which is likely not what you want. To only pull
specific SalesForce metadata objects specify their names with the `SalesForceExtractor.OBJECT_NAMES_KEY`.
There is no real notion of a schema for the SalesForce metadata objects but you still need to specify one.
```python
from databuilder.extractor.salesforce_extractor import SalesForceExtractor
extractor = SalesForceExtractor()
task = SalesForceExtractor(extractor=extractor, loader=FsNeo4jCSVLoader())
job_config = ConfigFactory.from_dict({
f"extractor.salesforce_metadata.{SalesForceExtractor.USERNAME_KEY}": "user",
f"extractor.salesforce_metadata.{SalesForceExtractor.PASSWORD_KEY}": "password",
f"extractor.salesforce_metadata.{SalesForceExtractor.SECURITY_TOKEN_KEY}": "token",
f"extractor.salesforce_metadata.{SalesForceExtractor.SCHEMA_KEY}": "default",
f"extractor.salesforce_metadata.{SalesForceExtractor.CLUSTER_KEY}": "gold",
f"extractor.salesforce_metadata.{SalesForceExtractor.DATABASE_KEY}": "salesforce",
f"extractor.salesforce_metadata.{SalesForceExtractor.OBJECT_NAMES_KEY}": ["Account", "Profile"]
})
job = DefaultJob(conf=job_config,
task=task,
publisher=Neo4jCsvPublisher())
job.launch()
```
#### [EventBridgeExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/eventbridge_extractor.py "EventBridgeExtractor")
An extractor that extracts schema metadata from AWS EventBridge schema registries.
A sample job config is shown below.
```python
job_config = ConfigFactory.from_dict({
f"extractor.eventbridge.{EventBridgeExtractor.REGION_NAME_KEY}": "aws_region",
f"extractor.eventbridge.{EventBridgeExtractor.REGISTRY_NAME_KEY}": "eventbridge_schema_registry_name",
})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=EventBridgeExtractor(),
loader=AnyLoader()))
job.launch()
```
#### [KafkaSchemaRegistryExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/kafka_schema_registry_extractor.py "KafkaSchemaRegistryExtractor")
An extractor that extracts schema metadata Confluent Kafka Schema registry with Avro format.
A sample job config is shown below.
```python
job_config = ConfigFactory.from_dict({
f"extractor.kafka_schema_registry.{KafkaSchemaRegistryExtractor.REGISTRY_URL_KEY}": "http://localhost:8081",
f"extractor.kafka_schema_registry.{KafkaSchemaRegistryExtractor.REGISTRY_USERNAME_KEY}": "username",
f"extractor.kafka_schema_registry.{KafkaSchemaRegistryExtractor.REGISTRY_PASSWORD_KEY}": "password",
})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=KafkaSchemaRegistryExtractor(),
loader=AnyLoader()))
job.launch()
```
**Note: username and password are not mandatory. Only provide if you schema registry need authorization.**
## List of transformers
Transformers are implemented by subclassing [Transformer](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/transformer/base_transformer.py#L12 "Transformer") and implementing `transform(self, record)`. A transformer can:
- Modify a record and return it,
- Return `None` to filter a record out,
- Yield multiple records. This is useful for e.g. inferring metadata (such as ownership) from table descriptions.
#### [ChainedTransformer](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/transformer/base_transformer.py#L41 "ChainedTransformer")
A chanined transformer that can take multiple transformers, passing each record through the chain.
#### [RegexStrReplaceTransformer](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/transformer/regex_str_replace_transformer.py "RegexStrReplaceTransformer")
Generic string replacement transformer using REGEX. User can pass list of tuples where tuple contains regex and replacement pair.
```python
job_config = ConfigFactory.from_dict({
'transformer.regex_str_replace.{}'.format(REGEX_REPLACE_TUPLE_LIST): [(',', ' '), ('"', '')],
'transformer.regex_str_replace.{}'.format(ATTRIBUTE_NAME): 'instance_field_name',})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=AnyExtractor(),
transformer=RegexStrReplaceTransformer(),
loader=AnyLoader()))
job.launch()
```
#### [TemplateVariableSubstitutionTransformer](./databuilder/transformer/template_variable_substitution_transformer.py)
Adds or replaces field in Dict by string.format based on given template and provide record Dict as a template parameter.
#### [DictToModel](./databuilder/transformer/dict_to_model.py)
Transforms dictionary into model.
#### [TimestampStringToEpoch](./databuilder/transformer/timestamp_string_to_epoch.py)
Transforms string timestamp into int epoch.
#### [RemoveFieldTransformer](./databuilder/transformer/remove_field_transformer.py)
Remove fields from the Dict.
#### [TableTagTransformer](./databuilder/transformer/table_tag_transformer.py)
Adds the same set of tags to all tables produced by the job.
#### [GenericTransformer](./databuilder/transformer/generic_transformer.py)
Transforms dictionary based on callback function that user provides.
#### [ComplexTypeTransformer](./databuilder/transformer/complex_type_transformer.py)
Transforms complex types for columns in a table by using a configured parsing function. The transformer takes a `TableMetadata` object and iterates over its list of `ColumnMetadata` objects. The configured parser takes each column as input and sets the column's `type_metadata` field with the parsed results contained in a `TypeMetadata` object.
**If you use Hive as a data store:**
Configure this transformer with the [Hive parser](./databuilder/utils/hive_complex_type_parser.py).
**If you do not use Hive as a data store:**
You will need to write a custom parsing function for transforming column type strings into nested `TypeMetadata` objects. You are free to use the [Hive parser](./databuilder/utils/hive_complex_type_parser.py) as a starting point. You can also look online to try to find either a grammar or some OSS prior art, as writing a parser from scratch can get a little involved. We strongly recommend leveraging PyParsing instead of regex, etc.
New parsing functions should take the following arguments:
- Column type string
- Column name
- `ColumnMetadata` object itself
Within the parsing function, [TypeMetadata](./databuilder/models/type_metadata.py) objects should be created by passing its name, parent object, and type string.
**Things to know about [TypeMetadata](./databuilder/models/type_metadata.py)**
- If the existing subclasses do not cover all the required complex types, the base class can be extended to create any new ones that are needed.
- Each new subclass should implement a `is_terminal_type` function, which allows the node and relation iterators to check whether to continue creating the next nested level or to stop due to reaching a terminal node.
- `ScalarTypeMetadata` is the default type class that represents a terminal state. This should be used to set any column's `type_metadata` when it is not a complex type, or for the innermost terminal state for any complex type. Having all the columns set the `type_metadata` field allows the frontend to know to use the correct nested column display.
- Subclasses should set a `kind` field that specifies what kind of complex type they are. This is used by the frontend for specific type handling. For example, for arrays and maps a smaller row is inserted in the display table to differentiate them from named nested columns such as structs.
## List of loader
#### [FsNeo4jCSVLoader](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/loader/file_system_neo4j_csv_loader.py "FsNeo4jCSVLoader")
Write node and relationship CSV file(s) that can be consumed by Neo4jCsvPublisher. It assumes that the record it consumes is instance of Neo4jCsvSerializable.
```python
job_config = ConfigFactory.from_dict({
'loader.filesystem_csv_neo4j.{}'.format(FsNeo4jCSVLoader.NODE_DIR_PATH): node_files_folder,
'loader.filesystem_csv_neo4j.{}'.format(FsNeo4jCSVLoader.RELATION_DIR_PATH): relationship_files_folder},)
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=AnyExtractor(),
loader=FsNeo4jCSVLoader()),
publisher=Neo4jCsvPublisher())
job.launch()
```
#### [GenericLoader](./databuilder/loader/generic_loader.py)
Loader class that calls user provided callback function with record as a parameter
Example that pushes Mode Dashboard accumulated usage via GenericLoader where callback_function expected to insert record to data warehouse.
```python
extractor = ModeDashboardUsageExtractor()
task = DefaultTask(extractor=extractor,
loader=GenericLoader(), )
job_config = ConfigFactory.from_dict({
'{}.{}'.format(extractor.get_scope(), ORGANIZATION): organization,
'{}.{}'.format(extractor.get_scope(), MODE_BEARER_TOKEN): mode_bearer_token,
'loader.generic.callback_function': callback_function
})
job = DefaultJob(conf=job_config, task=task)
job.launch()
```
#### [FSElasticsearchJSONLoader](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/loader/file_system_elasticsearch_json_loader.py "FSElasticsearchJSONLoader")
Write Elasticsearch document in JSON format which can be consumed by ElasticsearchPublisher. It assumes that the record it consumes is instance of ElasticsearchDocument.
```python
data_file_path = '/var/tmp/amundsen/search_data.json'
job_config = ConfigFactory.from_dict({
'loader.filesystem.elasticsearch.{}'.format(FSElasticsearchJSONLoader.FILE_PATH_CONFIG_KEY): data_file_path,
'loader.filesystem.elasticsearch.{}'.format(FSElasticsearchJSONLoader.FILE_MODE_CONFIG_KEY): 'w',})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=AnyExtractor(),
loader=FSElasticsearchJSONLoader()),
publisher=ElasticsearchPublisher())
job.launch()
```
#### [FsAtlasCSVLoader](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/loader/file_system_atlas_csv_loader.py "FileSystemCSVLoader")
Write node and relationship CSV file(s) that can be consumed by AtlasCsvPublisher. It assumes that the record it
consumes is instance of AtlasSerializable.
```python
from pyhocon import ConfigFactory
from databuilder.job.job import DefaultJob
from databuilder.loader.file_system_atlas_csv_loader import FsAtlasCSVLoader
from databuilder.task.task import DefaultTask
tmp_folder = f'/tmp/amundsen/dashboard'
job_config = ConfigFactory.from_dict({
f'loader.filesystem_csv_atlas.{FsAtlasCSVLoader.ENTITY_DIR_PATH}': f'{tmp_folder}/entities',
f'loader.filesystem_csv_atlas.{FsAtlasCSVLoader.RELATIONSHIP_DIR_PATH}': f'{tmp_folder}/relationships'
})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=AnyExtractor(),
loader=FsAtlasCSVLoader()),
publisher=AnyPublisher())
job.launch()
```
## List of publisher
#### [Neo4jCsvPublisher](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/publisher/neo4j_csv_publisher.py "Neo4jCsvPublisher")
A Publisher takes two folders for input and publishes to Neo4j.
One folder will contain CSV file(s) for Node where the other folder will contain CSV file(s) for Relationship. Neo4j follows Label Node properties Graph and refer to [here](https://neo4j.com/docs/developer-manual/current/introduction/graphdb-concepts/ "here") for more information
```python
node_files_folder = '{tmp_folder}/nodes/'.format(tmp_folder=tmp_folder)
relationship_files_folder = '{tmp_folder}/relationships/'.format(tmp_folder=tmp_folder)
job_config = ConfigFactory.from_dict({
'loader.filesystem_csv_neo4j.{}'.format(FsNeo4jCSVLoader.NODE_DIR_PATH): node_files_folder,
'loader.filesystem_csv_neo4j.{}'.format(FsNeo4jCSVLoader.RELATION_DIR_PATH): relationship_files_folder,
'publisher.neo4j.{}'.format(neo4j_csv_publisher.NODE_FILES_DIR): node_files_folder,
'publisher.neo4j.{}'.format(neo4j_csv_publisher.RELATION_FILES_DIR): relationship_files_folder,
'publisher.neo4j.{}'.format(neo4j_csv_publisher.NEO4J_END_POINT_KEY): neo4j_endpoint,
'publisher.neo4j.{}'.format(neo4j_csv_publisher.NEO4J_USER): neo4j_user,
'publisher.neo4j.{}'.format(neo4j_csv_publisher.NEO4J_PASSWORD): neo4j_password,
'publisher.neo4j.{}'.format(neo4j_csv_publisher.NEO4J_ENCRYPTED): True,
})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=AnyExtractor(),
loader=FsNeo4jCSVLoader()),
publisher=Neo4jCsvPublisher())
job.launch()
```
#### [ElasticsearchPublisher](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/publisher/elasticsearch_publisher.py "ElasticsearchPublisher")
Elasticsearch Publisher uses Bulk API to load data from JSON file. Elasticsearch publisher supports atomic operation by utilizing alias in Elasticsearch.
A new index is created and data is uploaded into it. After the upload is complete, index alias is swapped to point to new index from old index and traffic is routed to new index.
```python
data_file_path = '/var/tmp/amundsen/search_data.json'
job_config = ConfigFactory.from_dict({
'loader.filesystem.elasticsearch.{}'.format(FSElasticsearchJSONLoader.FILE_PATH_CONFIG_KEY): data_file_path,
'loader.filesystem.elasticsearch.{}'.format(FSElasticsearchJSONLoader.FILE_MODE_CONFIG_KEY): 'w',
'publisher.elasticsearch.{}'.format(ElasticsearchPublisher.FILE_PATH_CONFIG_KEY): data_file_path,
'publisher.elasticsearch.{}'.format(ElasticsearchPublisher.FILE_MODE_CONFIG_KEY): 'r',
'publisher.elasticsearch{}'.format(ElasticsearchPublisher.ELASTICSEARCH_CLIENT_CONFIG_KEY): elasticsearch_client,
'publisher.elasticsearch.{}'.format(ElasticsearchPublisher.ELASTICSEARCH_NEW_INDEX_CONFIG_KEY): elasticsearch_new_index,
'publisher.elasticsearch.{}'.format(ElasticsearchPublisher.ELASTICSEARCH_DOC_TYPE_CONFIG_KEY): elasticsearch_doc_type,
'publisher.elasticsearch.{}'.format(ElasticsearchPublisher.ELASTICSEARCH_ALIAS_CONFIG_KEY): elasticsearch_index_alias,
})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=AnyExtractor(),
loader=FSElasticsearchJSONLoader()),
publisher=ElasticsearchPublisher())
job.launch()
```
#### [AtlasCsvPublisher](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/publisher/atlas_csv_publisher.py "AtlasCsvPublisher")
A Publisher takes two folders for input and publishes to Atlas.
One folder will contain CSV file(s) for Entity where the other folder will contain CSV file(s) for Relationship.
##### Amundsen <> Atlas Types
Atlas publisher requires registering appropriate entity types in Atlas. This can be achieved in two ways:
###### Register entity types directly in publisher
By default publisher will register proper entity types for you. This is achieved with `register_entity_types` configuration option of the publisher, which defaults to `True`.
If your metadata synchronization job consists of several extractors leveraging `AtlasCSVPublisher` it is recommended to have this option turned on only for the first extractor.
###### Register entity types using standalone script
You can register entity types separately - below script might serve as a baseline and will probably need adjusting AtlasClient to your environment.
```python3
from apache_atlas.client.base_client import AtlasClient
from databuilder.types.atlas import AtlasEntityInitializer
client = AtlasClient('http://localhost:21000', ('admin', 'admin'))
init = AtlasEntityInitializer(client)
init.create_required_entities()
```
###### Caution!
Whenever you upgrade your databuilder version it is important to re-run `AtlasEntityInitializer` as there might be new changes to entity types required for Atlas integration to work properly.
##### Sample script
```python
from apache_atlas.client.base_client import AtlasClient
from pyhocon import ConfigFactory
from databuilder.job.job import DefaultJob
from databuilder.loader.file_system_atlas_csv_loader import FsAtlasCSVLoader
from databuilder.publisher.atlas_csv_publisher import AtlasCSVPublisher
from databuilder.task.task import DefaultTask
tmp_folder = f'/tmp/amundsen/dashboard'
job_config = ConfigFactory.from_dict({
f'loader.filesystem_csv_atlas.{FsAtlasCSVLoader.ENTITY_DIR_PATH}': f'{tmp_folder}/entities',
f'loader.filesystem_csv_atlas.{FsAtlasCSVLoader.RELATIONSHIP_DIR_PATH}': f'{tmp_folder}/relationships',
f'publisher.atlas_csv_publisher.{AtlasCSVPublisher.ATLAS_CLIENT}': AtlasClient('http://localhost:21000', ('admin', 'admin')) ,
f'publisher.atlas_csv_publisher.{AtlasCSVPublisher.ENTITY_DIR_PATH}': f'{tmp_folder}/entities',
f'publisher.atlas_csv_publisher.{AtlasCSVPublisher.RELATIONSHIP_DIR_PATH}': f'{tmp_folder}/relationships',
f'publisher.atlas_csv_publisher.{AtlasCSVPublisher.ATLAS_ENTITY_CREATE_BATCH_SIZE}': 10,
f'publisher.atlas_csv_publisher.{AtlasCSVPublisher.REGISTER_ENTITY_TYPES}': True
})
job = DefaultJob(
conf=job_config,
task=DefaultTask(
extractor=AnyExtractor(),
loader=FsAtlasCSVLoader()),
publisher=AtlasCSVPublisher())
job.launch()
```
#### [Callback](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/callback/call_back.py "Callback")
Callback interface is built upon a [Observer pattern](https://en.wikipedia.org/wiki/Observer_pattern "Observer pattern") where the participant want to take any action when target's state changes.
Publisher is the first one adopting Callback where registered Callback will be called either when publish succeeded or when publish failed. In order to register callback, Publisher provides [register_call_back](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/publisher/base_publisher.py#L50 "register_call_back") method.
One use case is for Extractor that needs to commit when job is finished (e.g: Kafka). Having Extractor register a callback to Publisher to commit when publish is successful, extractor can safely commit by implementing commit logic into [on_success](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/callback/call_back.py#L18 "on_success") method.
### REST API Query
Databuilder now has a generic REST API Query capability that can be joined each other.
Most of the cases of extraction is currently from Database or Datawarehouse that is queryable via SQL. However, not all metadata sources provide our access to its Database and they mostly provide REST API to consume their metadata.
The challenges come with REST API is that:
1. there's no explicit standard in REST API. Here, we need to conform to majority of cases (HTTP call with JSON payload & response) but open for extension for different authentication scheme, and different way of pagination, etc.
2. It is hardly the case that you would get what you want from one REST API call. It is usually the case that you need to snitch (JOIN) multiple REST API calls together to get the information you want.
To solve this challenges, we introduce [RestApiQuery](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/rest_api/rest_api_query.py)
RestAPIQuery is:
1. Assuming that REST API is using HTTP(S) call with GET method -- RestAPIQuery intention's is **read**, not write -- where basic HTTP auth is supported out of the box. There's extension point on other authentication scheme such as Oauth, and pagination, etc. (See [ModePaginatedRestApiQuery](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/rest_api/mode_analytics/mode_paginated_rest_api_query.py) for pagination)
2. Usually, you want the subset of the response you get from the REST API call -- value extraction. To extract the value you want, RestApiQuery uses [JSONPath](https://goessner.net/articles/JsonPath/) which is similar product as XPATH of XML.
3. You can JOIN multiple RestApiQuery together.
More detail on JOIN operation in RestApiQuery:
1. It joins multiple RestApiQuery together by accepting prior RestApiQuery as a constructor -- a [Decorator pattern](https://en.wikipedia.org/wiki/Decorator_pattern)
2. In REST API, URL is the one that locates the resource we want. Here, JOIN simply means we need to find resource **based on the identifier that other query's result has**. In other words, when RestApiQuery forms URL, it uses previous query's result to compute the URL `e.g: Previous record: {"dashboard_id": "foo"}, URL before: http://foo.bar/dashboard/{dashboard_id} URL after compute: http://foo.bar/dashboard/foo`
With this pattern RestApiQuery supports 1:1 and 1:N JOIN relationship.
(GROUP BY or any other aggregation, sub-query join is not supported)
To see in action, take a peek at [ModeDashboardExtractor](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/dashboard/mode_analytics/mode_dashboard_extractor.py)
Also, take a look at how it extends to support pagination at [ModePaginatedRestApiQuery](./databuilder/rest_api/mode_analytics/mode_paginated_rest_api_query.py).
### Removing stale data in Neo4j -- [Neo4jStalenessRemovalTask](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/task/neo4j_staleness_removal_task.py):
As Databuilder ingestion mostly consists of either INSERT OR UPDATE, there could be some stale data that has been removed from metadata source but still remains in Neo4j database. Neo4jStalenessRemovalTask basically detects staleness and removes it.
In [Neo4jCsvPublisher](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/publisher/neo4j_csv_publisher.py), it adds attributes "published_tag" and "publisher_last_updated_epoch_ms" on every nodes and relations. You can use either of these two attributes to detect staleness and remove those stale node or relation from the database.
NOTE: data can exist without either attributes "published_tag" or "publisher_last_updated_epoch_ms" if it is created by an Amundsen user rather than by the publisher. In this case you may not want to have these nodes marked as stale and deleted. To keep these nodes, you can set a configured value `retain_data_with_no_publisher_metadata` to `True`:
task = Neo4jStalenessRemovalTask()
job_config_dict = {
'job.identifier': 'remove_stale_data_job',
'task.remove_stale_data.neo4j_endpoint': neo4j_endpoint,
'task.remove_stale_data.neo4j_user': neo4j_user,
'task.remove_stale_data.neo4j_password': neo4j_password,
'task.remove_stale_data.staleness_max_pct': 10,
'task.remove_stale_data.target_nodes': ['Table', 'Column'],
'task.remove_stale_data.job_publish_tag': '2020-03-31',
'task.remove_stale_data.retain_data_with_no_publisher_metadata': True
}
job_config = ConfigFactory.from_dict(job_config_dict)
job = DefaultJob(conf=job_config, task=task)
job.launch()
#### Using "published_tag" to remove stale data
Use *published_tag* to remove stale data, when it is certain that non-matching tag is stale once all the ingestion is completed. For example, suppose that you use current date (or execution date in Airflow) as a *published_tag*, "2020-03-31". Once Databuilder ingests all tables and all columns, all table nodes and column nodes should have *published_tag* as "2020-03-31". It is safe to assume that table nodes and column nodes whose *published_tag* is different -- such as "2020-03-30" or "2020-02-10" -- means that it is deleted from the source metadata. You can use Neo4jStalenessRemovalTask to delete those stale data.
task = Neo4jStalenessRemovalTask()
job_config_dict = {
'job.identifier': 'remove_stale_data_job',
'task.remove_stale_data.neo4j_endpoint': neo4j_endpoint,
'task.remove_stale_data.neo4j_user': neo4j_user,
'task.remove_stale_data.neo4j_password': neo4j_password,
'task.remove_stale_data.staleness_max_pct': 10,
'task.remove_stale_data.target_nodes': ['Table', 'Column'],
'task.remove_stale_data.job_publish_tag': '2020-03-31'
}
job_config = ConfigFactory.from_dict(job_config_dict)
job = DefaultJob(conf=job_config, task=task)
job.launch()
Note that there's protection mechanism, **staleness_max_pct**, that protect your data being wiped out when something is clearly wrong. "**staleness_max_pct**" basically first measure the proportion of elements that will be deleted and if it exceeds threshold per type ( 10% on the configuration above ), the deletion won't be executed and the task aborts.
#### Using "publisher_last_updated_epoch_ms" to remove stale data
You can think this approach as TTL based eviction. This is particularly useful when there are multiple ingestion pipelines and you cannot be sure when all ingestion is done. In this case, you might still can say that if specific node or relation has not been published past 3 days, it's stale data.
task = Neo4jStalenessRemovalTask()
job_config_dict = {
'job.identifier': 'remove_stale_data_job',
'task.remove_stale_data.neo4j_endpoint': neo4j_endpoint,
'task.remove_stale_data.neo4j_user': neo4j_user,
'task.remove_stale_data.neo4j_password': neo4j_password,
'task.remove_stale_data.staleness_max_pct': 10,
'task.remove_stale_data.target_relations': ['READ', 'READ_BY'],
'task.remove_stale_data.milliseconds_to_expire': 86400000 * 3
}
job_config = ConfigFactory.from_dict(job_config_dict)
job = DefaultJob(conf=job_config, task=task)
job.launch()
Above configuration is trying to delete stale usage relation (READ, READ_BY), by deleting READ or READ_BY relation that has not been published past 3 days. If number of elements to be removed is more than 10% per type, this task will be aborted without executing any deletion.
#### Using node and relation conditions to remove stale data
You may want to remove stale nodes and relations that meet certain conditions rather than all of a given type. To do this, you can specify the inputs to be a list of **TargetWithCondition** objects that each define a target type and a condition. Only stale nodes or relations of that type and that meet the condition will be removed when using this type of input.
Node conditions can make use of the predefined variable `target` which represents the node. Relation conditions can include the variables `target`, `start_node`, and `end_node` where `target` represents the relation and `start_node`/`end_node` represent the nodes on either side of the target relation. For some examples of conditions see below.
from databuilder.task.neo4j_staleness_removal_task import TargetWithCondition
task = Neo4jStalenessRemovalTask()
job_config_dict = {
'job.identifier': 'remove_stale_data_job',
'task.remove_stale_data.neo4j_endpoint': neo4j_endpoint,
'task.remove_stale_data.neo4j_user': neo4j_user,
'task.remove_stale_data.neo4j_password': neo4j_password,
'task.remove_stale_data.staleness_max_pct': 10,
'task.remove_stale_data.target_nodes': [TargetWithCondition('Table', '(target)-[:COLUMN]->(:Column)'), # All Table nodes that have a directional COLUMN relation to a Column node
TargetWithCondition('Column', '(target)-[]-(:Table) AND target.name=\'column_name\'')], # All Column nodes named 'column_name' that have some relation to a Table node
'task.remove_stale_data.target_relations': [TargetWithCondition('COLUMN', '(start_node:Table)-[target]->(end_node:Column)'), # All COLUMN relations that connect from a Table node to a Column node
TargetWithCondition('COLUMN', '(start_node:Column)-[target]-(end_node)')], # All COLUMN relations that connect any direction between a Column node and another node
'task.remove_stale_data.milliseconds_to_expire': 86400000 * 3
}
job_config = ConfigFactory.from_dict(job_config_dict)
job = DefaultJob(conf=job_config, task=task)
job.launch()
You can include multiple inputs of the same type with different conditions as seen in the **target_relations** list above. Attribute checks can also be added as shown in the **target_nodes** list.
#### Dry run
Deletion is always scary and it's better to perform dryrun before put this into action. You can use Dry run to see what sort of Cypher query will be executed.
task = Neo4jStalenessRemovalTask()
job_config_dict = {
'job.identifier': 'remove_stale_data_job',
'task.remove_stale_data.neo4j_endpoint': neo4j_endpoint,
'task.remove_stale_data.neo4j_user': neo4j_user,
'task.remove_stale_data.neo4j_password': neo4j_password,
'task.remove_stale_data.staleness_max_pct': 10,
'task.remove_stale_data.target_relations': ['READ', 'READ_BY'],
'task.remove_stale_data.milliseconds_to_expire': 86400000 * 3
'task.remove_stale_data.dry_run': True
}
job_config = ConfigFactory.from_dict(job_config_dict)
job = DefaultJob(conf=job_config, task=task)
job.launch()
---
## File: docs/authentication/oidc.md
# OIDC Authentication
Setting up end-to-end authentication using OIDC is fairly simple and can be done using a Flask wrapper i.e., [flaskoidc](https://github.com/verdan/flaskoidc).
`flaskoidc` leverages the Flask's `before_request` functionality to authenticate each request before passing that to
the views. It also accepts headers on each request if available in order to validate bearer token from incoming requests.
## Installation
_(If you are using flaskoidc<1.0.0, please follow the documentation [here](https://github.com/verdan/flaskoidc/tree/master#readme)_
**PREREQUISITE**: Please refer to the [flaskoidc Documentation](https://github.com/verdan/flaskoidc#readme)
for the installation and the configurations.
Note: You need to install and configure `flaskoidc` for each microservice of Amundsen
i.e., for frontendlibrary, metadatalibrary and searchlibrary in order to secure each of them.
## Amundsen Configuration
Once you have `flaskoidc` installed and configured for each microservice, please set the following environment variables:
- amundsenfrontendlibrary (`amundsen/frontend`):
```bash
FLASK_APP_MODULE_NAME: flaskoidc
FLASK_APP_CLASS_NAME: FlaskOIDC
```
- amundsenmetadatalibrary (`amundsen/metadata`):
```bash
FLASK_APP_MODULE_NAME: flaskoidc
FLASK_APP_CLASS_NAME: FlaskOIDC
```
- amundsensearchlibrary (`amundsen/search`):
```bash
FLASK_APP_MODULE_NAME: flaskoidc
FLASK_APP_CLASS_NAME: FlaskOIDC
```
By default `flaskoidc` whitelist the healthcheck URLs, to not authenticate them. In case of metadatalibrary and searchlibrary
we may want to whitelist the healthcheck APIs explicitly using following environment variable.
```bash
FLASK_OIDC_WHITELISTED_ENDPOINTS: 'api.healthcheck'
```
## Setting Up Request Headers
To communicate securely between the microservices, you need to pass the bearer token from frontend in each request
to metadatalibrary and searchlibrary. This should be done using `REQUEST_HEADERS_METHOD` config variable in frontendlibrary.
- Define a function to add the bearer token in each request in your config.py:
*version: flaskoidc<1.0.0*
```python
def get_access_headers(app):
try:
access_token = app.oidc.get_access_token()
return {'Authorization': 'Bearer {}'.format(access_token)}
except Exception:
return None
```
*version: flaskoidc>=1.0.0*
```python
from flask import Flask
def get_access_headers(app: Flask) -> Optional[Dict]:
try:
# noinspection PyUnresolvedReferences
access_token = json.dumps(app.auth_client.token)
return {'Authorization': 'Bearer {}'.format(access_token)}
except Exception:
pass
```
- Set the method as the request header method in your config.py:
```python
REQUEST_HEADERS_METHOD = get_access_headers
```
This function will be called using the current `app` instance to add the headers in each request when calling any endpoint of
metadatalibrary and searchlibrary [here](/frontend/amundsen_application/api/utils/request_utils.py)
## Setting Up Auth User Method
In order to get the current authenticated user (which is being used in Amundsen for many operations), we need to set
`AUTH_USER_METHOD` config variable in frontendlibrary.
This function should return email address, user id and any other required information.
- Define a function to fetch the user information in your config.py:
*version: flaskoidc<1.0.0*
```python
from flask import Flask
from amundsen_application.models.user import load_user, User
def get_auth_user(app: Flask) -> User:
from flask import g
user_info = load_user(g.oidc_id_token)
return user_info
```
*version: flaskoidc>=1.0.0*
```python
from flask import Flask, session
from amundsen_application.models.user import load_user, User
def get_auth_user(app: Flask) -> User:
user_info = load_user(session.get("user"))
return user_info
```
- Set the method as the auth user method in your config.py:
```python
AUTH_USER_METHOD = get_auth_user
```
Once done, you'll have the end-to-end authentication in Amundsen without any proxy or code changes.
## Using Okta with Amundsen on K8s
_Valid for flaskoidc<1.0.0_
Assumptions:
- You have access to okta (you can create a developer account for free!)
- You are using k8s to setup amundsen. See [amundsen-kube-helm](../../amundsen-kube-helm/README.md)
1. You need to have a stable DNS entry for amundsen-frontend that can be registered in okta.
- for example in AWS you can setup route53
I will assume for the rest of this tutorial that your stable uri is "http://amundsen-frontend"
2. You need to register amundsen in okta as an app. More info [here](https://developer.okta.com/blog/2018/07/12/flask-tutorial-simple-user-registration-and-login).
But here are specific instructions for amundsen:
- At this time, I have only succesfully tested integration after ALL grants were checked.
- Set the Login redirect URIs to: `http://amundsen-frontend/oidc_callback`
- No need to set a logout redirect URI
- Set the Initiate login URI to: `http://amundsen-frontend/`
(This is where okta will take you if users click on amundsen via okta landing page)
- Copy the Client ID and Client secret as you will need this later.
3. At present, there is no oidc build of the frontend. So you will need to build an oidc build yourself and upload it to, for example ECR, for use by k8s.
You can then specify which image you want to use as a property override for your helm install like so:
```yaml
frontEndServiceImage: 123.dkr.ecr.us-west-2.amazonaws.com/edmunds/amundsen-frontend:oidc-test
```
Please see further down in this doc for more instructions on how to build frontend.
4. When you start up helm you will need to provide some properties. Here are the properties that need to be overridden for oidc to work:
```yaml
oidcEnabled: true
createOidcSecret: true
OIDC_CLIENT_ID: YOUR_CLIENT_ID
OIDC_CLIENT_SECRET: YOUR_SECRET_ID
OIDC_ORG_URL: https://amundsen.okta.com
OIDC_AUTH_SERVER_ID: default
# You also will need a custom oidc frontend build too
frontEndServiceImage: 123.dkr.ecr.us-west-2.amazonaws.com/edmunds/amundsen-frontend:oidc-test
```
## Building frontend with OIDC
1. Please look at [this guide](../developer_guide.md) for instructions on how to build a custom frontend docker image.
2. The only difference to above is that in your docker file you will want to add the following at the end. This will make sure its ready to go for oidc.
You can take alook at the public.Dockerfile as a reference.
```dockerfile
RUN pip3 install .[oidc]
ENV FRONTEND_SVC_CONFIG_MODULE_CLASS=amundsen_application.oidc_config.OidcConfig
ENV FLASK_APP_MODULE_NAME=flaskoidc
ENV FLASK_APP_CLASS_NAME=FlaskOIDC
ENV FLASK_OIDC_WHITELISTED_ENDPOINTS=status,healthcheck,health
ENV SQLALCHEMY_DATABASE_URI=sqlite:///sessions.db
```
Please also take a look at this blog [post](https://nirav-langaliya.medium.com/setup-oidc-authentication-with-lyft-amundsen-via-okta-eb0b89d724d3) for more detail.
---
## File: docs/installation-aws-ecs/aws-ecs-deployment.md
# Deployment of non-production Amundsen on AWS ECS using aws-cli
The following is a set of instructions to run Amundsen on AWS Elastic Container Service. The current configuration is very basic but it is working. It is a migration of the docker-amundsen.yml to run on AWS ECS.
## Install ECS CLI
The first step is to install ECS CLI, please follow the instructions from AWS [documentation](https://docs.aws.amazon.com/AmazonECS/latest/developerguide/ECS_CLI_installation.html)
### Get your access and secret keys from IAM
```bash
# in ~//amundsenfrontendlibrary/docs/instalation-aws-ecs
$ export AWS_ACCESS_KEY_ID=xxxxxxxx
$ export AWS_SECRET_ACCESS_KEY=xxxxxx
$ export AWS_PROFILE=profilename
```
For the purpose of this instruction we used the [tutorial](https://docs.aws.amazon.com/AmazonECS/latest/developerguide/ecs-cli-tutorial-ec2.html#ECS_CLI_tutorial_compose_create) on AWS documentation
Enter the cloned directory:
```
cd amundsen/docs/installation-aws-ecs
```
## STEP 1: Create a cluster configuration:
```bash
# in ~//amundsenfrontendlibrary/docs/instalation-aws-ecs
$ ecs-cli configure --cluster amundsen --region us-west-2 --default-launch-type EC2 --config-name amundsen
```
### STEP 2: Create a profile using your access key and secret key:
```bash
# in ~//amundsen/docs/installation-aws-ecs
$ ecs-cli configure profile --access-key $AWS_ACCESS_KEY_ID --secret-key $AWS_SECRET_ACCESS_KEY --profile-name amundsen
```
### STEP 3: Create the Cluster Use profile name from \~/.aws/credentials
```bash
# in ~//amundsen/docs/installation-aws-ecs
$ ecs-cli up --keypair JoaoCorreia --extra-user-data userData.sh --capability-iam --size 1 --instance-type t2.large --cluster-config amundsen --verbose --force --aws-profile $AWS_PROFILE
```
### STEP 4: Deploy the Compose File to a Cluster
```bash
# in ~//amundsen/docs/installation-aws-ecs
$ ecs-cli compose --cluster-config amundsen --file docker-ecs-amundsen.yml up --create-log-groups
```
You can use the ECS CLI to see what tasks are running.
```bash
$ ecs-cli ps
```
### STEP 5 Open the EC2 Instance
Edit the Security Group to allow traffic to your IP, you should be able to see the frontend, elasticsearch and neo4j by visiting the URLs:
- http://xxxxxxx:5000/
- http://xxxxxxx:9200/
- http://xxxxxxx:7474/browser/
## TODO
- Configuration sent to services not working properly (amunsen.db vs graph.db)
- Create a persistent volume for graph/metadata storage. [See this](https://aws.amazon.com/blogs/compute/amazon-ecs-and-docker-volume-drivers-amazon-ebs/)
- Refactor the VPC and default security group permissions
## 2. Official Technical Reference & Guides (amundsen-io/amundsen-io.github.io)
# Amundsen Marketing Site
Built using NPM, Jekyll, and Tailwind.css via `npm start`.
## Repo structure and branches
We use GitHub pages to deploy amundsen.io. We use GitHub _organization_ pages (instead of _project_ pages). There are two ways to deploy GitHub pages:
1. You push the source (HTML, CSS, JS, etc.) to `master` branch and have GitHub build and deploy using Jekyll.
2. You push the distribution (usually a folder called `dist` from local build) to `master` branch. GitHub simply deploys the pre-built distribution then.
We only support option #2 for this repository. This means that `master` branch contains the code for generating the distribution. It also contains the locally built distribution in the `docs` folder (NOT `dist`). Note, as of now, you can only use `master` branch to put content of GitHub pages for an organization.
## Making changes
In order to make changes, you'd have to make changes locally, build using Jekyll, test changes locally and then push both the code and the distribution (in the `docs` folder) to the master branch. Use can follow [these instructions](https://docs.github.com/en/enterprise/2.14/user/articles/setting-up-your-github-pages-site-locally-with-jekyll) to do local set up. In order to build, use `gulp` and to serve locally, use `gulp serve`.