Overview

Use Case: Smart Fleet IoT Analytics

What You Will Learn

What You Will Build

Prerequisites

Snowflake CLI Connection: Before running the setup, configure a connection named default:

snow connection add --connection-name default --account <your-account> --user <your-user>

Note: At the time of writing, Spark does not support nanosecond timestamps or geospatial data types in Iceberg tables.

Objects Created in This Guide

Environment Setup

Step 1: Download the Assets

# Clone the sfquickstarts repo and navigate to the assets
git clone https://github.com/Snowflake-Labs/sfquickstarts.git
cd sfquickstarts/site/sfguides/src/iceberg-v3-tables-comprehensive-guide/assets

Step 2: Configure Variables

# Required: Snowflake Account Information
SNOWFLAKE_ACCOUNT="your_account_identifier"
SNOWFLAKE_USER="your_username"
SNOWFLAKE_ROLE="ACCOUNTADMIN"
SNOWFLAKE_WAREHOUSE="FLEET_ANALYTICS_WH"
SNOWFLAKE_DATABASE="FLEET_ANALYTICS_DB"

# External Volume Configuration: Choose ONE option

# OPTION A: Use an existing external volume (if you already have one configured)
USE_EXISTING_VOLUME=false
EXISTING_VOLUME_NAME=""

# OPTION B: Create a new external volume (configure provider below)
STORAGE_PROVIDER="S3"  # Options: S3, GCS, AZURE

# --- AWS S3 ---
S3_BUCKET_NAME="your-bucket-name"
S3_PATH="iceberg/fleet-analytics/"
AWS_ROLE_ARN=""  # Your IAM role ARN (see AWS S3 Setup below)

# --- Google Cloud Storage ---
GCS_BUCKET_NAME="your-bucket-name"
GCS_PATH="iceberg/fleet-analytics/"

# --- Azure Storage (Blob, ADLS Gen2, or OneLake) ---
# See: https://docs.snowflake.com/en/user-guide/tables-iceberg-configure-external-volume-azure
AZURE_TENANT_ID="your-tenant-id"
AZURE_STORAGE_ACCOUNT="your-storage-account"
AZURE_CONTAINER="your-container"
AZURE_PATH="iceberg/fleet-analytics/"
# For OneLake, set USE_ONELAKE=true and provide workspace/lakehouse GUIDs
USE_ONELAKE=false
ONELAKE_WORKSPACE_GUID=""
ONELAKE_LAKEHOUSE_GUID=""

Step 3: Run Setup Script

# Make the script executable
chmod +x setup.sh

# Run the setup
./setup.sh

Step 4: Import the Snowflake Notebook

Note: The notebook is pre-configured with SQL and Python cells that should import with the correct cell types.

External Volume Setup Details

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:GetObjectVersion",
                "s3:PutObject",
                "s3:DeleteObject",
                "s3:ListBucket",
                "s3:GetBucketLocation"
            ],
            "Resource": [
                "arn:aws:s3:::your-bucket-name/*",
                "arn:aws:s3:::your-bucket-name"
            ]
        }
    ]
}

Verify Setup

-- Connect to Snowflake and run:
USE DATABASE FLEET_ANALYTICS_DB;
SHOW ICEBERG TABLES;
SHOW DYNAMIC TABLES;
SHOW MASKING POLICIES;
SHOW DATA METRIC FUNCTIONS;
LIST @LOGS_STAGE;

Streaming and Ingest

Snowpipe Streaming: Real-Time Vehicle Telemetry

Review the Streaming Target Table

USE DATABASE FLEET_ANALYTICS_DB;

-- View the table structure
DESCRIBE TABLE VEHICLE_TELEMETRY_STREAM;

-- Confirm the table exists but is empty
SELECT COUNT(*) FROM VEHICLE_TELEMETRY_STREAM;
-- Returns: 0

-- View the table properties showing it's an Iceberg V3 table with VARIANT column
SELECT 
    GET_DDL('TABLE', 'VEHICLE_TELEMETRY_STREAM');

Start the Streaming Simulation

cd iceberg-v3-tables-comprehensive-guide/assets

# Activate the Python environment (created by setup.sh)
source iceberg_v3_demo_venv/bin/activate

# Start the streaming simulation (runs for 5 minutes max)
python stream_telemetry.py

Troubleshooting Connection Issues: If you're on a VPN or corporate network and get connection errors, you may need a network policy to allow inbound connections. Set ENABLE_NETWORK_POLICY=true in config.env and re-run setup, or manually run 08_network_policy.sql in Snowsight, then apply the policy to your user:

ALTER USER <your_username> SET NETWORK_POLICY = FLEET_STREAMING_POLICY;

# Example of what the streaming script sends
{
    "vehicle_id": "VH-2847",
    "event_timestamp": "2026-02-03T14:32:15.847293847",  # Nanosecond precision
    "telemetry_data": {
        "location": {"lat": 37.7749, "lon": -122.4194},
        "speed_mph": 65.3,
        "heading": 285,
        "engine": {
            "rpm": 3200,
            "temperature_f": 195,
            "oil_pressure_psi": 42,
            "fuel_level_pct": 67.5
        },
        "diagnostics": {
            "check_engine": false,
            "tire_pressure_warning": false,
            "codes": []
        },
        "driver_behavior": {
            "hard_acceleration_count": 2,
            "hard_brake_count": 1,
            "sharp_turn_count": 0
        }
    }
}

Observe Data Flowing In

-- Check current count of streaming events
SELECT COUNT(*) AS event_count FROM VEHICLE_TELEMETRY_STREAM;

