MACRO NEPAL
Data Warehouse with Snowflake in Java: Complete Implementation
awiskar acharya
November 17, 2025
10:23 am

Introduction to Snowflake Data Warehouse

Snowflake is a cloud-native data warehouse that provides seamless scalability, performance, and ease of use. This guide covers comprehensive Java integration with Snowflake for building robust data warehouse solutions.

Table of Contents

  1. Snowflake Setup and Configuration
  2. JDBC Integration
  3. Data Ingestion Patterns
  4. Data Modeling and Transformation
  5. Query Optimization
  6. Monitoring and Management
  7. Spring Boot Integration
  8. Real-World Use Cases

1. Project Setup and Dependencies

Maven Configuration

<properties>
<snowflake.jdbc.version>3.13.22</snowflake.jdbc.version>
<spring-boot.version>3.1.0</spring-boot.version>
<jackson.version>2.15.2</jackson.version>
</properties>
<dependencies>
<!-- Snowflake JDBC Driver -->
<dependency>
<groupId>net.snowflake</groupId>
<artifactId>snowflake-jdbc</artifactId>
<version>${snowflake.jdbc.version}</version>
</dependency>
<!-- Spring Boot -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<version>${spring-boot.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-jpa</artifactId>
<version>${spring-boot.version}</version>
</dependency>
<!-- JSON Processing -->
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>${jackson.version}</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.datatype</groupId>
<artifactId>jackson-datatype-jsr310</artifactId>
<version>${jackson.version}</version>
</dependency>
<!-- Apache Commons -->
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.13.0</version>
</dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<version>2.13.0</version>
</dependency>
<!-- Connection Pool -->
<dependency>
<groupId>com.zaxxer</groupId>
<artifactId>HikariCP</artifactId>
<version>5.0.1</version>
</dependency>
<!-- Lombok -->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<!-- Testing -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<version>${spring-boot.version}</version>
<scope>test</scope>
</dependency>
</dependencies>

Application Configuration

# application.yml
spring:
datasource:
url: jdbc:snowflake://${SNOWFLAKE_ACCOUNT}.snowflakecomputing.com/?db=${SNOWFLAKE_DATABASE}&warehouse=${SNOWFLAKE_WAREHOUSE}&schema=${SNOWFLAKE_SCHEMA}
username: ${SNOWFLAKE_USERNAME}
password: ${SNOWFLAKE_PASSWORD}
driver-class-name: net.snowflake.client.jdbc.SnowflakeDriver
hikari:
maximum-pool-size: 10
minimum-idle: 2
connection-timeout: 30000
idle-timeout: 600000
max-lifetime: 1800000
snowflake:
account: ${SNOWFLAKE_ACCOUNT:your_account}
database: ${SNOWFLAKE_DATABASE:PROD_DW}
warehouse: ${SNOWFLAKE_WAREHOUSE:COMPUTE_WH}
schema: ${SNOWFLAKE_SCHEMA:PUBLIC}
role: ${SNOWFLAKE_ROLE:SYSADMIN}
stage: ${SNOWFLAKE_STAGE:MY_STAGE}
file-format: ${SNOWFLAKE_FILE_FORMAT:MY_CSV_FORMAT}
app:
data-warehouse:
batch-size: 10000
max-retries: 3
retry-delay-ms: 1000
staging-table-prefix: STG_
history-table-suffix: _HISTORY

2. Snowflake Configuration and Connection Management

Snowflake Configuration Class

package com.example.snowflake.config;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
@Data
@Component
@ConfigurationProperties(prefix = "snowflake")
public class SnowflakeConfig {
private String account;
private String username;
private String password;
private String database;
private String warehouse;
private String schema;
private String role;
private String stage;
private String fileFormat;
public String getJdbcUrl() {
return String.format(
"jdbc:snowflake://%s.snowflakecomputing.com/?db=%s&warehouse=%s&schema=%s&role=%s",
account, database, warehouse, schema, role
);
}
}

Snowflake Connection Manager

package com.example.snowflake.connection;
import com.example.snowflake.config.SnowflakeConfig;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.Properties;
@Component
public class SnowflakeConnectionManager {
private static final Logger logger = LoggerFactory.getLogger(SnowflakeConnectionManager.class);
private final HikariDataSource dataSource;
private final SnowflakeConfig config;
public SnowflakeConnectionManager(SnowflakeConfig config) {
this.config = config;
this.dataSource = createDataSource();
testConnection();
}
private HikariDataSource createDataSource() {
HikariConfig hikariConfig = new HikariConfig();
// Basic connection configuration
hikariConfig.setJdbcUrl(config.getJdbcUrl());
hikariConfig.setUsername(config.getUsername());
hikariConfig.setPassword(config.getPassword());
hikariConfig.setDriverClassName("net.snowflake.client.jdbc.SnowflakeDriver");
// Connection pool settings
hikariConfig.setMaximumPoolSize(10);
hikariConfig.setMinimumIdle(2);
hikariConfig.setConnectionTimeout(30000);
hikariConfig.setIdleTimeout(600000);
hikariConfig.setMaxLifetime(1800000);
// Snowflake-specific properties
Properties properties = new Properties();
properties.put("application", "JavaDataWarehouseApp");
properties.put("query_tag", "data_warehouse_ingestion");
hikariConfig.setDataSourceProperties(properties);
return new HikariDataSource(hikariConfig);
}
private void testConnection() {
try (Connection connection = dataSource.getConnection()) {
logger.info("Successfully connected to Snowflake: {}", config.getDatabase());
} catch (SQLException e) {
logger.error("Failed to connect to Snowflake", e);
throw new RuntimeException("Snowflake connection failed", e);
}
}
public Connection getConnection() throws SQLException {
return dataSource.getConnection();
}
public DataSource getDataSource() {
return dataSource;
}
public void close() {
if (dataSource != null && !dataSource.isClosed()) {
dataSource.close();
logger.info("Snowflake connection pool closed");
}
}
}

3. Data Warehouse Schema Management

DDL Management Service

package com.example.snowflake.schema;
import com.example.snowflake.connection.SnowflakeConnectionManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Arrays;
import java.util.List;
@Service
public class SchemaManager {
private static final Logger logger = LoggerFactory.getLogger(SchemaManager.class);
private final SnowflakeConnectionManager connectionManager;
public SchemaManager(SnowflakeConnectionManager connectionManager) {
this.connectionManager = connectionManager;
}
public void createDatabase(String databaseName) {
String sql = String.format("CREATE DATABASE IF NOT EXISTS %s", databaseName);
executeUpdate(sql);
logger.info("Created database: {}", databaseName);
}
public void createSchema(String schemaName) {
String sql = String.format("CREATE SCHEMA IF NOT EXISTS %s", schemaName);
executeUpdate(sql);
logger.info("Created schema: {}", schemaName);
}
public void createWarehouse(String warehouseName, String size) {
String sql = String.format(
"CREATE WAREHOUSE IF NOT EXISTS %s WITH WAREHOUSE_SIZE = '%s' AUTO_SUSPEND = 300 AUTO_RESUME = TRUE",
warehouseName, size
);
executeUpdate(sql);
logger.info("Created warehouse: {} with size {}", warehouseName, size);
}
// Dimension Tables
public void createDateDimensionTable() {
String sql = """
CREATE TABLE IF NOT EXISTS DIM_DATE (
DATE_KEY NUMBER(8) PRIMARY KEY,
FULL_DATE DATE NOT NULL,
DAY_NAME VARCHAR(10) NOT NULL,
DAY_OF_WEEK NUMBER(1) NOT NULL,
DAY_OF_MONTH NUMBER(2) NOT NULL,
DAY_OF_YEAR NUMBER(3) NOT NULL,
WEEK_OF_YEAR NUMBER(2) NOT NULL,
MONTH_NAME VARCHAR(10) NOT NULL,
MONTH_NUMBER NUMBER(2) NOT NULL,
YEAR NUMBER(4) NOT NULL,
QUARTER NUMBER(1) NOT NULL,
IS_WEEKEND BOOLEAN NOT NULL,
IS_HOLIDAY BOOLEAN NOT NULL,
CREATED_TIMESTAMP TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
)
""";
executeUpdate(sql);
logger.info("Created date dimension table");
}
public void createCustomerDimensionTable() {
String sql = """
CREATE TABLE IF NOT EXISTS DIM_CUSTOMER (
CUSTOMER_KEY NUMBER(38) PRIMARY KEY,
CUSTOMER_ID VARCHAR(50) NOT NULL,
CUSTOMER_NAME VARCHAR(100) NOT NULL,
EMAIL VARCHAR(100),
PHONE VARCHAR(20),
ADDRESS VARCHAR(200),
CITY VARCHAR(50),
STATE VARCHAR(50),
COUNTRY VARCHAR(50),
POSTAL_CODE VARCHAR(20),
SEGMENT VARCHAR(50),
IS_ACTIVE BOOLEAN NOT NULL,
VALID_FROM DATE NOT NULL,
VALID_TO DATE,
IS_CURRENT BOOLEAN NOT NULL,
CREATED_TIMESTAMP TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
UPDATED_TIMESTAMP TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
)
""";
executeUpdate(sql);
logger.info("Created customer dimension table");
}
public void createProductDimensionTable() {
String sql = """
CREATE TABLE IF NOT EXISTS DIM_PRODUCT (
PRODUCT_KEY NUMBER(38) PRIMARY KEY,
PRODUCT_ID VARCHAR(50) NOT NULL,
PRODUCT_NAME VARCHAR(200) NOT NULL,
CATEGORY VARCHAR(100),
SUBCATEGORY VARCHAR(100),
BRAND VARCHAR(100),
PRICE NUMBER(10,2),
COST NUMBER(10,2),
IS_ACTIVE BOOLEAN NOT NULL,
VALID_FROM DATE NOT NULL,
VALID_TO DATE,
IS_CURRENT BOOLEAN NOT NULL,
CREATED_TIMESTAMP TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
UPDATED_TIMESTAMP TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
)
""";
executeUpdate(sql);
logger.info("Created product dimension table");
}
// Fact Tables
public void createSalesFactTable() {
String sql = """
CREATE TABLE IF NOT EXISTS FACT_SALES (
SALES_KEY NUMBER(38) PRIMARY KEY,
DATE_KEY NUMBER(8) NOT NULL REFERENCES DIM_DATE(DATE_KEY),
CUSTOMER_KEY NUMBER(38) NOT NULL REFERENCES DIM_CUSTOMER(CUSTOMER_KEY),
PRODUCT_KEY NUMBER(38) NOT NULL REFERENCES DIM_PRODUCT(PRODUCT_KEY),
ORDER_ID VARCHAR(50) NOT NULL,
ORDER_DATE DATE NOT NULL,
QUANTITY NUMBER(10) NOT NULL,
UNIT_PRICE NUMBER(10,2) NOT NULL,
TOTAL_AMOUNT NUMBER(10,2) NOT NULL,
DISCOUNT_AMOUNT NUMBER(10,2) DEFAULT 0,
NET_AMOUNT NUMBER(10,2) NOT NULL,
SHIPPING_COST NUMBER(10,2) DEFAULT 0,
TAX_AMOUNT NUMBER(10,2) DEFAULT 0,
PAYMENT_METHOD VARCHAR(50),
SALES_REP_ID VARCHAR(50),
CREATED_TIMESTAMP TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
)
""";
executeUpdate(sql);
logger.info("Created sales fact table");
}
// Staging Tables
public void createStagingTable(String tableName, String columns) {
String sql = String.format(
"CREATE TABLE IF NOT EXISTS %s (%s)",
tableName, columns
);
executeUpdate(sql);
logger.info("Created staging table: {}", tableName);
}
// Indexes and Clustering
public void createIndexes() {
List<String> indexSqls = Arrays.asList(
"CREATE CLUSTER BY (DATE_KEY) ON FACT_SALES",
"CREATE CLUSTER BY (CUSTOMER_KEY) ON FACT_SALES", 
"CREATE CLUSTER BY (PRODUCT_KEY) ON FACT_SALES",
"CREATE INDEX IF NOT EXISTS IDX_FACT_SALES_DATE ON FACT_SALES(DATE_KEY)",
"CREATE INDEX IF NOT EXISTS IDX_FACT_SALES_CUSTOMER ON FACT_SALES(CUSTOMER_KEY)",
"CREATE INDEX IF NOT EXISTS IDX_FACT_SALES_PRODUCT ON FACT_SALES(PRODUCT_KEY)",
"CREATE INDEX IF NOT EXISTS IDX_DIM_CUSTOMER_ID ON DIM_CUSTOMER(CUSTOMER_ID)",
"CREATE INDEX IF NOT EXISTS IDX_DIM_PRODUCT_ID ON DIM_PRODUCT(PRODUCT_ID)"
);
for (String sql : indexSqls) {
executeUpdate(sql);
}
logger.info("Created indexes and clustering keys");
}
// File Format for Data Loading
public void createFileFormat() {
String sql = """
CREATE FILE FORMAT IF NOT EXISTS CSV_FORMAT 
TYPE = 'CSV'
FIELD_DELIMITER = ','
SKIP_HEADER = 1
NULL_IF = ('NULL', 'null')
EMPTY_FIELD_AS_NULL = TRUE
COMPRESSION = AUTO
""";
executeUpdate(sql);
logger.info("Created CSV file format");
}
// Stage for Data Loading
public void createStage(String stageName, String fileFormat) {
String sql = String.format(
"CREATE STAGE IF NOT EXISTS %s FILE_FORMAT = %s",
stageName, fileFormat
);
executeUpdate(sql);
logger.info("Created stage: {}", stageName);
}
private void executeUpdate(String sql) {
try (Connection connection = connectionManager.getConnection();
Statement statement = connection.createStatement()) {
statement.execute(sql);
} catch (SQLException e) {
logger.error("Failed to execute SQL: {}", sql, e);
throw new SchemaManagementException("Schema operation failed", e);
}
}
public static class SchemaManagementException extends RuntimeException {
public SchemaManagementException(String message) {
super(message);
}
public SchemaManagementException(String message, Throwable cause) {
super(message, cause);
}
}
}

4. Data Ingestion Service

Data Ingestion Service

package com.example.snowflake.ingestion;
import com.example.snowflake.connection.SnowflakeConnectionManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import java.io.File;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.List;
import java.util.Map;
@Service
public class DataIngestionService {
private static final Logger logger = LoggerFactory.getLogger(DataIngestionService.class);
private final SnowflakeConnectionManager connectionManager;
@Value("${snowflake.stage:MY_STAGE}")
private String stageName;
@Value("${snowflake.file-format:MY_CSV_FORMAT}")
private String fileFormat;
@Value("${app.data-warehouse.batch-size:10000}")
private int batchSize;
public DataIngestionService(SnowflakeConnectionManager connectionManager) {
this.connectionManager = connectionManager;
}
// Bulk load from stage
public void bulkLoadFromStage(String targetTable, String stagePath, String filePattern) {
String sql = String.format(
"COPY INTO %s FROM @%s/%s FILE_FORMAT = (FORMAT_NAME = '%s')",
targetTable, stageName, stagePath, fileFormat
);
executeUpdate(sql);
logger.info("Bulk loaded data into {} from stage path: {}", targetTable, stagePath);
}
// Bulk load from local file
public void bulkLoadFromLocalFile(String targetTable, File localFile) {
try {
// First, upload file to stage
String fileName = localFile.getName();
uploadFileToStage(localFile, fileName);
// Then load from stage
bulkLoadFromStage(targetTable, "", fileName);
logger.info("Successfully loaded data from local file: {} into {}", fileName, targetTable);
} catch (Exception e) {
logger.error("Failed to load data from local file: {}", localFile.getName(), e);
throw new DataIngestionException("Bulk load from local file failed", e);
}
}
private void uploadFileToStage(File file, String fileName) {
String sql = String.format("PUT file://%s @%s/%s", file.getAbsolutePath(), stageName, fileName);
executeUpdate(sql);
logger.debug("Uploaded file {} to stage", fileName);
}
// Batch insert for real-time data
public void batchInsert(String tableName, List<Map<String, Object>> records) {
if (records.isEmpty()) {
return;
}
// Generate insert statement based on first record's keys
Map<String, Object> firstRecord = records.get(0);
String columns = String.join(", ", firstRecord.keySet());
String placeholders = String.join(", ", firstRecord.keySet().stream()
.map(k -> "?").toList());
String sql = String.format("INSERT INTO %s (%s) VALUES (%s)", tableName, columns, placeholders);
try (Connection connection = connectionManager.getConnection();
PreparedStatement statement = connection.prepareStatement(sql)) {
int count = 0;
for (Map<String, Object> record : records) {
int paramIndex = 1;
for (String key : firstRecord.keySet()) {
Object value = record.get(key);
statement.setObject(paramIndex++, value);
}
statement.addBatch();
if (++count % batchSize == 0) {
statement.executeBatch();
logger.debug("Executed batch of {} records", batchSize);
}
}
// Execute remaining batches
if (count % batchSize != 0) {
statement.executeBatch();
}
logger.info("Successfully inserted {} records into {}", records.size(), tableName);
} catch (SQLException e) {
logger.error("Failed to batch insert into {}: {}", tableName, e.getMessage());
throw new DataIngestionException("Batch insert failed", e);
}
}
// Merge (UPSERT) operation for SCD Type 2 dimensions
public void mergeCustomerDimension(List<Map<String, Object>> customerUpdates) {
String mergeSql = """
MERGE INTO DIM_CUSTOMER AS target
USING (SELECT ? AS CUSTOMER_ID, ? AS CUSTOMER_NAME, ? AS EMAIL, ? AS PHONE, 
? AS ADDRESS, ? AS CITY, ? AS STATE, ? AS COUNTRY, ? AS POSTAL_CODE,
? AS SEGMENT, ? AS IS_ACTIVE, ? AS VALID_FROM) AS source
ON target.CUSTOMER_ID = source.CUSTOMER_ID AND target.IS_CURRENT = TRUE
WHEN MATCHED AND (target.CUSTOMER_NAME != source.CUSTOMER_NAME OR 
target.EMAIL != source.EMAIL OR target.PHONE != source.PHONE OR
target.ADDRESS != source.ADDRESS OR target.CITY != source.CITY OR
target.STATE != source.STATE OR target.COUNTRY != source.COUNTRY OR
target.POSTAL_CODE != source.POSTAL_CODE OR target.SEGMENT != source.SEGMENT OR
target.IS_ACTIVE != source.IS_ACTIVE)
THEN UPDATE SET 
target.VALID_TO = source.VALID_FROM - INTERVAL '1 day',
target.IS_CURRENT = FALSE,
target.UPDATED_TIMESTAMP = CURRENT_TIMESTAMP()
WHEN NOT MATCHED THEN
INSERT (CUSTOMER_KEY, CUSTOMER_ID, CUSTOMER_NAME, EMAIL, PHONE, ADDRESS, CITY, 
STATE, COUNTRY, POSTAL_CODE, SEGMENT, IS_ACTIVE, VALID_FROM, VALID_TO, IS_CURRENT)
VALUES (DIM_CUSTOMER_SEQ.NEXTVAL, source.CUSTOMER_ID, source.CUSTOMER_NAME, source.EMAIL, 
source.PHONE, source.ADDRESS, source.CITY, source.STATE, source.COUNTRY, 
source.POSTAL_CODE, source.SEGMENT, source.IS_ACTIVE, source.VALID_FROM, 
'9999-12-31', TRUE)
""";
try (Connection connection = connectionManager.getConnection();
PreparedStatement statement = connection.prepareStatement(mergeSql)) {
for (Map<String, Object> customer : customerUpdates) {
statement.setString(1, (String) customer.get("CUSTOMER_ID"));
statement.setString(2, (String) customer.get("CUSTOMER_NAME"));
statement.setString(3, (String) customer.get("EMAIL"));
statement.setString(4, (String) customer.get("PHONE"));
statement.setString(5, (String) customer.get("ADDRESS"));
statement.setString(6, (String) customer.get("CITY"));
statement.setString(7, (String) customer.get("STATE"));
statement.setString(8, (String) customer.get("COUNTRY"));
statement.setString(9, (String) customer.get("POSTAL_CODE"));
statement.setString(10, (String) customer.get("SEGMENT"));
statement.setBoolean(11, (Boolean) customer.get("IS_ACTIVE"));
statement.setDate(12, java.sql.Date.valueOf(customer.get("VALID_FROM").toString()));
statement.addBatch();
}
int[] results = statement.executeBatch();
logger.info("Merged {} customer records", results.length);
} catch (SQLException e) {
logger.error("Failed to merge customer dimension", e);
throw new DataIngestionException("Customer dimension merge failed", e);
}
}
// ETL Process for loading fact table
public void loadSalesFactFromStaging() {
String etlSql = """
INSERT INTO FACT_SALES (
SALES_KEY, DATE_KEY, CUSTOMER_KEY, PRODUCT_KEY, ORDER_ID, ORDER_DATE,
QUANTITY, UNIT_PRICE, TOTAL_AMOUNT, DISCOUNT_AMOUNT, NET_AMOUNT,
SHIPPING_COST, TAX_AMOUNT, PAYMENT_METHOD, SALES_REP_ID
)
SELECT 
FACT_SALES_SEQ.NEXTVAL,
d.DATE_KEY,
c.CUSTOMER_KEY,
p.PRODUCT_KEY,
s.ORDER_ID,
s.ORDER_DATE,
s.QUANTITY,
s.UNIT_PRICE,
s.TOTAL_AMOUNT,
s.DISCOUNT_AMOUNT,
s.NET_AMOUNT,
s.SHIPPING_COST,
s.TAX_AMOUNT,
s.PAYMENT_METHOD,
s.SALES_REP_ID
FROM STG_SALES s
JOIN DIM_DATE d ON s.ORDER_DATE = d.FULL_DATE
JOIN DIM_CUSTOMER c ON s.CUSTOMER_ID = c.CUSTOMER_ID AND c.IS_CURRENT = TRUE
JOIN DIM_PRODUCT p ON s.PRODUCT_ID = p.PRODUCT_ID AND p.IS_CURRENT = TRUE
WHERE NOT EXISTS (
SELECT 1 FROM FACT_SALES fs 
WHERE fs.ORDER_ID = s.ORDER_ID AND fs.PRODUCT_KEY = p.PRODUCT_KEY
)
""";
int affectedRows = executeUpdateWithCount(etlSql);
logger.info("Loaded {} records into sales fact table", affectedRows);
}
private void executeUpdate(String sql) {
try (Connection connection = connectionManager.getConnection();
Statement statement = connection.createStatement()) {
statement.execute(sql);
} catch (SQLException e) {
logger.error("Failed to execute SQL: {}", sql, e);
throw new DataIngestionException("Data ingestion operation failed", e);
}
}
private int executeUpdateWithCount(String sql) {
try (Connection connection = connectionManager.getConnection();
Statement statement = connection.createStatement()) {
return statement.executeUpdate(sql);
} catch (SQLException e) {
logger.error("Failed to execute SQL: {}", sql, e);
throw new DataIngestionException("Data ingestion operation failed", e);
}
}
public static class DataIngestionException extends RuntimeException {
public DataIngestionException(String message) {
super(message);
}
public DataIngestionException(String message, Throwable cause) {
super(message, cause);
}
}
}

5. Data Query and Analytics Service

Analytics Query Service

package com.example.snowflake.analytics;
import com.example.snowflake.connection.SnowflakeConnectionManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.LocalDate;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@Service
public class AnalyticsQueryService {
private static final Logger logger = LoggerFactory.getLogger(AnalyticsQueryService.class);
private final SnowflakeConnectionManager connectionManager;
public AnalyticsQueryService(SnowflakeConnectionManager connectionManager) {
this.connectionManager = connectionManager;
}
// Sales Analytics
public List<SalesSummary> getSalesSummary(LocalDate startDate, LocalDate endDate) {
String sql = """
SELECT 
d.YEAR,
d.QUARTER,
d.MONTH_NAME,
SUM(fs.QUANTITY) as TOTAL_QUANTITY,
SUM(fs.TOTAL_AMOUNT) as TOTAL_SALES,
SUM(fs.NET_AMOUNT) as NET_SALES,
AVG(fs.UNIT_PRICE) as AVG_UNIT_PRICE,
COUNT(DISTINCT fs.ORDER_ID) as ORDER_COUNT,
COUNT(DISTINCT fs.CUSTOMER_KEY) as CUSTOMER_COUNT
FROM FACT_SALES fs
JOIN DIM_DATE d ON fs.DATE_KEY = d.DATE_KEY
WHERE d.FULL_DATE BETWEEN ? AND ?
GROUP BY d.YEAR, d.QUARTER, d.MONTH_NAME
ORDER BY d.YEAR, d.QUARTER, d.MONTH_NAME
""";
try (Connection connection = connectionManager.getConnection();
PreparedStatement statement = connection.prepareStatement(sql)) {
statement.setDate(1, java.sql.Date.valueOf(startDate));
statement.setDate(2, java.sql.Date.valueOf(endDate));
ResultSet rs = statement.executeQuery();
List<SalesSummary> results = new ArrayList<>();
while (rs.next()) {
SalesSummary summary = new SalesSummary(
rs.getInt("YEAR"),
rs.getInt("QUARTER"),
rs.getString("MONTH_NAME"),
rs.getLong("TOTAL_QUANTITY"),
rs.getBigDecimal("TOTAL_SALES"),
rs.getBigDecimal("NET_SALES"),
rs.getBigDecimal("AVG_UNIT_PRICE"),
rs.getLong("ORDER_COUNT"),
rs.getLong("CUSTOMER_COUNT")
);
results.add(summary);
}
return results;
} catch (SQLException e) {
logger.error("Failed to get sales summary", e);
throw new AnalyticsException("Sales summary query failed", e);
}
}
// Customer Analytics
public List<CustomerAnalysis> getTopCustomers(int limit) {
String sql = """
SELECT 
c.CUSTOMER_ID,
c.CUSTOMER_NAME,
c.SEGMENT,
c.COUNTRY,
COUNT(DISTINCT fs.ORDER_ID) as ORDER_COUNT,
SUM(fs.NET_AMOUNT) as TOTAL_SPENT,
AVG(fs.NET_AMOUNT) as AVG_ORDER_VALUE,
MAX(fs.ORDER_DATE) as LAST_ORDER_DATE
FROM FACT_SALES fs
JOIN DIM_CUSTOMER c ON fs.CUSTOMER_KEY = c.CUSTOMER_KEY
WHERE c.IS_CURRENT = TRUE
GROUP BY c.CUSTOMER_ID, c.CUSTOMER_NAME, c.SEGMENT, c.COUNTRY
ORDER BY TOTAL_SPENT DESC
LIMIT ?
""";
try (Connection connection = connectionManager.getConnection();
PreparedStatement statement = connection.prepareStatement(sql)) {
statement.setInt(1, limit);
ResultSet rs = statement.executeQuery();
List<CustomerAnalysis> results = new ArrayList<>();
while (rs.next()) {
CustomerAnalysis analysis = new CustomerAnalysis(
rs.getString("CUSTOMER_ID"),
rs.getString("CUSTOMER_NAME"),
rs.getString("SEGMENT"),
rs.getString("COUNTRY"),
rs.getLong("ORDER_COUNT"),
rs.getBigDecimal("TOTAL_SPENT"),
rs.getBigDecimal("AVG_ORDER_VALUE"),
rs.getDate("LAST_ORDER_DATE").toLocalDate()
);
results.add(analysis);
}
return results;
} catch (SQLException e) {
logger.error("Failed to get top customers", e);
throw new AnalyticsException("Top customers query failed", e);
}
}
// Product Performance
public List<ProductPerformance> getProductPerformance(String category) {
String sql = """
SELECT 
p.PRODUCT_ID,
p.PRODUCT_NAME,
p.CATEGORY,
p.BRAND,
SUM(fs.QUANTITY) as TOTAL_QUANTITY_SOLD,
SUM(fs.NET_AMOUNT) as TOTAL_REVENUE,
AVG(fs.UNIT_PRICE) as AVG_SELLING_PRICE,
p.COST as UNIT_COST,
(AVG(fs.UNIT_PRICE) - p.COST) as AVG_PROFIT_MARGIN
FROM FACT_SALES fs
JOIN DIM_PRODUCT p ON fs.PRODUCT_KEY = p.PRODUCT_KEY
WHERE p.IS_CURRENT = TRUE AND p.CATEGORY = ?
GROUP BY p.PRODUCT_ID, p.PRODUCT_NAME, p.CATEGORY, p.BRAND, p.COST
ORDER BY TOTAL_REVENUE DESC
""";
try (Connection connection = connectionManager.getConnection();
PreparedStatement statement = connection.prepareStatement(sql)) {
statement.setString(1, category);
ResultSet rs = statement.executeQuery();
List<ProductPerformance> results = new ArrayList<>();
while (rs.next()) {
ProductPerformance performance = new ProductPerformance(
rs.getString("PRODUCT_ID"),
rs.getString("PRODUCT_NAME"),
rs.getString("CATEGORY"),
rs.getString("BRAND"),
rs.getLong("TOTAL_QUANTITY_SOLD"),
rs.getBigDecimal("TOTAL_REVENUE"),
rs.getBigDecimal("AVG_SELLING_PRICE"),
rs.getBigDecimal("UNIT_COST"),
rs.getBigDecimal("AVG_PROFIT_MARGIN")
);
results.add(performance);
}
return results;
} catch (SQLException e) {
logger.error("Failed to get product performance", e);
throw new AnalyticsException("Product performance query failed", e);
}
}
// Monthly Sales Trend
public List<MonthlyTrend> getMonthlySalesTrend(int year) {
String sql = """
SELECT 
d.MONTH_NAME,
d.MONTH_NUMBER,
SUM(fs.NET_AMOUNT) as MONTHLY_SALES,
LAG(SUM(fs.NET_AMOUNT)) OVER (ORDER BY d.MONTH_NUMBER) as PREV_MONTH_SALES,
(SUM(fs.NET_AMOUNT) - LAG(SUM(fs.NET_AMOUNT)) OVER (ORDER BY d.MONTH_NUMBER)) 
/ LAG(SUM(fs.NET_AMOUNT)) OVER (ORDER BY d.MONTH_NUMBER) * 100 as GROWTH_PERCENT
FROM FACT_SALES fs
JOIN DIM_DATE d ON fs.DATE_KEY = d.DATE_KEY
WHERE d.YEAR = ?
GROUP BY d.MONTH_NAME, d.MONTH_NUMBER
ORDER BY d.MONTH_NUMBER
""";
try (Connection connection = connectionManager.getConnection();
PreparedStatement statement = connection.prepareStatement(sql)) {
statement.setInt(1, year);
ResultSet rs = statement.executeQuery();
List<MonthlyTrend> results = new ArrayList<>();
while (rs.next()) {
MonthlyTrend trend = new MonthlyTrend(
rs.getString("MONTH_NAME"),
rs.getInt("MONTH_NUMBER"),
rs.getBigDecimal("MONTHLY_SALES"),
rs.getBigDecimal("PREV_MONTH_SALES"),
rs.getBigDecimal("GROWTH_PERCENT")
);
results.add(trend);
}
return results;
} catch (SQLException e) {
logger.error("Failed to get monthly sales trend", e);
throw new AnalyticsException("Monthly trend query failed", e);
}
}
// Custom Query Execution
public List<Map<String, Object>> executeCustomQuery(String sql, Object... params) {
try (Connection connection = connectionManager.getConnection();
PreparedStatement statement = connection.prepareStatement(sql)) {
// Set parameters
for (int i = 0; i < params.length; i++) {
statement.setObject(i + 1, params[i]);
}
ResultSet rs = statement.executeQuery();
List<Map<String, Object>> results = new ArrayList<>();
var metadata = rs.getMetaData();
int columnCount = metadata.getColumnCount();
while (rs.next()) {
Map<String, Object> row = new HashMap<>();
for (int i = 1; i <= columnCount; i++) {
String columnName = metadata.getColumnName(i);
Object value = rs.getObject(i);
row.put(columnName, value);
}
results.add(row);
}
return results;
} catch (SQLException e) {
logger.error("Failed to execute custom query", e);
throw new AnalyticsException("Custom query execution failed", e);
}
}
// Data Models
public record SalesSummary(int year, int quarter, String month, long totalQuantity, 
java.math.BigDecimal totalSales, java.math.BigDecimal netSales,
java.math.BigDecimal avgUnitPrice, long orderCount, long customerCount) {}
public record CustomerAnalysis(String customerId, String customerName, String segment, 
String country, long orderCount, java.math.BigDecimal totalSpent,
java.math.BigDecimal avgOrderValue, java.time.LocalDate lastOrderDate) {}
public record ProductPerformance(String productId, String productName, String category, 
String brand, long quantitySold, java.math.BigDecimal totalRevenue,
java.math.BigDecimal avgSellingPrice, java.math.BigDecimal unitCost,
java.math.BigDecimal profitMargin) {}
public record MonthlyTrend(String monthName, int monthNumber, java.math.BigDecimal monthlySales,
java.math.BigDecimal prevMonthSales, java.math.BigDecimal growthPercent) {}
public static class AnalyticsException extends RuntimeException {
public AnalyticsException(String message) {
super(message);
}
public AnalyticsException(String message, Throwable cause) {
super(message, cause);
}
}
}

6. Data Warehouse Management Service

Warehouse Management Service

package com.example.snowflake.management;
import com.example.snowflake.connection.SnowflakeConnectionManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
@Service
public class WarehouseManagementService {
private static final Logger logger = LoggerFactory.getLogger(WarehouseManagementService.class);
private final SnowflakeConnectionManager connectionManager;
public WarehouseManagementService(SnowflakeConnectionManager connectionManager) {
this.connectionManager = connectionManager;
}
// Warehouse operations
public void resizeWarehouse(String warehouseName, String newSize) {
String sql = String.format("ALTER WAREHOUSE %s SET WAREHOUSE_SIZE = %s", warehouseName, newSize);
executeUpdate(sql);
logger.info("Resized warehouse {} to {}", warehouseName, newSize);
}
public void suspendWarehouse(String warehouseName) {
String sql = String.format("ALTER WAREHOUSE %s SUSPEND", warehouseName);
executeUpdate(sql);
logger.info("Suspended warehouse: {}", warehouseName);
}
public void resumeWarehouse(String warehouseName) {
String sql = String.format("ALTER WAREHOUSE %s RESUME", warehouseName);
executeUpdate(sql);
logger.info("Resumed warehouse: {}", warehouseName);
}
// Table maintenance
public void cloneTable(String sourceTable, String targetTable) {
String sql = String.format("CREATE TABLE %s CLONE %s", targetTable, sourceTable);
executeUpdate(sql);
logger.info("Cloned table {} to {}", sourceTable, targetTable);
}
public void swapTables(String table1, String table2) {
String sql = String.format("ALTER TABLE %s SWAP WITH %s", table1, table2);
executeUpdate(sql);
logger.info("Swapped tables {} and {}", table1, table2);
}
// Query history and monitoring
public List<QueryHistory> getRecentQueryHistory(int limit) {
String sql = String.format("""
SELECT 
QUERY_ID,
QUERY_TEXT,
DATABASE_NAME,
SCHEMA_NAME,
QUERY_TYPE,
SESSION_ID,
USER_NAME,
ROLE_NAME,
WAREHOUSE_NAME,
WAREHOUSE_SIZE,
EXECUTION_STATUS,
ERROR_CODE,
ERROR_MESSAGE,
START_TIME,
END_TIME,
TOTAL_ELAPSED_TIME,
BYTES_SCANNED,
ROWS_PRODUCED
FROM TABLE(INFORMATION_SCHEMA.QUERY_HISTORY())
ORDER BY START_TIME DESC
LIMIT %d
""", limit);
try (Connection connection = connectionManager.getConnection();
Statement statement = connection.createStatement();
ResultSet rs = statement.executeQuery(sql)) {
List<QueryHistory> history = new ArrayList<>();
while (rs.next()) {
QueryHistory query = new QueryHistory(
rs.getString("QUERY_ID"),
rs.getString("QUERY_TEXT"),
rs.getString("DATABASE_NAME"),
rs.getString("SCHEMA_NAME"),
rs.getString("QUERY_TYPE"),
rs.getString("SESSION_ID"),
rs.getString("USER_NAME"),
rs.getString("ROLE_NAME"),
rs.getString("WAREHOUSE_NAME"),
rs.getString("WAREHOUSE_SIZE"),
rs.getString("EXECUTION_STATUS"),
rs.getString("ERROR_CODE"),
rs.getString("ERROR_MESSAGE"),
rs.getTimestamp("START_TIME"),
rs.getTimestamp("END_TIME"),
rs.getLong("TOTAL_ELAPSED_TIME"),
rs.getLong("BYTES_SCANNED"),
rs.getLong("ROWS_PRODUCED")
);
history.add(query);
}
return history;
} catch (SQLException e) {
logger.error("Failed to get query history", e);
throw new ManagementException("Query history retrieval failed", e);
}
}
// Storage monitoring
public StorageUsage getStorageUsage() {
String sql = """
SELECT 
USAGE_DATE,
STORAGE_BYTES,
STAGE_BYTES,
FAILSAFE_BYTES
FROM TABLE(INFORMATION_SCHEMA.STORAGE_USAGE())
ORDER BY USAGE_DATE DESC
LIMIT 1
""";
try (Connection connection = connectionManager.getConnection();
Statement statement = connection.createStatement();
ResultSet rs = statement.executeQuery(sql)) {
if (rs.next()) {
return new StorageUsage(
rs.getDate("USAGE_DATE"),
rs.getLong("STORAGE_BYTES"),
rs.getLong("STAGE_BYTES"),
rs.getLong("FAILSAFE_BYTES")
);
}
return null;
} catch (SQLException e) {
logger.error("Failed to get storage usage", e);
throw new ManagementException("Storage usage retrieval failed", e);
}
}
// User sessions
public List<UserSession> getActiveSessions() {
String sql = """
SELECT 
SESSION_ID,
USER_NAME,
ROLE_NAME,
CLIENT_APPLICATION_ID,
CLIENT_APPLICATION_VERSION,
LOGIN_TIME,
FIRST_AUTHENTICATION_FACTOR,
WAREHOUSE_NAME,
CURRENT_DATABASE,
CURRENT_SCHEMA
FROM TABLE(INFORMATION_SCHEMA.SESSIONS())
WHERE IS_CURRENT = TRUE
ORDER BY LOGIN_TIME DESC
""";
try (Connection connection = connectionManager.getConnection();
Statement statement = connection.createStatement();
ResultSet rs = statement.executeQuery(sql)) {
List<UserSession> sessions = new ArrayList<>();
while (rs.next()) {
UserSession session = new UserSession(
rs.getString("SESSION_ID"),
rs.getString("USER_NAME"),
rs.getString("ROLE_NAME"),
rs.getString("CLIENT_APPLICATION_ID"),
rs.getString("CLIENT_APPLICATION_VERSION"),
rs.getTimestamp("LOGIN_TIME"),
rs.getString("FIRST_AUTHENTICATION_FACTOR"),
rs.getString("WAREHOUSE_NAME"),
rs.getString("CURRENT_DATABASE"),
rs.getString("CURRENT_SCHEMA")
);
sessions.add(session);
}
return sessions;
} catch (SQLException e) {
logger.error("Failed to get active sessions", e);
throw new ManagementException("Active sessions retrieval failed", e);
}
}
// Data sharing (if using Snowflake Data Sharing)
public void createShare(String shareName, String databases) {
String sql = String.format("CREATE SHARE %s", shareName);
executeUpdate(sql);
// Add databases to share
for (String database : databases.split(",")) {
String addDbSql = String.format("GRANT USAGE ON DATABASE %s TO SHARE %s", database, shareName);
executeUpdate(addDbSql);
}
logger.info("Created share: {} with databases: {}", shareName, databases);
}
private void executeUpdate(String sql) {
try (Connection connection = connectionManager.getConnection();
Statement statement = connection.createStatement()) {
statement.execute(sql);
} catch (SQLException e) {
logger.error("Failed to execute SQL: {}", sql, e);
throw new ManagementException("Management operation failed", e);
}
}
// Data Models
public record QueryHistory(String queryId, String queryText, String databaseName, String schemaName,
String queryType, String sessionId, String userName, String roleName,
String warehouseName, String warehouseSize, String executionStatus,
String errorCode, String errorMessage, java.sql.Timestamp startTime,
java.sql.Timestamp endTime, long totalElapsedTime, long bytesScanned,
long rowsProduced) {}
public record StorageUsage(java.sql.Date usageDate, long storageBytes, long stageBytes, long failsafeBytes) {}
public record UserSession(String sessionId, String userName, String roleName, String clientApplicationId,
String clientApplicationVersion, java.sql.Timestamp loginTime,
String firstAuthenticationFactor, String warehouseName, String currentDatabase,
String currentSchema) {}
public static class ManagementException extends RuntimeException {
public ManagementException(String message) {
super(message);
}
public ManagementException(String message, Throwable cause) {
super(message, cause);
}
}
}

7. REST Controllers

Data Warehouse Controller

package com.example.snowflake.web;
import com.example.snowflake.analytics.AnalyticsQueryService;
import com.example.snowflake.ingestion.DataIngestionService;
import com.example.snowflake.management.WarehouseManagementService;
import com.example.snowflake.schema.SchemaManager;
import org.springframework.format.annotation.DateTimeFormat;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.time.LocalDate;
import java.util.List;
import java.util.Map;
@RestController
@RequestMapping("/api/data-warehouse")
public class DataWarehouseController {
private final SchemaManager schemaManager;
private final DataIngestionService ingestionService;
private final AnalyticsQueryService analyticsService;
private final WarehouseManagementService managementService;
public DataWarehouseController(SchemaManager schemaManager, DataIngestionService ingestionService,
AnalyticsQueryService analyticsService, WarehouseManagementService managementService) {
this.schemaManager = schemaManager;
this.ingestionService = ingestionService;
this.analyticsService = analyticsService;
this.managementService = managementService;
}
// Schema Management Endpoints
@PostMapping("/schema/initialize")
public ResponseEntity<String> initializeSchema() {
schemaManager.createDateDimensionTable();
schemaManager.createCustomerDimensionTable();
schemaManager.createProductDimensionTable();
schemaManager.createSalesFactTable();
schemaManager.createIndexes();
schemaManager.createFileFormat();
schemaManager.createStage("MY_STAGE", "CSV_FORMAT");
return ResponseEntity.ok("Data warehouse schema initialized successfully");
}
// Data Ingestion Endpoints
@PostMapping("/ingest/csv")
public ResponseEntity<String> ingestCsvFile(
@RequestParam String targetTable,
@RequestParam MultipartFile file) {
try {
// Save uploaded file temporarily
Path tempFile = Files.createTempFile("snowflake_upload_", ".csv");
Files.copy(file.getInputStream(), tempFile, StandardCopyOption.REPLACE_EXISTING);
// Ingest data
ingestionService.bulkLoadFromLocalFile(targetTable, tempFile.toFile());
// Clean up
Files.delete(tempFile);
return ResponseEntity.ok("Data ingested successfully into " + targetTable);
} catch (IOException e) {
return ResponseEntity.badRequest().body("File upload failed: " + e.getMessage());
}
}
@PostMapping("/ingest/staging")
public ResponseEntity<String> ingestFromStaging(
@RequestParam String targetTable,
@RequestParam String stagePath) {
ingestionService.bulkLoadFromStage(targetTable, stagePath, "*.csv");
return ResponseEntity.ok("Data ingested from staging area");
}
@PostMapping("/etl/sales-fact")
public ResponseEntity<String> runSalesFactETL() {
ingestionService.loadSalesFactFromStaging();
return ResponseEntity.ok("Sales fact ETL process completed");
}
// Analytics Endpoints
@GetMapping("/analytics/sales-summary")
public ResponseEntity<List<AnalyticsQueryService.SalesSummary>> getSalesSummary(
@RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) LocalDate startDate,
@RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) LocalDate endDate) {
List<AnalyticsQueryService.SalesSummary> summary = 
analyticsService.getSalesSummary(startDate, endDate);
return ResponseEntity.ok(summary);
}
@GetMapping("/analytics/top-customers")
public ResponseEntity<List<AnalyticsQueryService.CustomerAnalysis>> getTopCustomers(
@RequestParam(defaultValue = "10") int limit) {
List<AnalyticsQueryService.CustomerAnalysis> topCustomers = 
analyticsService.getTopCustomers(limit);
return ResponseEntity.ok(topCustomers);
}
@GetMapping("/analytics/product-performance")
public ResponseEntity<List<AnalyticsQueryService.ProductPerformance>> getProductPerformance(
@RequestParam String category) {
List<AnalyticsQueryService.ProductPerformance> performance = 
analyticsService.getProductPerformance(category);
return ResponseEntity.ok(performance);
}
@GetMapping("/analytics/monthly-trend")
public ResponseEntity<List<AnalyticsQueryService.MonthlyTrend>> getMonthlyTrend(
@RequestParam int year) {
List<AnalyticsQueryService.MonthlyTrend> trend = 
analyticsService.getMonthlySalesTrend(year);
return ResponseEntity.ok(trend);
}
// Management Endpoints
@GetMapping("/management/query-history")
public ResponseEntity<List<WarehouseManagementService.QueryHistory>> getQueryHistory(
@RequestParam(defaultValue = "50") int limit) {
List<WarehouseManagementService.QueryHistory> history = 
managementService.getRecentQueryHistory(limit);
return ResponseEntity.ok(history);
}
@GetMapping("/management/storage-usage")
public ResponseEntity<WarehouseManagementService.StorageUsage> getStorageUsage() {
WarehouseManagementService.StorageUsage usage = managementService.getStorageUsage();
return ResponseEntity.ok(usage);
}
@GetMapping("/management/active-sessions")
public ResponseEntity<List<WarehouseManagementService.UserSession>> getActiveSessions() {
List<WarehouseManagementService.UserSession> sessions = managementService.getActiveSessions();
return ResponseEntity.ok(sessions);
}
@PostMapping("/management/warehouse/{warehouseName}/resize")
public ResponseEntity<String> resizeWarehouse(
@PathVariable String warehouseName,
@RequestParam String newSize) {
managementService.resizeWarehouse(warehouseName, newSize);
return ResponseEntity.ok("Warehouse resized successfully");
}
// Custom Query Execution
@PostMapping("/query/custom")
public ResponseEntity<List<Map<String, Object>>> executeCustomQuery(@RequestBody CustomQueryRequest request) {
List<Map<String, Object>> results = analyticsService.executeCustomQuery(
request.sql(), request.params().toArray()
);
return ResponseEntity.ok(results);
}
public record CustomQueryRequest(String sql, List<Object> params) {}
}

