GreenCart — Pipeline Snowflake

1

Upsert MERGE — staging.customers Auto-generado · snowflake_query_helper.py

staging_customers_merge.sql

python3 scripts/snowflake_query_helper.py merge --target customers --source stg_customers --key customer_id --columns company_name,plan,country,mrr_eur --schema staging

MERGE INTO staging.customers t
USING staging.stg_customers s
    ON t.customer_id = s.customer_id
WHEN MATCHED THEN
    UPDATE SET
        t.company_name  = s.company_name,
        t.plan          = s.plan,
        t.country       = s.country,
        t.mrr_eur       = s.mrr_eur,
        t.updated_at    = CURRENT_TIMESTAMP()
WHEN NOT MATCHED THEN
    INSERT (customer_id, company_name, plan, country, mrr_eur, updated_at)
    VALUES (s.customer_id, s.company_name, s.plan, s.country, s.mrr_eur, CURRENT_TIMESTAMP());

2

Dynamic Table — analytics.orders_metrics lag: 5 min · refresh incremental

orders_metrics_dt.sql

python3 scripts/snowflake_query_helper.py dynamic-table --name orders_metrics --warehouse transform_wh --lag "5 minutes" --source staging.orders ...

CREATE OR REPLACE DYNAMIC TABLE analytics.orders_metrics
    TARGET_LAG = '5 minutes'
    WAREHOUSE  = transform_wh
    AS
    SELECT
        customer_id,
        COUNT(*)        AS total_orders,
        SUM(total_eur)   AS gmv_eur,
        MAX(created_at)  AS last_order_at
    FROM staging.orders
    WHERE status != 'cancelled';

-- Verificar modo de refresco (incremental es el preferido):
-- SELECT name, refresh_mode, refresh_mode_reason
-- FROM TABLE(INFORMATION_SCHEMA.DYNAMIC_TABLES())
-- WHERE name = 'ORDERS_METRICS';

3

Cortex AI — Clasificación automática de tickets de soporte AI_CLASSIFY · AI_SENTIMENT

🤖 Dynamic Table con enriquecimiento AI

CREATE OR REPLACE DYNAMIC TABLE analytics.tickets_enriched
    TARGET_LAG = '15 minutes'
    WAREHOUSE  = transform_wh
    AS
    SELECT
        ticket_id,
        customer_id,
        message,
        created_at,
        -- Clasificacion de categoria (billing, bug, feature_request, onboarding, other)
        AI_CLASSIFY(message, ['billing', 'bug', 'feature_request', 'onboarding', 'other']):label::STRING AS category,
        -- Score de sentimiento (-1 negativo a +1 positivo)
        AI_SENTIMENT(message):score::FLOAT AS sentiment_score,
        -- Extraccion de entidades clave
        AI_EXTRACT(message, {'product_sku': 'string', 'urgency': 'string'}) AS extracted_entities
    FROM staging.support_tickets
    WHERE category IS NULL; -- Solo nuevos sin clasificar

Referencia de funciones Cortex AI (nombres vigentes)

Funcion actual	Proposito	NO usar (deprecated)
`AI_COMPLETE`	LLM completion (texto, imagenes, documentos)	COMPLETE
`AI_CLASSIFY`	Clasificar texto en categorias (hasta 500 etiquetas)	CLASSIFY_TEXT
`AI_SENTIMENT`	Score de sentimiento de -1 a 1	SENTIMENT
`AI_EXTRACT`	Extraccion estructurada de texto/imagenes	EXTRACT_ANSWER
`AI_FILTER`	Filtro booleano sobre texto o imagenes	—
`AI_PARSE_DOCUMENT`	OCR o extraccion de layout de documentos	PARSE_DOCUMENT

4

RBAC Least-Privilege — Grants auto-generados 2 roles · 3 schemas

grants_cs_analyst.sql --role cs_analyst_role --privileges SELECT

-- RBAC grants for role: cs_analyst_role
-- Principio: minimo privilegio

GRANT USAGE ON DATABASE
    GREENCART_PROD TO ROLE cs_analyst_role;

-- Schema: analytics (solo lectura)
GRANT USAGE ON SCHEMA
    GREENCART_PROD.analytics TO ROLE cs_analyst_role;
