I have a process which loads data from json to a bronze table. It then adds a couple of columns and creates a silver table. But the silver table has NULL values where there were values in the bronze tables. Process as follows:
def load_to_silver(source, table):
file_format = 'json'
context_path = f'/mnt/company-data-landing/{source}/{table}/'
# Schema evolution set to rescue - new schema breaking columns will be stored in rescued rather than failing the pipeline run.
# backfillInterval ensures that any missed files are eventually processed (weekly).
cloudfile = {
'cloudFiles.schemaEvolutionMode': 'rescue',
'cloudFiles.schemaLocation': f'/mnt/ecologi-data-landing/{source}/{table}/schema/',
'checkpointLocation': f'/mnt/ecologi-data-landing/{source}/{table}/checkpoints/',
'schemaInference.sampleSize.numFiles': '200',
'cloudFiles.format': file_format,
'columnNameOfCorruptRecord': '_corrupt_record',
'triggerOnce': 'true',
'cloudFiles.backfillInterval': '1 week'
}
# Create a materialised staging table to house returned records.
@dlt.table(name=f'bronze_{source}_{table}',
comment=f'{source}_{table} processed from landing, to bronze streaming table. Data appended, with no changes.',
)
def bronze_billing_object():
df = spark.readStream.format("cloudFiles") \
.options(**cloudfile) \
.load(context_path) \
.withColumn("load_date", lit(datetime.datetime.now())) \
.withColumn("source_file", input_file_name())
return df
dlt.create_streaming_live_table(name = f'silver_staging_{source}_{table}',
comment = f'{source}_{table} processed from bronze streaming table, to silver streaming table/ view. \
Data merged rather than appended.',
)
if f'{table}'=='AddonsOnSubcription':
dlt.apply_changes(
target = f'silver_staging_{source}_{table}',
source = f'bronze_{source}_{table}',
keys = ['subscriptionId'],
sequence_by = col('createdAt'),
stored_as_scd_type = 1
)
elif f'{table}'=='Renewal':
dlt.apply_changes(
target = f'silver_staging_{source}_{table}',
source = f'bronze_{source}_{table}',
keys = ['invoiceId'],
sequence_by = col('createdAt'),
stored_as_scd_type = 1
)
else:
dlt.apply_changes(
target = f'silver_staging_{source}_{table}',
source = f'bronze_{source}_{table}',
keys = ['id'],
sequence_by = col('createdAt'),
stored_as_scd_type = 1
)
@dlt.table(name=f'silver_{source}_{table}',
comment=f'{source}_{table} processed from silver staging streaming table/view, to silver non staging, batch live table. \
Dtypes applied. Data copied from prev.',
)
def silver_billing_object():
df1 = dlt.read(f'silver_staging_{source}_{table}')
df2 = df1.withColumn("createdAt",col("createdAt").cast("timestamp"))
return (df2)
for table in table_list:
load_to_silver(notebook_context, table)