-- View the latest streaming events with VARIANT extraction
SELECT 
    VEHICLE_ID,
    EVENT_TIMESTAMP,
    TELEMETRY_DATA:location:lat::FLOAT AS latitude,
    TELEMETRY_DATA:location:lon::FLOAT AS longitude,
    TELEMETRY_DATA:speed_mph::FLOAT AS speed_mph,
    TELEMETRY_DATA:engine:temperature_f::INT AS engine_temp,
    TELEMETRY_DATA:engine:fuel_level_pct::FLOAT AS fuel_level
FROM VEHICLE_TELEMETRY_STREAM
ORDER BY EVENT_TIMESTAMP DESC
LIMIT 10;

Pull from a Web API: Weather Data

Enable External Access in Your Notebook

Note: If you don't see OPEN_METEO_ACCESS, run this SQL first:

CREATE OR REPLACE NETWORK RULE FLEET_ANALYTICS_DB.RAW.OPEN_METEO_NETWORK_RULE
    MODE = EGRESS TYPE = HOST_PORT VALUE_LIST = ('api.open-meteo.com:443');
CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION OPEN_METEO_ACCESS
    ALLOWED_NETWORK_RULES = (FLEET_ANALYTICS_DB.RAW.OPEN_METEO_NETWORK_RULE) ENABLED = TRUE;

View the Target Table

-- The table was created during setup
DESCRIBE TABLE API_WEATHER_DATA;

-- Confirm it's empty
SELECT COUNT(*) AS current_count FROM API_WEATHER_DATA;

Create and Run the API Ingestion Function

# This is pre-populated in the Snowflake Notebook
import snowflake.snowpark as snowpark
from snowflake.snowpark.functions import col, parse_json, current_timestamp
import requests

def fetch_weather_data(session: snowpark.Session, cities: list) -> str:
    """
    Fetches weather data from Open-Meteo API and loads into Iceberg table.
    Open-Meteo is a free, open-source weather API requiring no API key.
    """
    
    weather_records = []
    
    for city in cities:
        # Using Open-Meteo free API
        url = f"https://api.open-meteo.com/v1/forecast"
        params = {
            "latitude": city["lat"],
            "longitude": city["lon"],
            "current": "temperature_2m,wind_speed_10m,precipitation",
            "hourly": "temperature_2m,precipitation_probability"
        }
        
        response = requests.get(url, params=params)
        if response.status_code == 200:
            weather_records.append({
                "city_name": city["name"],
                "latitude": city["lat"],
                "longitude": city["lon"],
                "api_response": response.text,
                "fetched_at": datetime.now().isoformat()
            })
    
    # Create DataFrame and write to Iceberg table
    df = session.create_dataframe(weather_records)
    df = df.with_column("WEATHER_DATA", parse_json(col("API_RESPONSE")))
    df.select(
        col("CITY_NAME"),
        col("LATITUDE"),
        col("LONGITUDE"), 
        col("WEATHER_DATA"),
        current_timestamp().alias("INGESTED_AT")
    ).write.mode("append").save_as_table("API_WEATHER_DATA")
    
    return f"Loaded weather data for {len(weather_records)} cities"

# Define cities for fleet operations
fleet_cities = [
    {"name": "San Francisco", "lat": 37.7749, "lon": -122.4194},
    {"name": "Los Angeles", "lat": 34.0522, "lon": -118.2437},
    {"name": "Seattle", "lat": 47.6062, "lon": -122.3321},
    {"name": "Denver", "lat": 39.7392, "lon": -104.9903},
    {"name": "Chicago", "lat": 41.8781, "lon": -87.6298}
]

# Execute the function
result = fetch_weather_data(session, fleet_cities)
print(result)

Query the Weather Data

-- View the loaded weather data
SELECT 
    CITY_NAME,
    WEATHER_DATA:current:temperature_2m::FLOAT AS current_temp_c,
    WEATHER_DATA:current:wind_speed_10m::FLOAT AS wind_speed_kmh,
    WEATHER_DATA:current:precipitation::FLOAT AS precipitation_mm,
    INGESTED_AT
FROM API_WEATHER_DATA;

-- Extract hourly forecast for a city
SELECT 
    CITY_NAME,
    f.value::FLOAT AS hourly_temp
FROM API_WEATHER_DATA,
LATERAL FLATTEN(input => WEATHER_DATA:hourly:temperature_2m) f
WHERE CITY_NAME = 'San Francisco'
LIMIT 24;

Tip: This ingestion pattern can be orchestrated natively in Snowflake using Tasks or Serverless Tasks for scheduled API pulls.

Batch Ingest JSON Logs

View the Staged Files

-- List the staged JSON files
LIST @LOGS_STAGE;

Load JSON Logs into Iceberg Table

-- Load JSON files into Iceberg table
COPY INTO MAINTENANCE_LOGS (LOG_ID, VEHICLE_ID, LOG_TIMESTAMP, LOG_DATA, SOURCE_FILE)
FROM (
    SELECT 
        $1:log_id::VARCHAR,
        $1:vehicle_id::VARCHAR,
        $1:log_timestamp::TIMESTAMP_NTZ,
        $1,
        METADATA$FILENAME
    FROM @LOGS_STAGE
)
FILE_FORMAT = (TYPE = 'JSON' STRIP_OUTER_ARRAY = TRUE)
PATTERN = '.*maintenance_log.*\.json';

-- View the loaded maintenance logs with VARIANT data
SELECT 
    LOG_ID,
    VEHICLE_ID,
    LOG_TIMESTAMP,
    LOG_DATA:event_type::STRING AS event_type,
    LOG_DATA:severity::STRING AS severity,
    LOG_DATA:description::STRING AS description,
    LOG_DATA:total_cost::FLOAT AS total_cost