GRANT SELECT ON ALL TABLES IN SCHEMA
    GREENCART_PROD.analytics TO ROLE cs_analyst_role;
GRANT SELECT ON FUTURE TABLES IN SCHEMA
    GREENCART_PROD.analytics TO ROLE cs_analyst_role;
GRANT SELECT ON ALL VIEWS IN SCHEMA
    GREENCART_PROD.analytics TO ROLE cs_analyst_role;
GRANT SELECT ON FUTURE VIEWS IN SCHEMA
    GREENCART_PROD.analytics TO ROLE cs_analyst_role;

grants_etl_role.sql --role etl_role --schemas raw,staging --privileges SELECT,INSERT,UPDATE

-- RBAC grants for role: etl_role

GRANT USAGE ON DATABASE
    GREENCART_PROD TO ROLE etl_role;

-- Schema: raw (carga de datos)
GRANT USAGE   ON SCHEMA GREENCART_PROD.raw TO ROLE etl_role;
GRANT SELECT  ON ALL TABLES IN SCHEMA GREENCART_PROD.raw TO ROLE etl_role;
GRANT INSERT  ON ALL TABLES IN SCHEMA GREENCART_PROD.raw TO ROLE etl_role;
GRANT UPDATE  ON ALL TABLES IN SCHEMA GREENCART_PROD.raw TO ROLE etl_role;
GRANT SELECT  ON FUTURE TABLES IN SCHEMA GREENCART_PROD.raw TO ROLE etl_role;
GRANT INSERT  ON FUTURE TABLES IN SCHEMA GREENCART_PROD.raw TO ROLE etl_role;
GRANT UPDATE  ON FUTURE TABLES IN SCHEMA GREENCART_PROD.raw TO ROLE etl_role;

-- Schema: staging (transformaciones)
GRANT USAGE   ON SCHEMA GREENCART_PROD.staging TO ROLE etl_role;
GRANT SELECT  ON ALL TABLES IN SCHEMA GREENCART_PROD.staging TO ROLE etl_role;
GRANT INSERT  ON ALL TABLES IN SCHEMA GREENCART_PROD.staging TO ROLE etl_role;
GRANT UPDATE  ON ALL TABLES IN SCHEMA GREENCART_PROD.staging TO ROLE etl_role;

5

Anti-patrones detectados en el codigo inicial del cliente 3 issues corregidos

⚠️

Missing colon prefix en stored procedure — SELECT name INTO result FROM customers WHERE id = p_id;
Causa: "invalid identifier" en runtime. Correccion: SELECT name INTO :result FROM customers WHERE id = :p_id;

⚠️

Funcion Cortex deprecated — El codigo usaba CLASSIFY_TEXT(message, ...).
Reemplazado por AI_CLASSIFY(message, [...]) — la nueva API devuelve { label, score }.

⚠️

Task creada pero no reanudada — CREATE TASK process_orders ... sin RESUME.
Las tasks en Snowflake se crean en estado SUSPENDED. Siempre añadir: ALTER TASK process_orders RESUME;

6

Configuracion dbt — Materializaciones para Snowflake

models/analytics/orders_metrics.sql dbt run --select orders_metrics

{{
  config(
    materialized='dynamic_table',
    snowflake_warehouse='transform_wh',
    target_lag='5 minutes',
    transient=true,
    copy_grants=true,
    query_tag='greencart_cs_team'
  )
}}

SELECT
    o.customer_id,
    c.company_name,
    c.plan,
    COUNT(o.order_id)         AS total_orders,
    SUM(o.total_eur)           AS gmv_eur,
    AVG(o.total_eur)           AS avg_order_eur,
    MAX(o.created_at)          AS last_order_at,
    COUNT(CASE WHEN o.status = 'returned'
               THEN 1 END)     AS returned_orders
FROM staging.orders o
JOIN staging.customers c USING (customer_id)
WHERE o.status != 'cancelled'
GROUP BY 1, 2, 3

GreenCart — Pipeline de Datos Snowflake

Arquitectura del pipeline — GREENCART_PROD

🤖 Dynamic Table con enriquecimiento AI