snowflake.sql

USE ROLE ACCOUNTADMIN;
CREATE OR REPLACE WAREHOUSE RIDESHARE_INGEST;
ALTER WAREHOUSE RIDESHARE_INGEST SET WAREHOUSE_SIZE='MEDIUM';
CREATE OR REPLACE DATABASE RIDESHARE_DEMO;

ALTER SESSION SET TIMEZONE = 'UTC';

CREATE OR REPLACE ROLE RIDESHARE_INGEST;
GRANT ALL ON WAREHOUSE RIDESHARE_INGEST TO ROLE RIDESHARE_INGEST;
GRANT EXECUTE TASK ON account TO ROLE RIDESHARE_INGEST;
GRANT CREATE INTEGRATION ON account TO ROLE RIDESHARE_INGEST;
GRANT CREATE EXTERNAL VOLUME ON account TO ROLE RIDESHARE_INGEST;
GRANT OWNERSHIP ON DATABASE RIDESHARE_DEMO TO ROLE RIDESHARE_INGEST;
GRANT OWNERSHIP ON SCHEMA RIDESHARE_DEMO.PUBLIC TO ROLE RIDESHARE_INGEST;

CREATE OR REPLACE USER RIDESHARE_INGEST PASSWORD='RIDESHARE_INGEST' LOGIN_NAME='RIDESHARE_INGEST' MUST_CHANGE_PASSWORD=FALSE, DISABLED=FALSE, DEFAULT_WAREHOUSE='RIDESHARE_INGEST', DEFAULT_NAMESPACE='RIDESHARE_DEMO.PUBLIC', DEFAULT_ROLE='RIDESHARE_INGEST';
GRANT ROLE RIDESHARE_INGEST TO USER RIDESHARE_INGEST;
-- change to your current logged in user
GRANT ROLE RIDESHARE_INGEST TO USER BHARAT;

ALTER USER RIDESHARE_INGEST SET RSA_PUBLIC_KEY='YOUR-PUBLIC-KEY';

USE ROLE RIDESHARE_INGEST;
USE DATABASE RIDESHARE_DEMO;

-- query kafka connector created tables
SELECT * FROM RIDERS_STAGE;
SELECT * FROM DRIVERS_STAGE;
SELECT * FROM TRIPS_STAGE;

-- deduplicated riders table
CREATE OR REPLACE TABLE RIDERS (
    ID VARCHAR(255) NOT NULL,
    FIRST_NAME VARCHAR(255),
    LAST_NAME VARCHAR(255),
    EMAIL VARCHAR(255),
    PHONE_NUMBER VARCHAR(50),
    DATE_OF_BIRTH TIMESTAMP_NTZ(6),
    CREATED_AT TIMESTAMP_NTZ(6),
    LOCATION_CITY VARCHAR(255),
    LOCATION_LAT FLOAT,
    LOCATION_LONG FLOAT,
    STATUS VARCHAR(50),
    PRIMARY KEY (ID)
);

-- Create a stream on the RIDERS_STAGE table
CREATE OR REPLACE STREAM RIDERS_STREAM ON TABLE RIDERS_STAGE;

-- Create a task to merge changes into the RIDERS table
CREATE OR REPLACE TASK MERGE_RIDERS_TASK
WAREHOUSE = RIDESHARE_INGEST
SCHEDULE = '1 minute'
WHEN
  SYSTEM$STREAM_HAS_DATA('RIDERS_STREAM')
