Forum Discussion

gopala000's avatar
gopala000
Frequent Visitor
1 year ago
Solved

Reading an existing DeltaTable throws DELTA_MISSING_DELTA_TABLE error

In the following code to move some tables from bronze to silver, I create the delta table in silver if it does not exists and merge the changes if a table exists already. 

delta.DeltaTable.forPath() call gives an error:

AnalysisException: [DELTA_MISSING_DELTA_TABLE] `abfss://<GUID REMOVED BY ME>@onelake.dfs.fabric.microsoft.com/<GUID REMOVED BY ME>/Tables/SettlementAccount/` is not a Delta table. I'm puzzled as the table exists in silver with the right format (see image). Please help to resolve this issue.

 

Source code:


 

import com.microsoft.spark.fabric
from pyspark.sql.functions import col, explode, when, array, to_timestamp

from com.microsoft.spark.fabric.Constants import Constants
from pyspark.sql.functions import sha2
from pyspark.sql.functions import lower, upper

# to convert a dataframe column to an array or to iterate df
import pandas as pd
import numpy as np

# for merge TODO: See if we can use SQL instead
import delta

# Address error reading dates before 1582-10-15 or timestamps before 1900-01-01T00:00:00Z from Parquet files can be ambiguous
spark.conf.set("spark.sql.parquet.datetimeRebaseModeInRead", "CORRECTED")
spark.conf.set("spark.sql.parquet.datetimeRebaseModeInWrite", "CORRECTED")

env = 'test01'

source_lh = f"{env}_lakehouse_bronze"
dest_lh = f"{env}_lakehouse_argentum"

# List of columns to anonymize
anon_cols = ['FirstName', 'LastName']

high_watermark_table = f"{dest_lh}.HighWaterMark"
high_watermark = spark.sql(f"select * from {high_watermark_table}")
# display(high_watermark)

# lower case the table names
tables = [x.name.lower() for x in spark.catalog.listTables(dest_lh)]
#for table in tables:
#    print(table)

# loop through each table and copy the data. 
# need to convert to pandas dataframe because only pandas df supports iterator
for index, row in high_watermark.toPandas().iterrows():
    # set the source and target tables
    x = row['TableName']
    source_table = f"{source_lh}.{x}"
    dest_table = f"{dest_lh}.{x}"
    dest_table_updates = f"{dest_lh}.{x}_updates"

    isNewTable = True
    # Check if the target table exists
    if x.lower() in tables:
        df_existing_check = spark.sql(f"select * from {dest_table} limit 1")

        # Check if rows exists in target table to look for changes
        # >= condition just in case some rows are not picked up due to datetime granularity
        if df_existing_check.count() > 0:
            isNewTable = False
            df_filtered = spark.sql(f"select * from {source_table} where PeriodStart >= (select max(PeriodStart) from {dest_table})")
        else:
            df_filtered = spark.sql(f"select * from {source_table}")
    else:
        df_filtered = spark.sql(f"select * from {source_table}")

    # If the column name exists, anonymize
    for y in anon_cols:
        if y in df_filtered.columns:
            print(f"Anonymizing column {y} from {source_table}...")
            df_filtered = df_filtered.withColumn(y, sha2(y, 256))

    if isNewTable:
        print(f"Writing {df_filtered.count()} rows from {source_table} to {dest_table}...")
        df_filtered.write.format('delta').mode('overwrite').saveAsTable(dest_table)
    else:
        #upsert
        #delta.DeltaTable.forName(dest_table) does not support db.tbl format
        print(f"Upsert {df_filtered.count()} rows from {source_table} to {dest_table} using {dest_table_updates}...")
        dest = delta.DeltaTable.forPath(spark, f"abfss://<GUID REMOVED>@onelake.dfs.fabric.microsoft.com/<GUID REMOVED>/Tables/{x}/")

        dest.alias('target').merge(dest_table_updates.alias('changes'), "target.Id = changes.Id").whenMatchedUpdateAll().whenNotMatchedInsertAll().whenNotMatchedBySourceDelete().execute()
		

 

 

 

 

  • Anonymous's avatar
    Anonymous
    1 year ago

    Hi gopala000 ,

     

    For the DELTA_MISSING_DELTA_TABLE error problem, this error typically indicates that the specified path does not contain a valid Delta table. Make sure that the path you provide to delta.DeltaTable.forPath() is correct and points to a valid Delta table.

    dest = delta.DeltaTable.forPath(spark, f"abfss://<GUID REMOVED>@onelake.dfs.fabric.microsoft.com/<GUID REMOVED>/Tables/{x}/")

     

    Best Regards,
    Adamk Kong

     

    If this post helps, then please consider Accept it as the solution to help the other members find it more quickly.

2 Replies

  • Anonymous's avatar
    Anonymous
    Not applicable

    Hi gopala000 ,

     

    For the DELTA_MISSING_DELTA_TABLE error problem, this error typically indicates that the specified path does not contain a valid Delta table. Make sure that the path you provide to delta.DeltaTable.forPath() is correct and points to a valid Delta table.

    dest = delta.DeltaTable.forPath(spark, f"abfss://<GUID REMOVED>@onelake.dfs.fabric.microsoft.com/<GUID REMOVED>/Tables/{x}/")

     

    Best Regards,
    Adamk Kong

     

    If this post helps, then please consider Accept it as the solution to help the other members find it more quickly.

    • gopala000's avatar
      gopala000
      Frequent Visitor

      Thanks for your response. I had to add x.lower() in the line to get the DeltaTable using forPath()

      to get past the error. Fixed a code bug in the last line to 

      dest.alias('target').merge(df_filtered.alias('changes'), "target.Id = changes.Id").whenMatchedUpdateAll().whenNotMatchedInsertAll().whenNotMatchedBySourceDelete().execute() to get it working! Thank you.

       Anonymous