FROM MAINTENANCE_LOGS
ORDER BY LOG_TIMESTAMP DESC
LIMIT 10;

Transformation

Dynamic Iceberg Tables Overview

-- View all dynamic tables in the pipeline
SHOW DYNAMIC TABLES IN DATABASE FLEET_ANALYTICS_DB;

Explore the Pipeline in the UI

View the Dynamic Table Graph

Monitor Incremental Refreshes

Query Transformed Data

-- Query enriched telemetry with vehicle details
SELECT 
    VEHICLE_ID,
    MAKE,
    MODEL,
    DRIVER_NAME,
    FLEET_REGION,
    speed_mph,
    engine_temp_f,
    engine_health_status,
    driving_behavior,
    EVENT_TIMESTAMP
FROM CURATED.TELEMETRY_ENRICHED
WHERE engine_health_status != 'NORMAL' OR driving_behavior = 'AGGRESSIVE'
ORDER BY EVENT_TIMESTAMP DESC
LIMIT 20;

Deletion Vectors Overview

Security and Governance

PII Detection with Classification

Classify the Schema

View Applied Tags

Tip: You can also create custom tags that automatically propagate to existing and new downstream objects.

Fine-Grained Access Control with Masking Policies

View Created Masking Policies

-- View all masking policies in the database
SHOW MASKING POLICIES IN DATABASE FLEET_ANALYTICS_DB;

-- View the policy definition
SELECT GET_DDL('MASKING_POLICY', 'PII_MASK');

CREATE OR REPLACE MASKING POLICY PII_MASK AS (val STRING) 
RETURNS STRING ->
    CASE
        WHEN CURRENT_ROLE() IN ('FLEET_ADMIN', 'ACCOUNTADMIN') THEN val
        WHEN CURRENT_ROLE() = 'FLEET_ANALYST' THEN 
            CONCAT(LEFT(val, 2), '****', RIGHT(val, 2))
        ELSE '********'
    END;

Apply Masking Policies via UI

Test Masking Behavior

-- As ACCOUNTADMIN, see full PII values (unmasked)
SELECT VEHICLE_ID, DRIVER_NAME, DRIVER_EMAIL, DRIVER_PHONE
FROM VEHICLE_REGISTRY LIMIT 5;

-- Create a test analyst role and grant it to the current user
CREATE ROLE IF NOT EXISTS FLEET_ANALYST;
GRANT USAGE ON DATABASE FLEET_ANALYTICS_DB TO ROLE FLEET_ANALYST;
GRANT USAGE ON SCHEMA FLEET_ANALYTICS_DB.RAW TO ROLE FLEET_ANALYST;
GRANT SELECT ON ALL TABLES IN SCHEMA FLEET_ANALYTICS_DB.RAW TO ROLE FLEET_ANALYST;
GRANT USAGE ON WAREHOUSE FLEET_ANALYTICS_WH TO ROLE FLEET_ANALYST;

-- Grant the role to the current user so we can switch to it
SET MY_USER = CURRENT_USER();
GRANT ROLE FLEET_ANALYST TO USER IDENTIFIER($MY_USER);

-- Switch to analyst role and observe MASKED values
USE ROLE FLEET_ANALYST;
SELECT VEHICLE_ID, DRIVER_NAME, DRIVER_EMAIL, DRIVER_PHONE
FROM VEHICLE_REGISTRY LIMIT 5;
-- Results show masked values: "Jo****hn", "jo****om"

-- Switch back to ACCOUNTADMIN for remaining operations
USE ROLE ACCOUNTADMIN;

Data Quality Monitoring

View Data Quality Setup

-- Show data metric functions
SHOW DATA METRIC FUNCTIONS IN DATABASE FLEET_ANALYTICS_DB;

CREATE OR REPLACE DATA METRIC FUNCTION CHECK_VALID_COORDINATES(
    arg_t TABLE(lat FLOAT, lon FLOAT)
)
RETURNS NUMBER
AS
$$
    SELECT COUNT(*) 
    FROM arg_t
    WHERE lat < -90 OR lat > 90 
       OR lon < -180 OR lon > 180
$$;

Review Data Quality Results

Note: The Data Quality dashboard will initially be empty. DMFs run on a schedule (default: when triggered by Snowflake). To see results sooner, set a shorter schedule and insert new data to trigger the DMF.

-- Set DMFs to trigger when data changes
ALTER ICEBERG TABLE RAW.VEHICLE_LOCATIONS SET DATA_METRIC_SCHEDULE = 'TRIGGER_ON_CHANGES';
ALTER ICEBERG TABLE RAW.SENSOR_READINGS SET DATA_METRIC_SCHEDULE = 'TRIGGER_ON_CHANGES';

-- Or set a specific schedule (e.g., every 5 minutes)
-- ALTER ICEBERG TABLE RAW.VEHICLE_LOCATIONS SET DATA_METRIC_SCHEDULE = '5 MINUTE';

-- Check which DMFs are attached to VEHICLE_LOCATIONS
SELECT * FROM TABLE(INFORMATION_SCHEMA.DATA_METRIC_FUNCTION_REFERENCES(
    REF_ENTITY_NAME => 'FLEET_ANALYTICS_DB.RAW.VEHICLE_LOCATIONS',
    REF_ENTITY_DOMAIN => 'TABLE'
));

