I have created a target table in the same dlt pipeline.
But when I read that table in different block of notebook with Live.table_path. It is not able to read it
Here is my code block 1 Creating a streaming table
# Define metadata tables
catalog = spark.conf.get("catalog")
entity_table_path = f"{catalog}.metadata.entity"
elements_table_path = f"{catalog}.metadata.elements"
user = spark.conf.get("user")
import sys
sys.path.append(f"/Workspace/Users/{user}/.bundle/wtd_ingestion_framework/dev/files/src/")
# Import necessary modules
import dlt
from pyspark.sql.types import StructType, StructField, StringType, DecimalType, IntegerType
from pyspark.sql.functions import col
from utils.dlt_utils import create_dlt_table, apply_schema_evolution
from core.metadata_management import apply_classification_tags, map_data_type
from utils.logging_handling import log_event, get_logger
# Initialize logger
logger = get_logger()
# Read metadata
entities = spark.read.table(entity_table_path)
elements = spark.read.table(elements_table_path)
# Iterate through each entity (table)
for entity_row in entities.collect():
table_name = entity_row["Entity_Technical_Name"]
target_catalog = entity_row["Target_Catalog"]
target_schema = entity_row["Target_Schema"]
table_path = f"{target_catalog}.{target_schema}.{table_name}"
filtered_elements = elements.filter(elements["Source_feed_name"] == table_name).orderBy("Position")
# Define schema using elements metadata
schema = StructType([
StructField(
row["Entity_Technical_Name"], # Adjusted to match column key
map_data_type(row["Datatype"], row["Length"] if "Length" in row else None),
row["Nullable"] == "yes"
)
for row in filtered_elements.filter(elements["Source_feed_name"] == table_name).collect()
])
# Check if the table exists using inline logic
try:
spark.sql(f"DESCRIBE TABLE {table_path}")
table_exists = True
except Exception:
table_exists = False
# Check if the table exists and is not empty
try:
table_exists = True
table_is_empty = spark.sql(f"SELECT COUNT(1) FROM {table_path}").collect()[0][0] == 0
except Exception:
table_exists = False
table_is_empty = True
# If the table exists and is not empty, use it as a streaming source
if table_exists and not table_is_empty:
dataframe = spark.readStream.table(table_path)
create_dlt_table(
dataframe=dataframe,
table_name=table_name,
schema=schema,
table_type="streaming"
)
log_event(logger, f"Streaming table {table_name} initialized with existing data.", "INFO")
# If the table doesn't exist or is empty, create an empty streaming table
else:
create_dlt_table(
table_name=table_name,
schema=schema,
table_type="streaming"
)
log_event(logger, f"Created new empty streaming table {table_path}.", "INFO")
# Apply the classification tags
# apply_classification_tags(target_catalog, target_schema, table_name)
the part of create_dlt_table function used in the above block(in a core.py file)
# If neither DataFrame nor input path is provided, create an empty table
else:
if not schema:
raise ValueError("Schema must be provided for empty tables without input_path or DataFrame.")
@Dlt.table(
name=table_name,
comment=f"Empty {table_type} DLT table created with schema."
)
def empty_table():
log_event(logger, f"Creating empty {table_type} table {table_name}.", "INFO")
if table_type == "streaming":
# Create a dummy streaming DataFrame with the `rate` source
streaming_df = spark.readStream.format("rate").option("rowsPerSecond", 1).load()
# Map the `rate` source columns to match the schema with default null values
empty_stream_df = streaming_df.select(
*[lit(None).cast(field.dataType).alias(field.name) for field in schema.fields]
).where("1=0") # Ensures the stream is empty
return empty_stream_df
else:
# Create an empty DataFrame for materialized view
return spark.createDataFrame([], schema)
log_event(logger, f"Empty {table_type} table {table_name} created successfully.", "INFO")
Code block 2: Ingestion data from auto_loader into the target_Table using append_flow and adding some extra columns
# Load parameters
import dlt
target_catalog = spark.conf.get("catalog").lower()
target_schema = spark.conf.get("target_schema").lower()
target_table_name = spark.conf.get("target_table_name").lower()
data_file_path = spark.conf.get("data_file_path")
header = spark.conf.get("header")
file_format = spark.conf.get("file_format")
user = spark.conf.get("user")
evolve_schema = spark.conf.get("evolve_schema")
mode = spark.conf.get("mode")
import sys
sys.path.append(f"/Workspace/Users/{user}/.bundle/wtd_ingestion_framework/dev/files/src/")
# Import necessary modules
from utils.logging_handling import get_logger, log_event, log_error
from utils.dlt_utils import create_dlt_table
from core.model_extensions import apply_model_extensions
from pyspark.sql import SparkSession
# Initialize logger and Spark session
logger = get_logger()
log_event(logger, "Starting file ingestion process.", "INFO")
table_path = f"{target_catalog}.{target_schema}.{target_table_name}"
filtered_elements = elements.filter(elements["Source_feed_name"] == "BB1123_loans").orderBy("Position")
# Define schema using elements metadata
schema = StructType([
StructField(
row["Entity_Technical_Name"], # Adjusted to match column key
map_data_type(row["Datatype"], row["Length"] if "Length" in row else None),
row["Nullable"] == "yes"
)
for row in filtered_elements.filter(elements["Source_feed_name"] == "BB1123_loans").collect()
])
log_event(logger, f"Schema: {schema}", "INFO")
# @Dlt.table(name ="test")
# def test_table():
# return spark.read.table(table_path)
header="true"
evolve_schema=True
schema_hints = None
if schema and evolve_schema:
schema_hints = ",".join([f"{field.name} {field.dataType.simpleString()}" for field in schema.fields])
# schema_hints = {field.name: field.dataType.simpleString() for field in schema.fields}
log_event(logger, f"Schema hints: {schema_hints}", "INFO")
# Determine schema evolution mode based on header and evolve_schema
if header == "true":
infer_column_types = "true" # Infer schema from file
schema_evolution_mode = "addNewColumns" if evolve_schema else "rescue"
else:
infer_column_types = "false" # Do not infer schema
schema_evolution_mode = "addNewColumns" if evolve_schema else "rescue"
# Define the cloud file configuration
cloudfile = {
"cloudFiles.format": file_format,
"cloudFiles.allowOverwrites": "false",
"cloudFiles.inferColumnTypes": infer_column_types,
"cloudFiles.schemaLocation": f"{data_file_path}/_schema/{table_name}/",
"cloudFiles.schemaEvolutionMode": schema_evolution_mode,
"cloudFiles.schemaHints": schema_hints,
"cloudFiles.useNotifications": "false",
"cloudFiles.includeExistingFiles": "false"
}
file_specific_config = {
"header": header, # Pass the header option directly
"delimiter": "|", # Set pipe as the delimiter
"quote": "\"", # Handle quoted fields
"escape": "\\" # Use backslash as escape character
}
table_type = "Streaming"
@dlt.append_flow(
target = target_table_name,
name = f"{target_table_name}_{file_format}_ingestion",
# name = f"{target_table_name}",
comment = f"{mode} Mode {table_type} DLT table created/updated from {data_file_path}."
)
def source_to_bronze():
input_df = (spark
.readStream
.format("cloudfiles")
.options(**cloudfile)
.options(**file_specific_config)
.option("checkpointLocation", f"{data_file_path}/_checkpoints/{target_table_name}")
.load(data_file_path))
# return df.writeStream.trigger(availableNow=True if mode == "batch" else False)
input_df = apply_model_extensions(input_df,target_catalog,target_schema,target_table_name)
log_event(logger, f"Inferred columns: {input_df.columns}", "INFO")
# Write to Delta
input_df.writeStream \
.format("delta") \
.option("mergeSchema", "true") \
.outputMode("append") \
.option("checkpointLocation", f"{data_file_path}/_checkpoints/{target_table_name}") \
.start(table_path)
return input_df
log_event(logger, f"DLT {table_type} table {table_name} created or updated from input path.", "INFO")
Code block 3: Doing some post inegstion checks using append_flow again. I want to use expectations as well on this block but cannot use bcz expectations not allowed in append_flow. (Plz provide a solution or idea for this as well)
import sys
from delta.tables import DeltaTable
from pyspark.sql.functions import expr, lit, count
# Add your module paths
sys.path.append(f"/Workspace/Users/{user}/.bundle/wtd_ingestion_framework/dev/files/src/")
# Import necessary modules
from utils.logging_handling import get_logger, log_event
from utils.dlt_utils import create_dlt_table
from utils.helper import get_rules, get_primary_key
# Initialize logger
logger = get_logger()
log_event(logger, "Started Post Ingestion Checks", "INFO")
# Get table details
target_table_name = spark.conf.get("target_table_name")
target_catalog = spark.conf.get("catalog")
table_path = f"{target_catalog}.{target_schema}.{target_table_name}"
log_event(logger, f"DLT table {table_path}", "INFO")
# @Dlt.expect_all(get_rules(table_path))
@dlt.append_flow(
target=f"{target_table_name}",
name = f"{target_table_name}_post_ingestion_checks"
)
def validation_table():
# Read the source Delta table
record_count = spark.read.table(f"LIVE.{table_path}").count()
log_event(logger, f"Record count: {record_count}", "INFO")
df = spark.readStream.table(f"LIVE.{table_path}")
# df.withColumn("record_count", lit(record_count))
# Add quarantine flag using the dynamic record_count
rules = get_rules(table_path)
log_event(logger, f"Quarantine rules: {list(rules.values())}", "INFO")
if len(rules) == 1:
quarantine_rules = f"NOT({list(rules.values())[0]})"
else:
quarantine_rules = f"NOT({' OR '.join(rules.values())})"
log_event(logger, f"Quarantine rules applied: {quarantine_rules}", "INFO")
df_quarantined = df.withColumn("record_count", lit(record_count)).withColumn("Quarantined_flag", expr(quarantine_rules))
df_quarantined = df_quarantined.drop("record_count")
# Write the updated DataFrame
query = (
df_quarantined.writeStream.format("delta")
.option("mergeSchema", "true")
.option("checkpointLocation", f"{data_file_path}/_checkpoints/{target_table_name}")
.outputMode("append")
.start(table_path)
)
Getting this error
py4j.protocol.Py4JJavaError: Traceback (most recent call last):
File "/Workspace/Users/ashraf@wtdanalytics.com/.bundle/wtd_ingestion_framework/dev/files/src/notebooks/load_metadata/task_schema_changes", cell 6, line 31, in validation_table
record_count = spark.read.table(f"LIVE.{table_path}").count()
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
py4j.protocol.Py4JJavaError: An error occurred while calling z:com.databricks.pipelines.Pipeline.readBatchInput.
: com.databricks.pipelines.common.errors.DLTSparkException: [DATASET_NOT_DEFINED] Failed to read dataset 'accounts_and_customer.bronze.bb1123_loans'. This dataset is not defined in the pipeline.
If this table is managed by another pipeline, then do not use `dlt.read` / `dlt.readStream` to read the table or prepend the name with the LIVE keyword.
