Forum Discussion
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()
- Anonymous1 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 KongIf this post helps, then please consider Accept it as the solution to help the other members find it more quickly.
2 Replies
- AnonymousNot 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 KongIf this post helps, then please consider Accept it as the solution to help the other members find it more quickly.
- gopala000Frequent 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