-- Insert a row to trigger the DMF (if using TRIGGER_ON_CHANGES)
INSERT INTO RAW.VEHICLE_LOCATIONS (LOCATION_ID, VEHICLE_ID, LOCATION_TIMESTAMP, LATITUDE, LONGITUDE, LOCATION_POINT, SPEED_MPH, HEADING_DEGREES, FLEET_REGION)
SELECT UUID_STRING(), VEHICLE_ID, CURRENT_TIMESTAMP(), LATITUDE, LONGITUDE, LOCATION_POINT, SPEED_MPH, HEADING_DEGREES, FLEET_REGION
FROM RAW.VEHICLE_LOCATIONS LIMIT 1;

-- Check DMF results (may take a minute to populate after data insert)
SELECT 
    TABLE_NAME,
    METRIC_NAME,
    VALUE,
    MEASUREMENT_TIME
FROM SNOWFLAKE.LOCAL.DATA_QUALITY_MONITORING_RESULTS
WHERE TABLE_DATABASE = 'FLEET_ANALYTICS_DB'
ORDER BY MEASUREMENT_TIME DESC
LIMIT 20;

Tip: Insert or modify data in the table to trigger the DMF if using TRIGGER_ON_CHANGES. The Data Quality tab will populate after the next scheduled run.

Data Lineage

View Lineage Graph

Analytics and AI

Semi-Structured Data Analytics

Query Nested Variant Data

-- Extract nested fields from VARIANT with dot notation
SELECT 
    VEHICLE_ID,
    TELEMETRY_DATA:speed_mph::FLOAT AS speed,
    TELEMETRY_DATA:engine:rpm::INT AS engine_rpm,
    TELEMETRY_DATA:engine:temperature_f::INT AS engine_temp,
    TELEMETRY_DATA:diagnostics:check_engine::BOOLEAN AS check_engine_light,
    TELEMETRY_DATA:driver_behavior:hard_brake_count::INT AS hard_brakes
FROM RAW.VEHICLE_TELEMETRY_STREAM
WHERE TELEMETRY_DATA:speed_mph::FLOAT > 50
LIMIT 20;

-- Aggregate on variant fields - find vehicles with check engine warnings
SELECT 
    TELEMETRY_DATA:diagnostics:check_engine::BOOLEAN AS check_engine,
    COUNT(*) AS event_count,
    ROUND(AVG(TELEMETRY_DATA:speed_mph::FLOAT), 1) AS avg_speed,
    COUNT(DISTINCT VEHICLE_ID) AS vehicle_count
FROM RAW.VEHICLE_TELEMETRY_STREAM
GROUP BY 1
ORDER BY event_count DESC;

Observe Query Pruning

-- Run with profile to see pruning efficiency
SELECT 
    VEHICLE_ID,
    EVENT_TIMESTAMP,
    TELEMETRY_DATA:speed_mph::FLOAT AS speed
FROM VEHICLE_TELEMETRY_STREAM
WHERE VEHICLE_ID = 'VH-1234'
  AND TELEMETRY_DATA:engine:temperature_f::INT > 200;

-- Check query profile in History tab to see partition pruning statistics

Time-Series Analytics and Forecasting

AS OF Join for Point-in-Time Analysis

-- AS OF join: Find closest sensor reading before each maintenance event
SELECT 
    m.VEHICLE_ID,
    m.LOG_TIMESTAMP AS maintenance_time,
    m.LOG_DATA:event_type::STRING AS maintenance_type,
    s.READING_TIMESTAMP AS closest_reading_time,
    s.ENGINE_TEMP_F,
    s.OIL_PRESSURE_PSI
FROM RAW.MAINTENANCE_LOGS m
ASOF JOIN RAW.SENSOR_READINGS s
    MATCH_CONDITION (m.LOG_TIMESTAMP >= s.READING_TIMESTAMP)
    ON m.VEHICLE_ID = s.VEHICLE_ID
WHERE m.LOG_DATA:severity::STRING IN ('CRITICAL', 'HIGH')
ORDER BY m.LOG_TIMESTAMP DESC
LIMIT 15;

Time-Series Aggregation with

-- Calculate rolling averages for engine temperature
SELECT 
    VEHICLE_ID,
    READING_TIMESTAMP,
    ENGINE_TEMP_F,
    AVG(ENGINE_TEMP_F) OVER (
        PARTITION BY VEHICLE_ID 
        ORDER BY READING_TIMESTAMP 
        RANGE BETWEEN INTERVAL '1 HOUR' PRECEDING AND CURRENT ROW
    ) AS rolling_avg_temp,
    MAX(ENGINE_TEMP_F) OVER (
        PARTITION BY VEHICLE_ID 
        ORDER BY READING_TIMESTAMP 
        RANGE BETWEEN INTERVAL '1 HOUR' PRECEDING AND CURRENT ROW
    ) AS rolling_max_temp
FROM RAW.SENSOR_READINGS
ORDER BY VEHICLE_ID, READING_TIMESTAMP
LIMIT 50;

ML Forecasting for Predictive Maintenance

⏱️ Training Time: Model training may take 2-5 minutes depending on data volume and warehouse size. To speed up training:

• Use a larger warehouse (e.g., MEDIUM or LARGE) - training scales with compute
• Reduce the number of time series by filtering to specific vehicles
• Limit historical data range (e.g., 14 days instead of 30)

