customer_orders.py

import dlt

source = "<catalog>.<schema>."

@dlt.table
def clean_orders():
  raw_orders = spark.read.table(f"{source}raw_orders")
  customers = spark.read.table(f"{source}customer_orders")
  join_conditions = [
    raw_orders.customer_id == customers.customer_id,
    raw_orders.customer_name == customers.customer_name,
  ]
  return (
    raw_orders.join(customers, on=join_conditions, how='left')
              .select(
                raw_orders.customer_id,
                raw_orders.customer_name,
                'order_datetime',
                'order_number',
                'ordered_product',
                'city',
                'state',
                'lat',
                'lon'
              )
  )

@dlt.table
def customer_orders():
    df = dlt.read("clean_orders")
    all_cols = df.columns
    needed_cols = [col for col in all_cols if col.startswith('customer') or col.startswith('order')]
    return df.select(needed_cols)

def select_all_filter_by(table, filter_column, filter_value):
  return dlt.read(table).filter(f"{filter_column}='{filter_value}'")

@dlt.table
def new_york_orders():
    return select_all_filter_by('clean_orders', 'state', 'NY')

@dlt.table
def texas_orders():
    return select_all_filter_by('clean_orders', 'state', 'TX')

@dlt.table
def seattle_orders():
  return select_all_filter_by('clean_orders', 'city', 'Seattle')