Forum Discussion

JonBFabric's avatar
JonBFabric
Helper I
8 months ago
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:   ...
data engineering
Lakehouse
  • JonBFabric's avatar
    JonBFabric
    8 months ago

    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 DeltaTable
    import json
     
    # Note that the table name must be lowercase
    table_path = '<Path_To_Table>'
     
    # Identify log files
    log_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 schema
    filesWithSchema = []

    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 schema
    latest = sorted(filesWithSchema, reverse=True)[0]
    content = notebookutils.fs.head(latest, 5000000)
     
    # Extract the schema
    JSONdocs = content.split('\n')

    for doc in JSONdocs:
        if 'schemaString' in doc:
            schemaString = json.loads(doc).get("metaData", {}).get("schemaString")

    # Extract Field Metadata
    FieldList = []
    OrdinalPosition = 0

    for field in json.loads(schemaString).get("fields") :
        OrdinalPosition += 1
        FieldDetails = {}
        FieldDetails['FieldName'] = field.get("name")
        FieldDetails['Nullable'] = field.get("nullable")
        FieldDetails['OrdinalPosition'] = OrdinalPosition

        match 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)