-- Create a forecast model for fuel consumption
-- For Iceberg tables, we must use SYSTEM$QUERY_REFERENCE instead of TABLE reference
CREATE OR REPLACE SNOWFLAKE.ML.FORECAST fuel_consumption_forecast(
    INPUT_DATA => SYSTEM$QUERY_REFERENCE('
        SELECT 
            VEHICLE_ID,
            READING_TIMESTAMP,
            FUEL_CONSUMPTION_GPH
        FROM RAW.SENSOR_READINGS
        WHERE READING_TIMESTAMP > DATEADD(''day'', -30, CURRENT_TIMESTAMP())
    '),
    TIMESTAMP_COLNAME => 'READING_TIMESTAMP',
    TARGET_COLNAME => 'FUEL_CONSUMPTION_GPH',
    SERIES_COLNAME => 'VEHICLE_ID'
);

-- Generate 7-day forecast and save results
CALL fuel_consumption_forecast!FORECAST(
    FORECASTING_PERIODS => 168,  -- 7 days * 24 hours
    CONFIG_OBJECT => {'prediction_interval': 0.95}
);

-- FORECAST: Generate a 7-day forecast (168 hours) and save to a table for visualization
CALL fuel_consumption_forecast!FORECAST(
    FORECASTING_PERIODS => 168,
    CONFIG_OBJECT => {'prediction_interval': 0.95}
);

-- Save forecast results to a table for visualization
CREATE OR REPLACE TEMPORARY TABLE FUEL_FORECAST_RESULTS AS
SELECT * FROM TABLE(RESULT_SCAN(LAST_QUERY_ID()));

-- Preview forecast results
SELECT * FROM FUEL_FORECAST_RESULTS ORDER BY SERIES, TS LIMIT 20;

Visualize Historical + Forecast Data

# 📈 Visualize Historical + Forecast Fuel Consumption
import streamlit as st
import pandas as pd
from snowflake.snowpark.context import get_active_session

session = get_active_session()

# Get historical data (last 30 days for one vehicle)
historical_df = session.sql("""
    SELECT 
        READING_TIMESTAMP AS ts,
        FUEL_CONSUMPTION_GPH AS value,
        'Historical' AS data_type
    FROM RAW.SENSOR_READINGS
    WHERE VEHICLE_ID = (SELECT SERIES FROM FUEL_FORECAST_RESULTS LIMIT 1)
      AND READING_TIMESTAMP > DATEADD('day', -30, CURRENT_TIMESTAMP())
    ORDER BY READING_TIMESTAMP
""").to_pandas()

# Get forecast data (column names vary - check with DESCRIBE first if needed)
forecast_df = session.sql("""
    SELECT 
        TS AS ts,
        FORECAST AS value,
        'Forecast' AS data_type
    FROM FUEL_FORECAST_RESULTS
    WHERE SERIES = (SELECT SERIES FROM FUEL_FORECAST_RESULTS LIMIT 1)
    ORDER BY TS
""").to_pandas()

# Get the vehicle ID for the title
vehicle_id = session.sql("SELECT SERIES FROM FUEL_FORECAST_RESULTS LIMIT 1").collect()[0][0]

st.subheader(f"📈 Fuel Consumption Forecast: {vehicle_id}")
st.caption("Historical data (30 days) + 7-day forecast")

# Normalize column names to lowercase
historical_df.columns = historical_df.columns.str.lower()
forecast_df.columns = forecast_df.columns.str.lower()

# Combine for plotting
combined_df = pd.concat([historical_df, forecast_df], ignore_index=True)
combined_df['ts'] = pd.to_datetime(combined_df['ts'])
combined_df = combined_df.sort_values('ts')

# Create the chart using Streamlit's native line chart
# Pivot data for multi-line chart
chart_data = combined_df.pivot_table(index='ts', columns='data_type', values='value', aggfunc='first')
st.line_chart(chart_data, use_container_width=True)

# Display metrics
col1, col2, col3 = st.columns(3)
col1.metric("Avg Historical (GPH)", f"{historical_df['value'].mean():.2f}")
col2.metric("Avg Forecast (GPH)", f"{forecast_df['value'].mean():.2f}")
change = ((forecast_df['value'].mean() - historical_df['value'].mean()) / historical_df['value'].mean()) * 100
col3.metric("Projected Change", f"{change:+.1f}%")

st.caption("💡 The forecast helps identify vehicles likely to have higher fuel consumption, enabling proactive maintenance scheduling.")

Geospatial Analytics

Basic Geospatial Queries

-- Calculate distance between consecutive positions
SELECT 
    VEHICLE_ID,
    LOCATION_TIMESTAMP,
    LAG(LOCATION_POINT) OVER (
        PARTITION BY VEHICLE_ID ORDER BY LOCATION_TIMESTAMP
    ) AS prev_location,
    LOCATION_POINT AS current_location,
    ST_DISTANCE(
        LOCATION_POINT,
        LAG(LOCATION_POINT) OVER (PARTITION BY VEHICLE_ID ORDER BY LOCATION_TIMESTAMP)
    ) / 1609.34 AS distance_miles  -- Convert meters to miles
FROM VEHICLE_LOCATIONS
ORDER BY VEHICLE_ID, LOCATION_TIMESTAMP
LIMIT 50;

Geofencing Analysis

-- Define a geofence (San Francisco downtown)
SET sf_downtown = ST_MAKEPOLYGON(TO_GEOGRAPHY(
    'LINESTRING(-122.42 37.78, -122.42 37.79, -122.40 37.79, -122.40 37.78, -122.42 37.78)'
));

-- Find vehicles inside the geofence
SELECT 
    VEHICLE_ID,
    LOCATION_TIMESTAMP,
    ST_X(LOCATION_POINT) AS longitude,
    ST_Y(LOCATION_POINT) AS latitude
FROM VEHICLE_LOCATIONS
WHERE ST_WITHIN(LOCATION_POINT, $sf_downtown)
ORDER BY LOCATION_TIMESTAMP DESC
LIMIT 20;

-- Geospatial: Aggregate vehicles by H3 cell (resolution 6)
SELECT 
    H3_POINT_TO_CELL_STRING(LOCATION_POINT, 6) AS h3_cell,
    FLEET_REGION,
    COUNT(DISTINCT VEHICLE_ID) AS vehicle_count,
    ROUND(AVG(SPEED_MPH), 1) AS avg_speed
FROM RAW.VEHICLE_LOCATIONS
WHERE LOCATION_TIMESTAMP > DATEADD('hour', -24, CURRENT_TIMESTAMP())
GROUP BY 1, 2
ORDER BY vehicle_count DESC
LIMIT 20;

Interactive Map Visualization with Streamlit

# Visualize fleet locations on an interactive map using Streamlit
import streamlit as st
from snowflake.snowpark.context import get_active_session

session = get_active_session()

# Query location data with H3 cell centers (use ST_X/ST_Y for GEOGRAPHY type)
df = session.sql("""
    SELECT 
        H3_POINT_TO_CELL_STRING(LOCATION_POINT, 6) AS h3_cell,
        ST_X(H3_CELL_TO_POINT(H3_POINT_TO_CELL_STRING(LOCATION_POINT, 6))) AS longitude,
        ST_Y(H3_CELL_TO_POINT(H3_POINT_TO_CELL_STRING(LOCATION_POINT, 6))) AS latitude,
        FLEET_REGION,
        COUNT(DISTINCT VEHICLE_ID) AS vehicle_count,
        ROUND(AVG(SPEED_MPH), 1) AS avg_speed
    FROM RAW.VEHICLE_LOCATIONS
    WHERE LOCATION_TIMESTAMP > DATEADD('hour', -24, CURRENT_TIMESTAMP())
    GROUP BY 1, 2, 3, 4
    ORDER BY vehicle_count DESC
    LIMIT 100
""").to_pandas()

st.subheader("🗺️ Fleet Vehicle Distribution")
st.caption("Each point represents an H3 cell with vehicle activity in the last 24 hours")

# Display metrics
col1, col2, col3 = st.columns(3)
col1.metric("Total H3 Cells", len(df))
col2.metric("Total Vehicles", int(df['VEHICLE_COUNT'].sum()))
col3.metric("Avg Speed", f"{df['AVG_SPEED'].mean():.1f} mph")

# Interactive map visualization - point size scales with vehicle count
st.map(df, latitude='LATITUDE', longitude='LONGITUDE', size='VEHICLE_COUNT')

# Show data table
st.dataframe(df[['FLEET_REGION', 'VEHICLE_COUNT', 'AVG_SPEED']].head(10))

AI Agents with Snowflake Intelligence

Note: Cortex Agents require Enterprise Edition or higher.

Region Availability and Cross-Region Inference

None of the preferred models are authorized or available in your region...

-- Enable cross-region inference for your account
USE ROLE ACCOUNTADMIN;
ALTER ACCOUNT SET CORTEX_ENABLED_CROSS_REGION = 'ANY_REGION';

Privacy Note: With cross-region inference, only the AI inference is routed cross-region. Your underlying data remains in your account's region and is protected by Snowflake's governance controls.

Agent Configuration

Note: If the agent creation fails due to syntax issues, you can configure it manually through Cortex AI → Agents → FLEET_ANALYTICS_AGENT.

Using the Fleet Analytics Agent

Agent Configuration

-- View agent details
SHOW AGENTS IN SCHEMA FLEET_ANALYTICS_DB.RAW;

-- Describe the agent
DESCRIBE AGENT FLEET_ANALYTICS_DB.RAW.FLEET_ANALYTICS_AGENT;

Business Continuity and Disaster Recovery

Note: Replication features require Business Critical Edition or higher.

Iceberg Table Replication Overview

Setting Up Replication via UI

Step 1: Create a Replication Group

Step 2: Select Target Account

Step 3: Configure Objects

Step 4: Enable and Monitor

Verify Replication

-- Check that Iceberg tables exist
USE DATABASE FLEET_ANALYTICS_DB;
SHOW ICEBERG TABLES;

-- Query replicated data
SELECT COUNT(*) FROM VEHICLE_TELEMETRY_STREAM;
SELECT COUNT(*) FROM MAINTENANCE_LOGS;

Failover Testing

Interoperability

Prerequisites for Spark

Start the Spark Environment

cd sfguide-iceberg-v3-comprehensive-guide/assets

# Activate the Spark environment
conda activate fleet-spark

# Start Jupyter notebook
jupyter notebook spark_iceberg_interop.ipynb

Spark Notebook Walkthrough

Cell 1: Setup and Imports

from pyspark.sql import SparkSession
import os

# Load configuration (from the same config.env used during setup)
from dotenv import load_dotenv
load_dotenv('../config.env')

SNOWFLAKE_ACCOUNT = os.getenv('SNOWFLAKE_ACCOUNT')
SNOWFLAKE_USER = os.getenv('SNOWFLAKE_USER')

Cell 2: Create Spark Session with Horizon Catalog

from pyspark.sql import SparkSession

# Configuration
SNOWFLAKE_PASSWORD = os.getenv('SNOWFLAKE_PASSWORD')
SNOWFLAKE_ROLE = 'ACCOUNTADMIN'
SNOWFLAKE_SCHEMA = 'RAW'
SNOWFLAKE_WAREHOUSE = os.getenv('SNOWFLAKE_WAREHOUSE', 'FLEET_ANALYTICS_WH')
SF_URL = f"{SNOWFLAKE_ACCOUNT_URL}.snowflakecomputing.com"

# Versions - Note: Spark 3.5 required for Snowflake Connector masking support
SNOWFLAKE_JDBC_VERSION = "3.24.0"
SNOWFLAKE_SPARK_CONNECTOR_VERSION = "3.1.6"

# Create Spark session with Iceberg and Snowflake catalog configuration
# Note: 
# - driver.host and bindAddress ensure Spark uses localhost (avoids VPN issues)
# - Using Spark 4.0 + Iceberg 1.10.1 for native VARIANT support
spark = SparkSession.builder \
    .appName("Fleet Analytics - Iceberg V3 Interop") \
    .master("local[*]") \
    .config("spark.driver.host", "127.0.0.1") \
    .config("spark.driver.bindAddress", "127.0.0.1") \
    .config("spark.jars.packages", 
            f"org.apache.iceberg:iceberg-spark-runtime-4.0_{SCALA_VERSION}:{ICEBERG_VERSION},"
            f"org.apache.iceberg:iceberg-aws-bundle:{ICEBERG_VERSION},"
            f"net.snowflake:snowflake-jdbc:{SNOWFLAKE_JDBC_VERSION},"
            f"net.snowflake:spark-snowflake_{SCALA_VERSION}:{SNOWFLAKE_SPARK_CONNECTOR_VERSION}") \
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.horizon", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.horizon.type", "rest") \
    .config("spark.sql.catalog.horizon.uri", f"https://{SNOWFLAKE_ACCOUNT_URL}.snowflakecomputing.com/polaris/api/catalog") \
    .config("spark.sql.catalog.horizon.credential", SNOWFLAKE_ACCOUNTADMIN_TOKEN) \
    .config("spark.sql.catalog.horizon.warehouse", SNOWFLAKE_DATABASE) \
    .config("spark.sql.catalog.horizon.scope", f"session:role:{SNOWFLAKE_ROLE}") \
    .config("spark.sql.catalog.horizon.header.X-Iceberg-Access-Delegation","vended-credentials") \
    .config("spark.snowflake.sfURL", SF_URL) \
    .config("spark.snowflake.sfUser", os.getenv('SNOWFLAKE_USER')) \
    .config("spark.snowflake.sfPassword", SNOWFLAKE_FLEET_ANALYST_TOKEN) \
    .config("spark.snowflake.sfDatabase", SNOWFLAKE_DATABASE) \
    .config("spark.snowflake.sfSchema", SNOWFLAKE_SCHEMA) \
    .config("spark.snowflake.sfRole", SNOWFLAKE_ROLE) \
    .config("spark.snowflake.sfWarehouse", SNOWFLAKE_WAREHOUSE) \
    .config("spark.sql.iceberg.vectorization.enabled", "false") \
    .getOrCreate()