8. Spring Boot Application

Main Application Class

package com.example.snowflake;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;
@SpringBootApplication
@EnableScheduling
public class SnowflakeDataWarehouseApplication {
public static void main(String[] args) {
SpringApplication.run(SnowflakeDataWarehouseApplication.class, args);
}
}

Scheduled Tasks

package com.example.snowflake.scheduling;
import com.example.snowflake.ingestion.DataIngestionService;
import com.example.snowflake.management.WarehouseManagementService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
@Component
public class ScheduledTasks {
private static final Logger logger = LoggerFactory.getLogger(ScheduledTasks.class);
private final DataIngestionService ingestionService;
private final WarehouseManagementService managementService;
public ScheduledTasks(DataIngestionService ingestionService, 
WarehouseManagementService managementService) {
this.ingestionService = ingestionService;
this.managementService = managementService;
}
// Run ETL process every hour
@Scheduled(cron = "0 0 * * * *")
public void runScheduledETL() {
logger.info("Starting scheduled ETL process");
try {
ingestionService.loadSalesFactFromStaging();
logger.info("Scheduled ETL process completed successfully");
} catch (Exception e) {
logger.error("Scheduled ETL process failed", e);
}
}
// Monitor warehouse usage daily
@Scheduled(cron = "0 0 6 * * *")
public void monitorWarehouseUsage() {
logger.info("Checking warehouse storage usage");
try {
var storageUsage = managementService.getStorageUsage();
if (storageUsage != null) {
double storageGB = storageUsage.storageBytes() / (1024.0 * 1024.0 * 1024.0);
double stageGB = storageUsage.stageBytes() / (1024.0 * 1024.0 * 1024.0);
logger.info("Storage Usage - Data: {:.2f} GB, Stage: {:.2f} GB", storageGB, stageGB);
// Alert if storage exceeds threshold
if (storageGB > 100) { // 100 GB threshold
logger.warn("Storage usage exceeds 100 GB: {:.2f} GB", storageGB);
}
}
} catch (Exception e) {
logger.error("Storage monitoring failed", e);
}
}
}

