yujunjun/OpenMetadata
1
0
Fork 0
mirror of https://github.com/open-metadata/OpenMetadata.git synced 2025-09-26 09:22:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

Added DBT Workflow fixes (#9419)

Browse Source 
* Added DBT fixes

* Addressed review comments
This commit is contained in:
Onkar Ravgan 2022-12-20 19:06:01 +05:30 committed by GitHub
parent f81867dd97
commit c1fdc59e84
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 25 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
ingestion/src/metadata/ingestion/source/database/dbt.py
View File

@ -286,28 +286,30 @@ class DbtSource(DbtServiceSource): # pylint: disable=too-many-public-methods
self.context.dbt_tests = {} self.context.dbt_tests = {}
for key, manifest_node in manifest_entities.items(): for key, manifest_node in manifest_entities.items():
try: try:
# Skip the analysis node since it does not contain relevant metatada
if manifest_node["resource_type"] in ["analysis"]:
continue
# If the run_results file is passed then only DBT tests will be processed # If the run_results file is passed then only DBT tests will be processed
if dbt_files.dbt_run_results: if (
dbt_files.dbt_run_results
and manifest_node["resource_type"] == "test"
):
# Test nodes will be processed further in the topology # Test nodes will be processed further in the topology
if manifest_node["resource_type"] == "test": self.context.dbt_tests[key] = manifest_node
self.context.dbt_tests[key] = manifest_node self.context.dbt_tests[key][
self.context.dbt_tests[key][ "upstream"
"upstream" ] = self.parse_upstream_nodes(manifest_entities, manifest_node)
] = self.parse_upstream_nodes( self.context.dbt_tests[key][
manifest_entities, manifest_node "results"
) ] = next( # pylint: disable=stop-iteration-return
self.context.dbt_tests[key][ item
"results" for item in dbt_files.dbt_run_results.get("results")
] = next( # pylint: disable=stop-iteration-return if item["unique_id"] == key
item )
for item in dbt_files.dbt_run_results.get("results") continue
if item["unique_id"] == key
) # Skip the analysis and test nodes
continue if manifest_node["resource_type"] in ("analysis", "test"):
logger.info(f"Skipping DBT node: {key}.")
continue
model_name = ( model_name = (
manifest_node["alias"] manifest_node["alias"]

8
ingestion/src/metadata/utils/dbt_config.py
View File

@ -79,7 +79,7 @@ def _(config: DbtLocalConfig):
) )
with open(config.dbtManifestFilePath, "r", encoding="utf-8") as manifest: with open(config.dbtManifestFilePath, "r", encoding="utf-8") as manifest:
dbt_manifest = manifest.read() dbt_manifest = manifest.read()
if config.dbtRunResultsFilePath is not None: if config.dbtRunResultsFilePath:
logger.debug( logger.debug(
f"Reading [dbtRunResultsFilePath] from: {config.dbtRunResultsFilePath}" f"Reading [dbtRunResultsFilePath] from: {config.dbtRunResultsFilePath}"
) )
@ -87,7 +87,7 @@ def _(config: DbtLocalConfig):
config.dbtRunResultsFilePath, "r", encoding="utf-8" config.dbtRunResultsFilePath, "r", encoding="utf-8"
) as run_results: ) as run_results:
dbt_run_results = run_results.read() dbt_run_results = run_results.read()
if config.dbtCatalogFilePath is not None: if config.dbtCatalogFilePath:
logger.debug( logger.debug(
f"Reading [dbtCatalogFilePath] from: {config.dbtCatalogFilePath}" f"Reading [dbtCatalogFilePath] from: {config.dbtCatalogFilePath}"
) )
@ -113,7 +113,7 @@ def _(config: DbtHttpConfig):
config.dbtManifestHttpPath config.dbtManifestHttpPath
) )
dbt_run_results = None dbt_run_results = None
if config.dbtRunResultsHttpPath is not None: if config.dbtRunResultsHttpPath:
logger.debug( logger.debug(
f"Requesting [dbtRunResultsHttpPath] to: {config.dbtRunResultsHttpPath}" f"Requesting [dbtRunResultsHttpPath] to: {config.dbtRunResultsHttpPath}"
) )
@ -122,7 +122,7 @@ def _(config: DbtHttpConfig):
) )
dbt_catalog = None dbt_catalog = None
if config.dbtCatalogHttpPath is not None: if config.dbtCatalogHttpPath:
logger.debug( logger.debug(
f"Requesting [dbtCatalogHttpPath] to: {config.dbtCatalogHttpPath}" f"Requesting [dbtCatalogHttpPath] to: {config.dbtCatalogHttpPath}"
) )