print("Spark session created successfully!")
print(f"Spark version: {spark.version}")

Cell 3: List Available Tables

# Show all tables visible to Spark
spark.sql("SHOW TABLES IN horizon.RAW").show(truncate=False)
spark.sql("SHOW TABLES IN horizon.CURATED").show(truncate=False)
spark.sql("SHOW TABLES IN horizon.ANALYTICS").show(truncate=False)

+---------+------------------------+-----------+
|namespace|tableName               |isTemporary|
+---------+------------------------+-----------+
|RAW      |API_WEATHER_DATA        |false      |
|RAW      |MAINTENANCE_LOGS        |false      |
|RAW      |SENSOR_READINGS         |false      |
|RAW      |VEHICLE_LOCATIONS       |false      |
|RAW      |VEHICLE_REGISTRY        |false      |
|RAW      |VEHICLE_TELEMETRY_STREAM|false      |
+---------+------------------------+-----------+

+---------+--------------------+-----------+
|namespace|tableName           |isTemporary|
+---------+--------------------+-----------+
|CURATED  |MAINTENANCE_ANALYSIS|false      |
|CURATED  |TELEMETRY_ENRICHED  |false      |
+---------+--------------------+-----------+

+---------+--------------------+-----------+
|namespace|tableName           |isTemporary|
+---------+--------------------+-----------+
|ANALYTICS|DAILY_FLEET_SUMMARY |false      |
|ANALYTICS|VEHICLE_HEALTH_SCORE|false      |
+---------+--------------------+-----------+