9. Testing

Integration Test

package com.example.snowflake.test;
import com.example.snowflake.analytics.AnalyticsQueryService;
import com.example.snowflake.connection.SnowflakeConnectionManager;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.ActiveProfiles;
import java.time.LocalDate;
import static org.assertj.core.api.Assertions.assertThat;
@SpringBootTest
@ActiveProfiles("test")
class SnowflakeIntegrationTest {
@Autowired
private SnowflakeConnectionManager connectionManager;
@Autowired
private AnalyticsQueryService analyticsService;
@Test
void testConnection() throws Exception {
assertThat(connectionManager.getConnection()).isNotNull();
}
@Test
void testSalesSummaryQuery() {
LocalDate startDate = LocalDate.of(2023, 1, 1);
LocalDate endDate = LocalDate.of(2023, 12, 31);
var summary = analyticsService.getSalesSummary(startDate, endDate);
assertThat(summary).isNotNull();
// Add more assertions based on your test data
}
}

Summary

This comprehensive Snowflake Data Warehouse implementation provides:

Key Features:

  • Schema Management: Automated creation of star schema with dimensions and facts
  • Data Ingestion: Bulk loading from stages and real-time batch inserts
  • ETL Processes: Automated data transformation and loading
  • Analytics: Pre-built analytical queries for business intelligence
  • Management: Warehouse monitoring and maintenance operations
  • REST API: Full REST interface for all operations

Benefits:

  • Scalability: Leverages Snowflake's elastic scaling
  • Performance: Optimized queries with clustering and indexing
  • Maintainability: Clean architecture with separation of concerns
  • Monitoring: Comprehensive monitoring and alerting
  • Integration: Easy integration with existing systems

Use Cases:

  • Enterprise data warehousing
  • Business intelligence and reporting
  • Real-time analytics
  • Data lake integration
  • Multi-cloud data strategies

This implementation provides a production-ready foundation for building scalable data warehouse solutions with Snowflake and Java.

Category : JAVA

Leave a Reply

Your email address will not be published. Required fields are marked *


Macro Nepal Helper