Mastech BrickLabs

/* SAS DATA Step with Conditions */
DATA work.customers_filtered;
    SET raw.customers;
    WHERE status = 'ACTIVE';

    /* Calculate tenure */
    tenure_years = INTCK('YEAR',
        join_date, TODAY());

    /* Categorize customers */
    IF tenure_years >= 5 THEN
        segment = 'LOYAL';
    ELSE IF tenure_years >= 2 THEN
        segment = 'ESTABLISHED';
    ELSE
        segment = 'NEW';
RUN;

# PySpark DataFrame Transformation
from pyspark.sql import functions as F
from pyspark.sql.functions import when

customers_filtered = (
    spark.table("raw.customers")
    .filter(F.col("status") == "ACTIVE")
    .withColumn(
        "tenure_years",
        F.floor(F.datediff(
            F.current_date(),
            F.col("join_date")) / 365)
    )
    .withColumn(
        "segment",
        when(F.col("tenure_years") >= 5, "LOYAL")
        .when(F.col("tenure_years") >= 2, "ESTABLISHED")
        .otherwise("NEW")
    )
)

PROC SQL;
    CREATE TABLE work.summary AS
    SELECT
        region,
        product_category,
        COUNT(*) AS order_count,
        SUM(amount) AS total_sales,
        AVG(amount) AS avg_order_value
    FROM raw.orders
    WHERE order_date >= '01JAN2024'd
    GROUP BY region, product_category
    HAVING COUNT(*) > 100
    ORDER BY total_sales DESC;
QUIT;

# PySpark Aggregation
summary = (
    spark.table("raw.orders")
    .filter(F.col("order_date") >= "2024-01-01")
    .groupBy("region", "product_category")
    .agg(
        F.count("*").alias("order_count"),
        F.sum("amount").alias("total_sales"),
        F.avg("amount").alias("avg_order_value")
    )
    .filter(F.col("order_count") > 100)
    .orderBy(F.col("total_sales").desc())
)

-- Informatica Expression Transformation
Transformation: EXP_Customer_Calc
Type: Expression

-- Input Ports
customer_id    IN
first_name     IN
last_name      IN
birth_date     IN
annual_income  IN

-- Output Ports (Expressions)
full_name = CONCAT(first_name, ' ', last_name)
age = TRUNC(DATE_DIFF(SYSDATE, birth_date, 'YY'))
income_tier = IIF(annual_income > 100000,
    'HIGH',
    IIF(annual_income > 50000, 'MEDIUM', 'LOW'))

# PySpark Expression Transformation
from pyspark.sql import functions as F
from pyspark.sql.functions import when, concat, lit

customer_calc = (
    source_df
    .withColumn(
        "full_name",
        concat(F.col("first_name"),
               lit(" "),
               F.col("last_name"))
    )
    .withColumn(
        "age",
        F.floor(F.datediff(
            F.current_date(),
            F.col("birth_date")) / 365)
    )
    .withColumn(
        "income_tier",
        when(F.col("annual_income") > 100000, "HIGH")
        .when(F.col("annual_income") > 50000, "MEDIUM")
        .otherwise("LOW")
    )
)

-- Informatica Joiner Transformation
Transformation: JNR_Orders_Customers
Type: Joiner

Master Source: SQ_CUSTOMERS
Detail Source: SQ_ORDERS

Join Type: Normal Join (Inner)

Join Condition:
CUSTOMERS.customer_id = ORDERS.customer_id

-- Output: All ports from both sources

# PySpark Join Transformation
customers_df = spark.table("raw.customers")
orders_df = spark.table("raw.orders")

# Inner Join (Normal Join)
orders_customers = (
    orders_df.alias("orders")
    .join(
        customers_df.alias("customers"),
        F.col("orders.customer_id") ==
        F.col("customers.customer_id"),
        "inner"
    )
    .select(
        "customers.*",
        "orders.order_id",
        "orders.order_date",
        "orders.amount"
    )
)

-- Informatica Aggregator Transformation
Transformation: AGG_Sales_Summary
Type: Aggregator

Group By:
  - region
  - product_category
  - sale_month

Aggregate Functions:
total_sales = SUM(sale_amount)
order_count = COUNT(order_id)
avg_order = AVG(sale_amount)
max_sale = MAX(sale_amount)

Filter: sale_date >= '2024-01-01'

-- models/marts/sales_summary.sql
{{
  config(
    materialized='incremental',
    unique_key=['region', 'product_category', 'sale_month']
  )
}}

SELECT
    region,
    product_category,
    DATE_TRUNC('month', sale_date) AS sale_month,
    SUM(sale_amount) AS total_sales,
    COUNT(order_id) AS order_count,
    AVG(sale_amount) AS avg_order,
    MAX(sale_amount) AS max_sale
FROM {{ ref('stg_sales') }}
WHERE sale_date >= '2024-01-01'
GROUP BY 1, 2, 3

-- Informatica Lookup Transformation
Transformation: LKP_Product_Details
Type: Lookup

Lookup Table: DIM_PRODUCTS
Lookup Condition:
  IN_product_id = PRODUCT_ID

Return Ports:
  - product_name
  - category
  - unit_price

Default Values:
  product_name = 'Unknown'
  category = 'Uncategorized'

-- models/intermediate/int_orders_enriched.sql
{{
  config(materialized='view')
}}

SELECT
    o.*,
    COALESCE(p.product_name, 'Unknown')
        AS product_name,
    COALESCE(p.category, 'Uncategorized')
        AS category,
    p.unit_price
FROM {{ ref('stg_orders') }} o
LEFT JOIN {{ ref('dim_products') }} p
    ON o.product_id = p.product_id

<!-- SSIS Derived Column Transform -->
<component name="DRV_Calculate_Metrics"
    componentClassID="DTSTransform.DerivedColumn">

  <outputColumn
    name="profit_margin"
    expression="(revenue - cost) / revenue * 100"/>

  <outputColumn
    name="full_address"
    expression="street + ', ' + city + ' ' + zip_code"/>

  <outputColumn
    name="order_status"
    expression="shipped_date == NULL ? 'PENDING' : 'SHIPPED'"/>

</component>

# PySpark Derived Columns
from pyspark.sql import functions as F
from pyspark.sql.functions import when, concat

calculated_df = (
    source_df
    .withColumn(
        "profit_margin",
        (F.col("revenue") - F.col("cost"))
        / F.col("revenue") * 100
    )
    .withColumn(
        "full_address",
        concat(
            F.col("street"), F.lit(", "),
            F.col("city"), F.lit(" "),
            F.col("zip_code"))
    )
    .withColumn(
        "order_status",
        when(F.col("shipped_date").isNull(), "PENDING")
        .otherwise("SHIPPED")
    )
)

<!-- SSIS Lookup Transform -->
<component name="LKP_Customer_Info"
    componentClassID="DTSTransform.Lookup">

  <property name="SqlCommand">
    SELECT customer_id, customer_name,
           credit_limit, segment
    FROM dim_customer
    WHERE is_active = 1
  </property>

  <property name="NoMatchBehavior">
    Redirect to No Match Output
  </property>

</component>

# PySpark Lookup with No-Match Handling
# Load lookup reference data
customer_lookup = (
    spark.table("dim_customer")
    .filter(F.col("is_active") == 1)
    .select("customer_id", "customer_name",
            "credit_limit", "segment")
)

# Perform lookup join
matched_df = source_df.join(
    customer_lookup,
    "customer_id",
    "left"
)

# Split matched vs unmatched
matched = matched_df.filter(
    F.col("customer_name").isNotNull())
no_match = matched_df.filter(
    F.col("customer_name").isNull())