Snowflake-Labs/openflow-structured-data-pipeline-demo

Snowflake-Labs/openflow-structured-data-pipeline-demo

Description: A comprehensive demonstration of Snowflake's OpenFlow platform showcasing intelligent data ingestion, schema evolution, and real-time analytics for music festival and concert data

Language: Python

License: Apache-2.0

Stars: 2

Forks: 4

Open issues: 0

Created: 2025-09-19T06:19:49Z

Pushed: 2025-11-04T16:07:13Z

Default branch: main

Fork: no

Archived: no

README:

🎵 MusicFlow - Snowflake OpenFlow Demo

A comprehensive demonstration of Snowflake's OpenFlow platform showcasing intelligent data ingestion, schema evolution, and real-time analytics for music festival and concert data. This demo leverages Apache NiFi, Snowflake Cortex AI, and Apache Iceberg to create an end-to-end data pipeline that automatically detects schema changes and evolves data structures.

🎬 Demo Video


Watch the demo in action: MusicFlow - Intelligent Data Pipelines for Music Festival Analytics

🌟 Overview

The Problem We Solve

Building data pipelines for every new data source can be a real headache, taking up tons of time and money – sometimes even months! Even when the info is similar, different systems have totally different setups. This means what should be a quick job turns into months of intense, manual engineering.

Our Solution

What if all that manual data wrangling just poof, disappeared? With Snowflake OpenFlow, we can whip up smart data pipelines that tap into Cortex to figure out schemas from any incoming data – messy or neat. And thanks to Snowflake OpenCatalog, schema changes are a breeze, instantly building unified Iceberg analytics tables.

MusicFlow demonstrates how to build intelligent data pipelines that can:

• Automatically ingest CSV files from various music festival sources (SoundWave Festival, Harmony Grove, Beat Valley, Music Mountain, Coastal Beats)
• Intelligently detect schema changes using Snowflake Cortex AI
• Dynamically evolve Iceberg table schemas without manual intervention
• Provide real-time schema registry monitoring and analytics
• Support multiple data sources with unified schema mapping

Key Features

• 🤖 AI-Powered Schema Intelligence: Uses Snowflake Cortex to analyze CSV structure and determine schema evolution needs
• 🔄 Dynamic Schema Evolution: Automatically adds new columns when new data fields are detected
• 📊 Real-time Monitoring: Streamlit-based schema registry dashboard for pipeline visibility
• 🏗️ External REST Catalog: Integrates with Snowflake OpenCatalog for Iceberg table management
• 🎯 Multi-Source Support: Handles diverse festival data formats with semantic field mapping

🏗️ Architecture

Data Flow

---
config:
layout: dagre
---
flowchart TD
A["Google Shared Drive Folder"] --> B["CSV Files"]
D["Schema Analysis"] --> E{"Cortex AI"}
F["S3 Bucket"] --> I["Load into Iceberg Table"]
G["Schema Evolution"] --> H["ALTER TABLE"]
H -- Move to --> F
B --> D
E --> n1["New Schema"] & G
n1 --> n2["CREATE"]
n2 -- Move to --> F
n1@{ shape: rect}
n2@{ shape: rect}

Key Components

1. Data Sources: Festival CSV files (SoundWave Festival, Harmony Grove, Beat Valley, etc.) 2. Ingestion Layer: Snowflake OpenFlow with NiFi flows for file processing and routing 3. AI Analysis: Snowflake Cortex for intelligent schema detection 4. Storage Layer: Apache Iceberg tables via Snowflake OpenCatalog 5. Monitoring: Streamlit dashboard for pipeline visibility 6. Schema Registry: Metadata tracking and version management

🛠️ Prerequisites

Required Tools & Software

| Tool | Version | Purpose | Installation | |------|---------|---------|--------------| | Snowflake CLI | Latest | Database operations and SPCS management | Install Guide | | Polaris CLI | Commit 52e30f0 | OpenCatalog management and operations | Clone Repository | | Python | 3.12+ | Application runtime | Python Downloads | | jq | Latest | JSON processing | brew install jq (macOS) or Download | | Task | Latest | Task runner | Taskfile.dev |

Snowflake Requirements

> ⚠️ IMPORTANT: Snowflake Trials are NOT supported for this demo. You need a full Snowflake account with the following capabilities:

• Snowflake Account with Enterprise Edition (Production/Full Account Required)
• Snowpark Container Services (SPCS) enabled and configured
• OpenFlow enabled and accessible
• Cortex AI access for schema analysis
• Snowflake OpenCatalog enabled and configured for Iceberg tables

Snowflake OpenCatalog Prerequisites

Before setting up the MusicFlow demo, you need to configure Snowflake OpenCatalog:

1. OpenCatalog Account: Create a Snowflake OpenCatalog account 2. Catalog Configuration: Set up a catalog with appropriate storage locations 3. Service Connection: Configure service connections and credentials 4. Snowflake Integration: Link OpenCatalog to your Snowflake account

AWS Prerequisites

• AWS CLI configured with credentials
• S3 Bucket for data ingestion staging your-music-flow-demo-ingest-data.
• AWS IAM User with appropriate permissions for S3 to be used with Snowflake Openflow to move file to data ingestion bucket.
• IAM Role with appropriate permissions for S3 and OpenCatalog access

Slack Prerequisites

• Slack channel for notifications - Slack bot oauth token for notifications and a channel to send notifications to.

🚀 Quick Start

1. Environment Setup

# Clone the repository
git clone
cd openflow-structured-data-pipeline-demo

# Install Python dependencies using uv
uv sync

# Set up environment variables
cp env.template .env
# Edit .env with your Snowflake and AWS credentials

2. Snowflake Configuration

# Login to Snowflake
snow login

# Setup Snowflake environment (creates database, warehouse, role, and user)
task setup

# Verify environment setup
task env_check

3. Snowflake OpenCatalog Setup

> 💡 Note: It's recommended to perform OpenCatalog setup in a new terminal session to avoid conflicts with existing...