Forum Discussion
Solved
Lakehouse Table Generate Create Table
Hi, I maintain server lakehouses, and due to issues with the deployment pipelines tend to apply schema changes through script, and to retain the data in the table use the following process: ...
One final update on this.
The logfile containing the schema is not necessarily the most recent, and there is not necessarily only one version of the schema. There is a schema associated with every modification made to the table structure, be that the original creation or subsequent alterations. Consequently, the logfile we need is the most recent with a schema.
from delta.tables import DeltaTableimport json# Note that the table name must be lowercasetable_path = '<Path_To_Table>'# Identify log fileslog_dir = f"{table_path}/_delta_log"files = [f.path for f in notebookutils.fs.ls(log_dir) if f.name.endswith(".json")]# Identify log files with a schemafilesWithSchema = []for file in sorted(files, reverse=True) :content = notebookutils.fs.head(file, 5000000)JSONdocs = content.split('\n')for doc in JSONdocs:if 'schemaString' in doc:filesWithSchema.append(file)# Load the header for the latest log file containing a schemalatest = sorted(filesWithSchema, reverse=True)[0]content = notebookutils.fs.head(latest, 5000000)# Extract the schemaJSONdocs = content.split('\n')for doc in JSONdocs:if 'schemaString' in doc:schemaString = json.loads(doc).get("metaData", {}).get("schemaString")# Extract Field MetadataFieldList = []OrdinalPosition = 0for field in json.loads(schemaString).get("fields") :OrdinalPosition += 1FieldDetails = {}FieldDetails['FieldName'] = field.get("name")FieldDetails['Nullable'] = field.get("nullable")FieldDetails['OrdinalPosition'] = OrdinalPositionmatch field.get("type").split('(')[0]:case 'string':FieldDetails['SQLType'] = field.get("metadata").get("__CHAR_VARCHAR_TYPE_STRING")case 'timestamp':FieldDetails['SQLType'] = 'timestamp'case 'date':FieldDetails['SQLType'] = 'date'case 'integer':FieldDetails['SQLType'] = 'int'case 'short':FieldDetails['SQLType'] = 'smallint'case 'long':FieldDetails['SQLType'] = 'bigint'case 'decimal':FieldDetails['SQLType'] = field.get("type")case 'boolean':FieldDetails['SQLType'] = 'boolean'FieldList.append(FieldDetails)display(FieldList)
Thanks.
This didn't quite work out of the box, possibly because it was originally written for DataBricks rather than Fabric, but I have got it working. I will explan the differences as I go:
from delta.tables import DeltaTable
import json
table_path = "<Path_To_Table>" # update this
dt = DeltaTable.forPath(spark, table_path)
print(dt.toDF().schema) # shows VarcharType(n) if declared
.schema shows only string as type, so the last 2 lines can be removed.
log_dir = f"{table_path}/_delta_log"
files = [f.path for f in notebookutils.fs.ls(log_dir) if f.name.endswith(".json")]
latest = sorted(files)[-1]
content = notebookutils.fs.head(latest, 5000000)
Note that dbutils is now replaced by notebookutils.
content can not however be read as JSON, as it is infact 3 JSON documents seperated by
'\n', and not just 1, and the document which contains the field metadata is the 2nd.
schemaString = json.loads(content.split('\n')[1]).get("metaData", {}).get("schemaString")
schemaString is actually an embedded JSON document held as a string, and so has to be converted also. The following snippet then prints the field name and the actual sql type.
for field in json.loads(schemaString).get("fields"):
print("FieldName:", field.get("name"), ", Type:", field.get("metadata").get("__CHAR_VARCHAR_TYPE_STRING"))
Certainly not a finished article, but it gives me what I need to build around.
Thanks for your help and patience.
Hi JonBFabric , Thanks for the update and the insights on how to solve this issue. We really appreciate it.
If you have any queries, please feel free to create a new post, we are always happy to help.
One final update on this.
The logfile containing the schema is not necessarily the most recent, and there is not necessarily only one version of the schema. There is a schema associated with every modification made to the table structure, be that the original creation or subsequent alterations. Consequently, the logfile we need is the most recent with a schema.
from delta.tables import DeltaTableimport json# Note that the table name must be lowercasetable_path = '<Path_To_Table>'# Identify log fileslog_dir = f"{table_path}/_delta_log"files = [f.path for f in notebookutils.fs.ls(log_dir) if f.name.endswith(".json")]# Identify log files with a schemafilesWithSchema = []for file in sorted(files, reverse=True) :content = notebookutils.fs.head(file, 5000000)JSONdocs = content.split('\n')for doc in JSONdocs:if 'schemaString' in doc:filesWithSchema.append(file)# Load the header for the latest log file containing a schemalatest = sorted(filesWithSchema, reverse=True)[0]content = notebookutils.fs.head(latest, 5000000)# Extract the schemaJSONdocs = content.split('\n')for doc in JSONdocs:if 'schemaString' in doc:schemaString = json.loads(doc).get("metaData", {}).get("schemaString")# Extract Field MetadataFieldList = []OrdinalPosition = 0for field in json.loads(schemaString).get("fields") :OrdinalPosition += 1FieldDetails = {}FieldDetails['FieldName'] = field.get("name")FieldDetails['Nullable'] = field.get("nullable")FieldDetails['OrdinalPosition'] = OrdinalPositionmatch field.get("type").split('(')[0]:case 'string':FieldDetails['SQLType'] = field.get("metadata").get("__CHAR_VARCHAR_TYPE_STRING")case 'timestamp':FieldDetails['SQLType'] = 'timestamp'case 'date':FieldDetails['SQLType'] = 'date'case 'integer':FieldDetails['SQLType'] = 'int'case 'short':FieldDetails['SQLType'] = 'smallint'case 'long':FieldDetails['SQLType'] = 'bigint'case 'decimal':FieldDetails['SQLType'] = field.get("type")case 'boolean':FieldDetails['SQLType'] = 'boolean'FieldList.append(FieldDetails)display(FieldList)