I am trying to replicate my existing spark pipeline in DLT.
I am not able to achieve desired result using DLT .
Current pipeline :
source set up : CSV file ingested in bronze using SCP
frequency : monthly
bronze dir : /cntdlt/bronze/emp/year=2022 /month=1
silver : read bronze and perform some transformation and write in silver with same partition column
val df =spark.read.option(...).csv(s"{basepath}/year=${year}/month=${month}").withColumn("some logic","")
spark.sql(s"CREATE DATABASE IF NOT EXISTS $dbName")
spark.conf.set("spark.sql.sources.partitionOverwriteMode","dynamic")
val partitions= List("year","month" )
df.write.partitionBy(partitions:_*)
.mode(saveMode)
.option("path", silverpath)
.saveAsTable(s"${dbName}.${tableName}")
spark.sql(s"msck repair table $dbName.$tableName")
Scenario: consider some source data issue if i want to reload data for some previous month in my spark pipleine set up i just run the job for specific year and month value (provided by airlfow)
and because of spark.conf.set("spark.sql.sources.partitionOverwriteMode","dynamic")
it update the data in partition under processing .
I am trying to achieve similar set up using DLT pipeline .
DLT Setup
import dlt
from pyspark.sql.functions import *
from pyspark.sql.types import *
from datetime import date
from pyspark.sql import functions as F
schema='id int, \
date date,\
salary int,\
city string,\
name string,\
age int, \
year int, \
month int'
@dlt.create_table(
comment="The raw employee, ingested from /databricks-datasets.",
table_properties={
"diggibyte_emp.quality": "bronze",
"pipelines.autoOptimize.managed": "true",
"delta.appendOnly":"false"
}
)
def t_emp_raw():
return (
spark.readStream.format("cloudFiles")
.option('cloudFiles.format', 'json')
.option("partitionColumns", "year,month")
.option('cloudFiles.allowOverwrites', 'true')
.schema(schema)
.load( '/mnt/cntdlt/bronze/emp/')
)
@dlt.create_table(
comment="The cleaned emp reocrd with partitioned by year, month",
partition_cols = ["year", "month"],
table_properties={
"diggibyte_emp.quality": "silver",
"pipelines.autoOptimize.managed": "true",
"delta.appendOnly":"false"
}
)
def t_emp_silver():
return (
dlt.readStream("t_emp_raw")
.withColumn("load_date",current_date())
)
pipeline conf
{
"id": "8496fc69-5ee3-4d61-9db5-47a38f130785",
"clusters": [
{
"label": "default",
"num_workers": 1
}
],
"development": true,
"continuous": false,
"channel": "CURRENT",
"edition": "ADVANCED",
"photon": false,
"libraries": [
{
"notebook": {
"path": "/demo/notebook_employe"
}
}
],
"name": "monthly_load_employe",
"storage": "dbfs:/pipelines/8496fc69-5ee3-4d61-9db5-47a38f130785",
"target": "org"
}
DLT pipeline :
I have run the pipeline when bronze is having only 1 partition.
/cntdlt/bronze/emp/year=2022 /month=1 [consist 10 record ]
result in
t_emp_raw for partition year=2022 /month=1 = 10 records
t_empl_silver for partition year=2022 /month=1 = 10 records
Added data in bronze source
/cntdlt/bronze/emp/year=2022 /month=2 [consist 10 records ]
result
t_emp_raw for partition year=2022 /month=2 = 10 records
t_empl_silver for partition year=2022 /month=2 = 10 records
Now consider ,source has send updated data for 1 month then
my partition on 01 need to be update and overwritten
/cntdlt/bronze/emp/year=2022 /month=1 [consist 5 records ]
expected result
t_emp_raw for partition year=2022 /month=1/ : 5 records
t_empl_silver for partition year=2022 /month=1 / : 5 records
Actual result
t_emp_raw for partition year=2022 /month=1/ : 15 records
t_empl_silver for partition year=2022 /month=1/ :15 records
How can i achieve the overwriting of partition when it is updated?
