yujunjun/datahub
2
0
Fork 0
mirror of https://github.com/datahub-project/datahub.git synced 2025-08-16 21:27:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ingest): snowflake - graceful error handling in snowflake classification (#6568)

Browse Source
This commit is contained in:
Mayuri Nehate 2022-11-29 16:54:24 +05:30 committed by GitHub
parent d4bf6ce3fc
commit ec056211a8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 30 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
metadata-ingestion/setup.py
View File

@ -43,7 +43,7 @@ framework_common = {
"stackprinter>=0.2.6", "stackprinter>=0.2.6",
"tabulate", "tabulate",
"progressbar2", "progressbar2",
"termcolor>=2.0.0", "termcolor>=1.0.0",
"psutil>=5.8.0", "psutil>=5.8.0",
"ratelimiter", "ratelimiter",
"Deprecated", "Deprecated",

5
metadata-ingestion/src/datahub/ingestion/source/snowflake/snowflake_config.py
View File

@ -39,7 +39,10 @@ class SnowflakeV2Config(SnowflakeConfig, SnowflakeUsageConfig):
default=None, description="Not supported" default=None, description="Not supported"
) )
classification: Optional[ClassificationConfig] = None classification: Optional[ClassificationConfig] = Field(
default=None,
description="For details, refer [Classification](../../../../metadata-ingestion/docs/dev_guides/classification.md).",
)
@root_validator(pre=False) @root_validator(pre=False)
def validate_unsupported_configs(cls, values: Dict) -> Dict: def validate_unsupported_configs(cls, values: Dict) -> Dict:

21
metadata-ingestion/src/datahub/ingestion/source/snowflake/snowflake_v2.py
View File

@ -566,9 +566,16 @@ class SnowflakeV2Source(
dataset_name = self.get_dataset_identifier(table.name, schema_name, db_name) dataset_name = self.get_dataset_identifier(table.name, schema_name, db_name)
if self.is_classification_enabled_for_table(dataset_name): if self.is_classification_enabled_for_table(dataset_name):
try:
table.sample_data = self.get_sample_values_for_table( table.sample_data = self.get_sample_values_for_table(
conn, table.name, schema_name, db_name conn, table.name, schema_name, db_name
) )
except Exception as e:
self.warn(
self.logger,
dataset_name,
f"unable to get table sample data due to error -> {e}",
)
lineage_info = None lineage_info = None
if self.config.include_table_lineage: if self.config.include_table_lineage:
@ -757,8 +764,20 @@ class SnowflakeV2Source(
self.snowflake_identifier(col) for col in table.sample_data.columns self.snowflake_identifier(col) for col in table.sample_data.columns
] ]
logger.debug(f"Classifying Table {dataset_name}") logger.debug(f"Classifying Table {dataset_name}")
try:
self.classify_schema_fields( self.classify_schema_fields(
dataset_name, schema_metadata, table.sample_data.to_dict(orient="list") dataset_name,
schema_metadata,
table.sample_data.to_dict(orient="list")
if table.sample_data is not None
else {},
)
except Exception as e:
self.warn(
self.logger,
dataset_name,
f"unable to classify table columns due to error -> {e}",
) )
return schema_metadata return schema_metadata