Cell 4: Describe Table with Variant Column

# See variant column in Iceberg table
spark.sql("DESCRIBE TABLE snowflake.RAW.VEHICLE_TELEMETRY_STREAM").show(truncate=False)

+-------------------+---------------+-------+
|col_name           |data_type      |comment|
+-------------------+---------------+-------+
|VEHICLE_ID         |string         |null   |
|EVENT_TIMESTAMP    |timestamp      |null   |
|TELEMETRY_DATA     |variant        |null   |
+-------------------+---------------+-------+

Cell 5: Query Data with Variant

# Use Spark's variant_get() function to extract nested fields from VARIANT
# This query succeeds, and no Snowflake compute is used, since there's no masking policy on the Iceberg table.
df = spark.sql(f"""
    SELECT 
        VEHICLE_ID,
        EVENT_TIMESTAMP,
        variant_get(TELEMETRY_DATA, '$.speed_mph', 'float') AS speed_mph,
        variant_get(TELEMETRY_DATA, '$.engine.temperature_f', 'int') AS engine_temp,
        variant_get(TELEMETRY_DATA, '$.location.lat', 'float') AS latitude,
        variant_get(TELEMETRY_DATA, '$.location.lon', 'float') AS longitude
    FROM snowflake.RAW.VEHICLE_TELEMETRY_STREAM
    WHERE variant_get(TELEMETRY_DATA, '$.speed_mph', 'float') > 60
    LIMIT 10
""")
df.show()

