DLT Pipeline

Anuradha_Mel · ‎04-03-2025

Hello,

I have written below simple code to write data to Catalogue table using simple DLT pipeline .

As part of Below program am reading a file from blob container and trying to write to a Catalogue table . New catalogue table got created but table does not have any data in it . Do I need to explicitly give a write command in the code ? OR is there any alternative way to do it .

Spoiler

import dlt

storage_account_name = <Storage_Account>
storage_account_key = <account-key>
container_name = "e2e-data"
blob_file_path = "Test/tab1/tab1.csv"

spark.conf.set(
f"fs.azure.account.key.{storage_account_name}.blob.core.windows.net",
storage_account_key
)

csv_file_path = f"abfss://{container_name}@{storage_account_name}.blob.core.windows.net/{blob_file_path}"

schema = StructType([StructField("col1", IntegerType(), True), StructField("col2", StringType(), True), StructField("col3", StringType(), True), StructField("col4", StringType(), True), StructField("col5", StringType(), True)])

@Dlt.table(
comment="Raw data loaded from cloud storage",
name="main.default.table_name",

table_properties={
"quality": "bronze"
}
)
def ingest_from_storage():
return (
spark.read.format("csv")
.option("header", "true")
.schema(schema)
.load(csv_file_path)
)

import dltstorage_account_name = <Storage_Account>storage_account_key = <account-key>container_name = "e2e-data"blob_file_path = "Test/tab1/tab1.csv"spark.conf.set(f"fs.azure.account.key.{storage_account_name}.blob.core.windows.net",storage_account_key)csv_file_path = f"abfss://{container_name}@{storage_account_name}.blob.core.windows.net/{blob_file_path}"schema = StructType([StructField("col1", IntegerType(), True), StructField("col2", StringType(), True), StructField("col3", StringType(), True), StructField("col4", StringType(), True), StructField("col5", StringType(), True)]).table(comment="Raw data loaded from cloud storage",name="main.default.table_name",table_properties={"quality": "bronze"})def ingest_from_storage():return (spark.read.format("csv").option("header", "true").schema(schema).load(csv_file_path))