yujunjun/OpenMetadata
1
0
Fork 0
mirror of https://github.com/open-metadata/OpenMetadata.git synced 2025-10-20 05:15:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix Redshift Usage (#3624)

Browse Source
This commit is contained in:
Ayush Shah 2022-03-24 10:37:25 +05:30 committed by GitHub
parent f104ab632b
commit a87ed206c0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 840 additions and 840 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1656
ingestion/examples/sample_data/datasets/query_log
View File

File diff suppressed because it is too large Load Diff

2
ingestion/src/metadata/ingestion/source/redshift_usage.py
View File

@ -90,7 +90,7 @@ class RedshiftUsageSource(Source[TableQuery]):
starttime=str(row["starttime"]), starttime=str(row["starttime"]),
endtime=str(row["endtime"]), endtime=str(row["endtime"]),
analysis_date=str(self.analysis_date), analysis_date=str(self.analysis_date),
database=row["database"], database=row["schema"],
aborted=row["aborted"], aborted=row["aborted"],
sql=row["querytxt"], sql=row["querytxt"],
service_name=self.config.service_name, service_name=self.config.service_name,

22
ingestion/src/metadata/ingestion/stage/table_usage.py
View File

@ -27,7 +27,7 @@ from metadata.ingestion.stage.file import FileStageConfig
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def get_table_column_join(table, table_aliases, joins): def get_table_column_join(table, table_aliases, joins, database):
table_column = None table_column = None
joined_with = [] joined_with = []
for join in joins: for join in joins:
@ -35,7 +35,11 @@ def get_table_column_join(table, table_aliases, joins):
if "." not in join: if "." not in join:
continue continue
jtable, column = join.split(".")[-2:] jtable, column = join.split(".")[-2:]
if table == jtable or jtable in table_aliases: if (
table == jtable
or table == f"{database}.{jtable}"
or jtable in table_aliases
):
table_column = TableColumn( table_column = TableColumn(
table=table_aliases[jtable] if jtable in table_aliases else jtable, table=table_aliases[jtable] if jtable in table_aliases else jtable,
column=column, column=column,
@ -93,14 +97,20 @@ class TableUsageStage(Stage[QueryParserData]):
if record.columns.get("join") is not None: if record.columns.get("join") is not None:
table_usage_count.joins.append( table_usage_count.joins.append(
get_table_column_join( get_table_column_join(
table, record.tables_aliases, record.columns["join"] table,
record.tables_aliases,
record.columns["join"],
record.database,
) )
) )
else: else:
joins = [] joins = []
if record.columns.get("join") is not None: if record.columns.get("join") is not None:
tbl_column_join = get_table_column_join( tbl_column_join = get_table_column_join(
table, record.tables_aliases, record.columns["join"] table,
record.tables_aliases,
record.columns["join"],
record.database,
) )
if tbl_column_join is not None: if tbl_column_join is not None:
joins.append(tbl_column_join) joins.append(tbl_column_join)
@ -114,12 +124,8 @@ class TableUsageStage(Stage[QueryParserData]):
) )
except Exception as exc: except Exception as exc:
logger.error("Error in staging record {}".format(exc)) logger.error("Error in staging record {}".format(exc))
self.status.failures(
f"Table: {table}", "Error in staging record {}".format(exc)
)
self.table_usage[table] = table_usage_count self.table_usage[table] = table_usage_count
logger.info(f"Successfully record staged for {table}") logger.info(f"Successfully record staged for {table}")
self.status.records_status(f"Table: {table}")
def get_status(self): def get_status(self):
return self.status return self.status