@Palash01
This is my bronze pipeline.
# goods_grp 테이블 load
tables = {
"USER": {"id": ["USER_NO"]}
}
def generate_tables(table, info):
@dlt.table(
name=f"{table.lower()}_cdc_raw",
table_properties={"quality": "bronze"},
comment=f"Raw(Source) MySQL Data from DMS for the table: {table}",
temporary=True,
)
def create_call_table():
stream = (
spark.readStream.format("cloudFiles")
.option("cloudFiles.format", "parquet")
# .option("cloudFiles.includeExistingFiles", "false")
.load(f"{INPUT_S3_PATH}/{table}/")
)
if "Op" not in stream.columns:
stream = stream.withColumn("Op", F.lit(None).cast(T.StringType()))
return stream.selectExpr("*", "_metadata as source_metadata", "source_metadata.file_name as source_file_name", "source_metadata.file_path as source_file_path")
dlt.create_streaming_table(
name=f"{table.lower()}",
comment="Bronze MySQL Data from DMS for the table: {table}",
table_properties={
"myCompanyPipeline.quality": "bronze",
"pipelines.autoOptimize.managed": "true",
}
)
dlt.apply_changes(
target=f"{table.lower()}",
source=f"{table.lower()}_cdc_raw",
keys=[*info["id"]],
sequence_by=F.col("ADD_TIME"),
apply_as_deletes=F.expr("Op = 'D'"),
except_column_list=["Op", "_rescued_data"],
stored_as_scd_type=1,
)
for table, info in tables.items():
generate_tables(table, info)
This is a bronze pipeline that uses autoloader to process parquet files in S3 and cdc processing.
In addition to this user table, I'm fetching other tables in the same way as above, and I want to make them into silver tables by doing JOIN and other operations.
@dlt.view
def user_raw():
user_raw = read_delta_table(
schema=TargetSchema.HMMALL, table_name="user", is_stream=False
).selectExpr('*', 'to_timestamp(ADD_TIME) as USER_ADD_TIME', 'REG_DT as USER_REG_DT')
return user_raw
@dlt.view
def shop_raw():
return read_delta_table(
schema=TargetSchema.HMMALL, table_name="shop", is_stream=False
).selectExpr("SHOP_NO", "DEF_SHOP_NM")
@dlt.view
def nation_gaon_raw():
return read_delta_table(
schema=TargetSchema.HMMALL, table_name="nation_gaon", is_stream=False
).selectExpr("NATION_NO", "NATION_NM_KR")
def join_table_with_expr(
left: DataFrame, right: DataFrame, join_expr: str, how_to: str
) -> DataFrame:
return left.join(right, on=F.expr(join_expr), how=how_to)
def generate_user_silver() -> DataFrame:
join_shop_df = join_table_with_expr(
left=dlt.read('user_raw'),
right=dlt.read('shop_raw'),
join_expr="REG_SHOP_NO = SHOP_NO",
how_to="left",
)
join_nation_df = join_table_with_expr(
left=join_shop_df,
right=dlt.read('nation_gaon_raw'),
join_expr="NATI_NO = NATION_NO",
how_to="left",
)
return join_nation_df
def split_user_register_date(df: DataFrame, datetime_cols: Column) -> DataFrame:
split_date_cols = {
"USER_REG_DATE": F.to_date(datetime_cols),
"USER_REG_YEAR" : F.year(datetime_cols),
"USER_REG_MONTH" : F.month(datetime_cols),
"USER_REG_DAY" : F.dayofmonth(datetime_cols),
"USER_REG_TIME" : F.date_format(datetime_cols, "HH"),
"USER_REG_DAY_OF_WEEK" : F.dayofweek(datetime_cols)
}
return df.withColumns(split_date_cols)
@dlt.create_table(
name=f"user_silver",
comment=f"Silver Table from user - Join shop, nation_gaon / Add broker_type",
table_properties={
"myCompanyPipeline.quality": "silver",
"pipelines.autoOptimize.managed": "true",
# "pipelines.reset.allowed": "true"
},
)
def user_silver():
# user_stream = dlt.readStream("user_raw")
# shop = dlt.read('shop_raw')
# nation_gaon = dlt.read('nation_gaon_raw')
# join_shop_df = join_table_with_expr(left=dlt.readStream('user_raw'), right=dlt.read('shop_raw'), join_expr="REG_SHOP_NO = SHOP_NO", how_to="left")
# join_nation_df = join_table_with_expr(left=join_shop_df, right=dlt.read('nation_gaon_raw'), join_expr="NATI_NO = NATION_NO", how_to="left")
join_df = generate_user_silver()
user_reg_dt_cols = F.col('USER_REG_DT')
result = split_user_register_date(df=join_df, datetime_cols=user_reg_dt_cols)
return lower_cols_name(target=result).drop(
"source_metadata", "source_file_name", "source_file_path"
)
The code above is a SILVER table.
The problem here is that bronze is streaming, so data is constantly coming in, but if you make the output of the silver table streaming, you will get an error saying that you need to turn on the skipCommitChanges option for future data.
If you turn on skipCommitChanges, data after the first time you turn on the pipeline will not be updated.
So, out of necessity, I read the bronze streaming table as a read and made it a materialized view.
The notebooks for bronze and silver are all different.
Is this how you typically create a silver table?
I would like to know the example code of how you usually create SILVER, and I would also like to know what to fix about the above code.
