This is the implementaiton of the original pos, for interest (excluded from the original post to reduce length)
# some blacklisting and schema renamings where necessary (due to conflicts and/or illegal characters)
SourceName=str
SchemaName=str
TableName=str
ColName=str
SourceSchemaName = str
TargetSchemaName = str
excl_schemas: Dict[SourceName, List[SchemaName]] = {
"my_first_source": [
"information_schema",
"sys",
"performance_schema",
"mysql",
"redcap",
],
"my_snd_source": [...
]
}
excl_tables: Dict[SourceName, Dict[SchemaName, Set[TableName]]] = {
'my_first_source': {},
'my_snd_source': {},
'data_dump': {
"datadump": {"table_logstore_huge_log_file"}
}
}
#todo: write to table as record of what has been excluded
rename_schemas: Dict[SourceName, Dict[SourceSchemaName, TargetSchemaName]] = {
"my_first_source": {
"my-schema": "my_schema",
"main": "mys_source_main",
},
}
# create a log and update the mirror of the source schema info
def python_to_sql_list(py_list: List[str]) -> str:
"""converts a python list into a sql-parsable list """
return "('" + "', '".join(py_list) + "')"
schemas_to_exclude: List[str] = excl_schemas[current_source]
excl_schema_str: str = python_to_sql_list(schemas_to_exclude)
info_schema_query: str = f"select TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, COLUMN_KEY from information_schema.columns WHERE TABLE_SCHEMA not in {excl_schema_str}"
generate_diff_table(query=info_schema_query,
table_namem=f"info_sch_{current_source}",
mirror_table_location=f"pipeline_utils.info_schema_change_logs",
diff_log_location=f"pipeline_utils.info_schema_change_logs")
# get a list of the tables to ingest from the mirror of the info_schema
tables_to_ingest: List[Tuple[str, List[str]]] = spark.sql(
f"Select distinct TABLE_SCHEMA, TABLE_NAME FROM pipeline_utils.info_schemas_mirror.{current_source} WHERE TABLE_SCHEMA NOT IN {ex_schema_str}"
).rdd.groupByKey().mapValues(list).collect()
# The actual ingestion bit
n_total: int = len(tables_to_ingest)
i: int = 1
# Iterate through each table to update the diff log and the local mirror
for schema_name, table_list in tables_to_ingest:
# deal with renaming as required
schema_name = schema_name
if source_schema_name in rename_schemas[current_source]:
target_schema_name = rename_schemas[current_source][source_schema_name]
else: target_schema_nm = schema_nm
# get any tables listed for exclusion, else return empty set.
tables_to_exclude: Set[str] = excl_tables[current_source].get(source_schema_name, set())
# process each table
for table_name in table_list:
print(f"{i}/{n_total}") # keep track of progress
i += 1
if table_name not in tables_to_exclude:
generate_diff_table(
query=f"select * from {source_schema_name}.{table_name}",
table_name=f"{schema_name}_{table_name}",
mirror_table_location=f"bronze.{target_schema_name}",
diff_log_loc=f"bronze.{target_schema_name}"
)