Cell 6: Demonstrate Access Control (Masking Enforcement)

# To enforce masking policies from Spark, we need the Snowflake Connector for Spark
# which routes queries through Snowflake for policy evaluation
# See: https://docs.snowflake.com/en/user-guide/tables-iceberg-query-using-external-query-engine-snowflake-horizon-enforce-access-policies

# Stop the previous Spark session first
spark.stop()
from pyspark.sql import SparkSession

# Configuration
SNOWFLAKE_PASSWORD = os.getenv('SNOWFLAKE_PASSWORD')
SNOWFLAKE_ROLE = 'FLEET_ANALYST'
SNOWFLAKE_SCHEMA = 'RAW'
SNOWFLAKE_WAREHOUSE = os.getenv('SNOWFLAKE_WAREHOUSE', 'FLEET_ANALYTICS_WH')
SF_URL = f"{SNOWFLAKE_ACCOUNT_URL}.snowflakecomputing.com"

# Versions - Note: Spark 3.5 required for Snowflake Connector masking support
SNOWFLAKE_JDBC_VERSION = "3.24.0"
SNOWFLAKE_SPARK_CONNECTOR_VERSION = "3.1.6"

spark_analyst = SparkSession.builder \
    .appName("Fleet Analytics - Analyst View") \
    .master("local[*]") \
    .config("spark.driver.host", "127.0.0.1") \
    .config("spark.driver.bindAddress", "127.0.0.1") \
    .config("spark.jars.packages", 
            f"org.apache.iceberg:iceberg-spark-runtime-4.0_{SCALA_VERSION}:{ICEBERG_VERSION},"
            f"org.apache.iceberg:iceberg-aws-bundle:{ICEBERG_VERSION},"
            f"net.snowflake:snowflake-jdbc:{SNOWFLAKE_JDBC_VERSION},"
            f"net.snowflake:spark-snowflake_{SCALA_VERSION}:{SNOWFLAKE_SPARK_CONNECTOR_VERSION}") \
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.defaultCatalog", "horizon") \
    .config("spark.sql.catalog.horizon", "org.apache.spark.sql.snowflake.catalog.SnowflakeFallbackCatalog") \
    .config("spark.sql.catalog.horizon.catalog-impl", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.horizon.type", "rest") \
    .config("spark.sql.catalog.horizon.uri", f"https://{SNOWFLAKE_ACCOUNT_URL}.snowflakecomputing.com/polaris/api/catalog") \
    .config("spark.sql.catalog.horizon.warehouse", SNOWFLAKE_DATABASE) \
    .config("spark.sql.catalog.horizon.scope", f"session:role:{SNOWFLAKE_ROLE}") \
    .config("spark.sql.catalog.horizon.credential", SNOWFLAKE_FLEET_ANALYST_TOKEN) \
    .config("spark.sql.catalog.horizon.io-impl", "org.apache.iceberg.aws.s3.S3FileIO") \
    .config("spark.sql.catalog.horizon.header.X-Iceberg-Access-Delegation", "vended-credentials") \
    .config("spark.snowflake.sfURL", SF_URL) \
    .config("spark.snowflake.sfUser", os.getenv('SNOWFLAKE_USER')) \
    .config("spark.snowflake.sfPassword", SNOWFLAKE_FLEET_ANALYST_TOKEN) \
    .config("spark.snowflake.sfDatabase", SNOWFLAKE_DATABASE) \
    .config("spark.snowflake.sfSchema", SNOWFLAKE_SCHEMA) \
    .config("spark.snowflake.sfRole", SNOWFLAKE_ROLE) \
    .config("spark.snowflake.sfWarehouse", SNOWFLAKE_WAREHOUSE) \
    .config("spark.sql.iceberg.vectorization.enabled", "false") \
    .getOrCreate()

spark_analyst.sparkContext.setLogLevel("ERROR")

spark_analyst.sql("SHOW TABLES in horizon.RAW").show(truncate=False)

spark_analyst.sql("""
    SELECT
        VEHICLE_ID,
        MAKE,
        MODEL,
        YEAR,
        LICENSE_PLATE,
        DRIVER_NAME,
        DRIVER_EMAIL,
        DRIVER_PHONE,
        FLEET_REGION
    FROM horizon.RAW.VEHICLE_REGISTRY
""").show(truncate=True)

Cleanup

USE ROLE ACCOUNTADMIN;

-- Drop the database (this removes all tables, schemas, and objects)
DROP DATABASE IF EXISTS FLEET_ANALYTICS_DB;

-- Drop the external volume (if you created one)
DROP EXTERNAL VOLUME IF EXISTS FLEET_ICEBERG_VOL;

-- Drop the warehouse
DROP WAREHOUSE IF EXISTS FLEET_ANALYTICS_WH;

-- Drop roles
DROP ROLE IF EXISTS FLEET_ANALYST;
DROP ROLE IF EXISTS FLEET_ENGINEER;
DROP ROLE IF EXISTS FLEET_ADMIN;

-- If you created replication groups, remove them
DROP REPLICATION GROUP IF EXISTS FLEET_ANALYTICS_REPLICATION;

-- Remove network policy from user (if applied)
ALTER USER <your_username> UNSET NETWORK_POLICY;
DROP NETWORK POLICY IF EXISTS FLEET_STREAMING_POLICY;

-- External access integration
DROP INTEGRATION IF EXISTS OPEN_METEO_ACCESS;

SELECT 'Uncomment the commands above to cleanup' AS NOTE;

Conclusion and Resources

What You Learned

Key Benefits of Iceberg V3 Tables in Snowflake

Related Resources