AS
MERGE INTO RIDESHARE_DEMO.PUBLIC.RIDERS AS target
USING (
  SELECT 
    ID,
    FIRST_NAME,
    LAST_NAME,
    EMAIL,
    PHONE_NUMBER,
    TRY_TO_TIMESTAMP_NTZ(DATE_OF_BIRTH) AS DATE_OF_BIRTH,
    TRY_TO_TIMESTAMP_NTZ(CREATED_AT) AS CREATED_AT,
    LOCATION_CITY,
    LOCATION_LAT,
    LOCATION_LONG,
    STATUS,
    ROW_NUMBER() OVER (PARTITION BY ID ORDER BY TRY_TO_TIMESTAMP(CREATED_AT) DESC) AS rn
  FROM RIDERS_STREAM
) AS source
ON target.ID = source.ID
WHEN MATCHED AND source.rn = 1 THEN
  UPDATE SET
    target.FIRST_NAME = source.FIRST_NAME,
    target.LAST_NAME = source.LAST_NAME,
    target.EMAIL = source.EMAIL,
    target.PHONE_NUMBER = source.PHONE_NUMBER,
    target.DATE_OF_BIRTH = source.DATE_OF_BIRTH,
    target.CREATED_AT = source.CREATED_AT,
    target.LOCATION_CITY = source.LOCATION_CITY,
    target.LOCATION_LAT = source.LOCATION_LAT,
    target.LOCATION_LONG = source.LOCATION_LONG,
    target.STATUS = source.STATUS
WHEN NOT MATCHED AND source.rn = 1 THEN
  INSERT (ID, FIRST_NAME, LAST_NAME, EMAIL, PHONE_NUMBER, DATE_OF_BIRTH, CREATED_AT, LOCATION_CITY, LOCATION_LAT, LOCATION_LONG, STATUS)
  VALUES (source.ID, source.FIRST_NAME, source.LAST_NAME, source.EMAIL, source.PHONE_NUMBER, source.DATE_OF_BIRTH, source.CREATED_AT, source.LOCATION_CITY, source.LOCATION_LAT, source.LOCATION_LONG, source.STATUS);

-- Start the task
ALTER TASK MERGE_RIDERS_TASK RESUME;

-- deduplicated drivers table
CREATE OR REPLACE TABLE DRIVERS (
    ID VARCHAR(255) NOT NULL,
    FIRST_NAME VARCHAR(255),
    LAST_NAME VARCHAR(255),
    EMAIL VARCHAR(255),
    PHONE_NUMBER VARCHAR(50),
    DATE_OF_BIRTH TIMESTAMP_NTZ(6),
    CREATED_AT TIMESTAMP_NTZ(6),
    LOCATION_CITY VARCHAR(255),
    LOCATION_LAT FLOAT,
    LOCATION_LONG FLOAT,
    STATUS VARCHAR(50),
    PRIMARY KEY (ID)
);

-- Create a stream on the DRIVERS_STAGE table
CREATE OR REPLACE STREAM DRIVERS_STREAM ON TABLE DRIVERS_STAGE;

-- Create a task to merge changes into the DRIVERS table
CREATE OR REPLACE TASK MERGE_DRIVERS_TASK
WAREHOUSE = RIDESHARE_INGEST
SCHEDULE = '1 minute'
WHEN
  SYSTEM$STREAM_HAS_DATA('DRIVERS_STREAM')
AS
MERGE INTO RIDESHARE_DEMO.PUBLIC.DRIVERS AS target
USING (
  SELECT 
    ID,
    FIRST_NAME,
    LAST_NAME,
    EMAIL,
    PHONE_NUMBER,
    TRY_TO_TIMESTAMP_NTZ(DATE_OF_BIRTH) AS DATE_OF_BIRTH,
    TRY_TO_TIMESTAMP_NTZ(CREATED_AT) AS CREATED_AT,
    LOCATION_CITY,
    LOCATION_LAT,
    LOCATION_LONG,
    STATUS,
    ROW_NUMBER() OVER (PARTITION BY ID ORDER BY TRY_TO_TIMESTAMP(CREATED_AT) DESC) AS rn
  FROM DRIVERS_STREAM
) AS source
ON target.ID = source.ID
WHEN MATCHED AND source.rn = 1 THEN
  UPDATE SET
    target.FIRST_NAME = source.FIRST_NAME,
    target.LAST_NAME = source.LAST_NAME,
    target.EMAIL = source.EMAIL,
    target.PHONE_NUMBER = source.PHONE_NUMBER,
    target.DATE_OF_BIRTH = source.DATE_OF_BIRTH,
    target.CREATED_AT = source.CREATED_AT,
    target.LOCATION_CITY = source.LOCATION_CITY,
    target.LOCATION_LAT = source.LOCATION_LAT,
    target.LOCATION_LONG = source.LOCATION_LONG,
    target.STATUS = source.STATUS
WHEN NOT MATCHED AND source.rn = 1 THEN
  INSERT (ID, FIRST_NAME, LAST_NAME, EMAIL, PHONE_NUMBER, DATE_OF_BIRTH, CREATED_AT, LOCATION_CITY, LOCATION_LAT, LOCATION_LONG, STATUS)
  VALUES (source.ID, source.FIRST_NAME, source.LAST_NAME, source.EMAIL, source.PHONE_NUMBER, source.DATE_OF_BIRTH, source.CREATED_AT, source.LOCATION_CITY, source.LOCATION_LAT, source.LOCATION_LONG, source.STATUS);

-- Start the task
ALTER TASK MERGE_DRIVERS_TASK RESUME;

-- deduplicated trips table
CREATE OR REPLACE TABLE TRIPS (
    ID VARCHAR(255) NOT NULL,
    DRIVER_ID VARCHAR(255),
    RIDER_ID VARCHAR(255),
    STATUS VARCHAR(50),
    REQUEST_TIME TIMESTAMP_NTZ(6),
    ACCEPT_TIME TIMESTAMP_NTZ(6),
    PICKUP_TIME TIMESTAMP_NTZ(6),
    DROPOFF_TIME TIMESTAMP_NTZ(6),
    FARE INTEGER,
    DISTANCE FLOAT,
    PICKUP_LAT FLOAT,
    PICKUP_LONG FLOAT,
    DROPOFF_LAT FLOAT,
    DROPOFF_LONG FLOAT,
    CITY VARCHAR(255),
    PRIMARY KEY (ID)
);

-- Create a stream on the TRIPS_STAGE table
CREATE OR REPLACE STREAM TRIPS_STREAM ON TABLE TRIPS_STAGE;

-- Create a task to merge changes into the TRIPS table
CREATE OR REPLACE TASK MERGE_TRIPS_TASK
WAREHOUSE = RIDESHARE_INGEST
SCHEDULE = '1 minute'
WHEN
  SYSTEM$STREAM_HAS_DATA('TRIPS_STREAM')
AS
MERGE INTO RIDESHARE_DEMO.PUBLIC.TRIPS AS target
USING (
  SELECT 
    ID,
    DRIVER_ID,
    RIDER_ID,
    STATUS,
    TRY_TO_TIMESTAMP_NTZ(REQUEST_TIME) AS REQUEST_TIME,
    TRY_TO_TIMESTAMP_NTZ(ACCEPT_TIME) AS ACCEPT_TIME,
    TRY_TO_TIMESTAMP_NTZ(PICKUP_TIME) AS PICKUP_TIME,
    TRY_TO_TIMESTAMP_NTZ(DROPOFF_TIME) AS DROPOFF_TIME,
    FARE,
    DISTANCE,
    PICKUP_LAT,
    PICKUP_LONG,
    DROPOFF_LAT,
    DROPOFF_LONG,
    CITY,
    ROW_NUMBER() OVER (PARTITION BY ID ORDER BY TRY_TO_TIMESTAMP_NTZ(RECORD_METADATA:CreateTime::STRING) DESC) AS rn
  FROM TRIPS_STREAM
) AS source
ON target.ID = source.ID
WHEN MATCHED AND source.rn = 1 THEN
  UPDATE SET
    target.DRIVER_ID = source.DRIVER_ID,
    target.RIDER_ID = source.RIDER_ID,
    target.STATUS = source.STATUS,
    target.REQUEST_TIME = source.REQUEST_TIME,
    target.ACCEPT_TIME = source.ACCEPT_TIME,
    target.PICKUP_TIME = source.PICKUP_TIME,
    target.DROPOFF_TIME = source.DROPOFF_TIME,
    target.FARE = source.FARE,
    target.DISTANCE = source.DISTANCE,
    target.PICKUP_LAT = source.PICKUP_LAT,
    target.PICKUP_LONG = source.PICKUP_LONG,
    target.DROPOFF_LAT = source.DROPOFF_LAT,
    target.DROPOFF_LONG = source.DROPOFF_LONG,
    target.CITY = source.CITY
WHEN NOT MATCHED AND source.rn = 1 THEN
  INSERT (ID, DRIVER_ID, RIDER_ID, STATUS, REQUEST_TIME, ACCEPT_TIME, PICKUP_TIME, DROPOFF_TIME, FARE, DISTANCE, PICKUP_LAT, PICKUP_LONG, DROPOFF_LAT, DROPOFF_LONG, CITY)
  VALUES (source.ID, source.DRIVER_ID, source.RIDER_ID, source.STATUS, source.REQUEST_TIME, source.ACCEPT_TIME, source.PICKUP_TIME, source.DROPOFF_TIME, source.FARE, source.DISTANCE, source.PICKUP_LAT, source.PICKUP_LONG, source.DROPOFF_LAT, source.DROPOFF_LONG, source.CITY);

-- Start the task
ALTER TASK MERGE_TRIPS_TASK RESUME;

-- verify deduplicated tables  
SELECT COUNT(*) FROM RIDERS;
SELECT COUNT(*) FROM DRIVERS;
SELECT COUNT(*) FROM TRIPS;

-- (optional) ingest trips csv from s3
-- follow the instructions here: https://docs.snowflake.com/en/user-guide/data-load-s3
CREATE STORAGE INTEGRATION rideshare_s3_int
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::'
  STORAGE_ALLOWED_LOCATIONS = ('s3://');

USE ROLE ACCOUNTADMIN;
GRANT USAGE ON INTEGRATION rideshare_s3_int TO ROLE RIDESHARE_INGEST;
USE ROLE RIDESHARE_INGEST;

DESC INTEGRATION rideshare_s3_int;

CREATE OR REPLACE STAGE rideshare_s3_stage
  STORAGE_INTEGRATION = rideshare_s3_int
  URL = 's3://';

SHOW STAGES;

COPY INTO trips
  FROM @rideshare_s3_stage
    FILES = ( 'trips.csv' );

SELECT COUNT(*) FROM trips;

-- create iceberg volume on s3
-- following the instructions here: https://docs.snowflake.com/en/user-guide/tables-iceberg-configure-external-volume
CREATE OR REPLACE EXTERNAL VOLUME s3_rideshare_iceberg_volume
   STORAGE_LOCATIONS = (
         (
            NAME = 's3_rideshare_iceberg_volume'
            STORAGE_PROVIDER = 'S3'
            STORAGE_BASE_URL = 's3://'
            STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::'
            STORAGE_AWS_EXTERNAL_ID = ''
         )
    );

DESC EXTERNAL VOLUME s3_rideshare_iceberg_volume;

-- create iceberg trips table
CREATE OR REPLACE ICEBERG TABLE trips_ice LIKE trips
CATALOG = 'SNOWFLAKE'
EXTERNAL_VOLUME = 's3_rideshare_iceberg_volume'
BASE_LOCATION = 'iceberg';

-- copy trips into iceberg table
INSERT INTO trips_ice
SELECT * FROM trips;

SELECT COUNT(*) FROM trips_ice;

-- cleanup query to remove any orphaned trips and reset the riders and drivers tables
DELETE FROM RIDERS;
DELETE FROM DRIVERS;
DELETE FROM TRIPS WHERE status != 'completed';

-- sample queries
SELECT 'trips' as entity, status, COUNT(*) as count
    FROM trips
    GROUP BY status
    UNION ALL
    SELECT 'riders' as entity, status, COUNT(*) as count
    FROM riders
    GROUP BY status
    UNION ALL
    SELECT 'drivers' as entity, status, COUNT(*) as count
    FROM drivers
    GROUP BY status
    ORDER BY entity, status;

-- cleanup demo
USE ROLE ACCOUNTADMIN;
DROP WAREHOUSE IF EXISTS RIDESHARE_INGEST;
DROP USER IF EXISTS RIDESHARE_INGEST;
DROP ROLE IF EXISTS RIDESHARE_INGEST;
DROP DATABASE IF EXISTS RIDESHARE_DEMO;
DROP EXTERNAL VOLUME s3_rideshare_iceberg_volume;