Snowflake database pipeline, task, procedure, etc sample

 //create or replace database scd1db

CREATE OR REPLACE STORAGE INTEGRATION SCD1_INT

  TYPE = EXTERNAL_STAGE

  STORAGE_PROVIDER = 'S3'

  ENABLED = TRUE

  STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::848922562186:role/scd1role'

  STORAGE_ALLOWED_LOCATIONS = ('s3://scd1databucket-ambi/data/');

// to describe the database integration details between S3 bucket to snowflake

  DESC INTEGRATION SCD1_INT;

  CREATE OR REPLACE STAGE SCD1_DB.PUBLIC.SCD1_STAGE

STORAGE_INTEGRATION = SCD1_INT

URL= 's3://scd1databucket-ambi/data/';

CREATE OR REPLACE PIPE SCD1_DB.PUBLIC.SCD1PIPE 

AUTO_INGEST = TRUE AS

COPY INTO SCD1_DB.PUBLIC.CUSTOMER_SOURCE

FROM 

(

SELECT

$1 AS CUSTOMERNAME,

$2 AS PHONE,

$3 ASADDRESSLINE1,

$4 AS ADDRESSLINE2,

$5 AS CITY,

$6 AS STATE,

$7 AS POSTALCODE,

$8 AS COUNTRY,

$9 AS TERRITORY,

$10 AS CONTACTFIRSTNAME,

$11 AS CONTACTLASTNAME

FROM @SCD1_DB.PUBLIC.SCD1_STAGE)

FILE_FORMAT = (TYPE = 'CSV' SKIP_HEADER=1 FIELD_OPTIONALLY_ENCLOSED_BY ='"');

USE SCHEMA SCD1_DB.PUBLIC;

SHOW PIPES;

USE ROLE ACCOUNTADMIN;

 

USE WAREHOUSE COMPUTE_WH;

 

USE SCHEMA SCD1_DB.PUBLIC;

CREATE OR REPLACE STREAM SCD1_DB.PUBLIC.CUSTOMER_STREAM

ON TABLE SCD1_DB.PUBLIC.CUSTOMER_SOURCE

APPEND_ONLY = TRUE;

SELECT * FROM SCD1_DB.PUBLIC.CUSTOMER_STREAM;

CREATE OR REPLACE TABLE SCD1_DB.PUBLIC.CUSTOMER

(

CONTACTFIRSTNAME STRING,

CONTACTLASTNAME STRING,

CUSTOMERNAME STRING,

PHONE STRING,

ADDRESSLINE1 STRING,

ADDRESSLINE2 STRING,

CITY STRING,

STATE STRING,

POSTALCODE STRING,

COUNTRY STRING,

TERRITORY STRING,

INSERT_DTS TIMESTAMP(6),

UPDATE_DTS TIMESTAMP(6)

);

CREATE OR REPLACE PROCEDURE SCD1_DB.PUBLIC.CUSTOMER_SP()

RETURNS VARCHAR(50)

LANGUAGE JAVASCRIPT

EXECUTE AS CALLER

AS

$$

try {

 

//Create statement BEGIN, Begins a transaction in the current session

 

snowflake.execute({sqlText:`BEGIN TRANSACTION;`});

 

//load data from Customer Stream to a temp table

    

snowflake.execute({sqlText:`CREATE OR REPLACE TEMPORARY TABLE SCD1_DB.PUBLIC.CUSTOMER_TEMP

AS

SELECT

CONTACTFIRSTNAME,

CONTACTLASTNAME,

CUSTOMERNAME,

PHONE,

ADDRESSLINE1,

ADDRESSLINE2,

CITY,

STATE,

POSTALCODE,

COUNTRY,

TERRITORY,

CURRENT_TIMESTAMP(6) AS INSERT_DTS,

CURRENT_TIMESTAMP(6) AS UPDATE_DTS

FROM 

SCD1_DB.PUBLIC.CUSTOMER_STREAM;`});

    

//Perfom the required SCD1 logic on the Customer Target table based on the primary column

 

snowflake.execute({sqlText:`MERGE INTO SCD1_DB.PUBLIC.CUSTOMER TGT

USING SCD1_DB.PUBLIC.CUSTOMER_TEMP TMP

ON TGT.CONTACTFIRSTNAME = TMP.CONTACTFIRSTNAME

AND TGT.CONTACTLASTNAME = TMP.CONTACTLASTNAME

 

WHEN MATCHED THEN UPDATE SET    

TGT.CUSTOMERNAME = TMP.CUSTOMERNAME,

TGT.PHONE = TMP.PHONE,

TGT.ADDRESSLINE1 = TMP.ADDRESSLINE1,

TGT.ADDRESSLINE2 = TMP.ADDRESSLINE2,

TGT.CITY = TMP.CITY,

TGT.STATE = TMP.STATE,

TGT.POSTALCODE = TMP.POSTALCODE,

TGT.COUNTRY = TMP.COUNTRY,

TGT.TERRITORY = TMP.TERRITORY,

TGT.UPDATE_DTS = TMP.UPDATE_DTS

            

WHEN NOT MATCHED THEN INSERT 

(

CONTACTFIRSTNAME,

CONTACTLASTNAME,

CUSTOMERNAME,

PHONE,

ADDRESSLINE1,

ADDRESSLINE2,

CITY,

STATE,

POSTALCODE,

COUNTRY,

TERRITORY,

INSERT_DTS,

UPDATE_DTS

)

VALUES 

(

TMP.CONTACTFIRSTNAME,

TMP.CONTACTLASTNAME,

TMP.CUSTOMERNAME,

TMP.PHONE,

TMP.ADDRESSLINE1,

TMP.ADDRESSLINE2,

TMP.CITY,

TMP.STATE,

TMP.POSTALCODE,

TMP.COUNTRY,

TMP.TERRITORY,

TMP.INSERT_DTS,

TMP.UPDATE_DTS

);`});

 

//Create statement COMMIT, Commits an open transaction in the current session

 

snowflake.execute({sqlText:`COMMIT;`});

 

//Statement returned for info and debuging purposes

 

return "Store Procedure Executed Successfully";

}

 

catch (err)

{

    result = 'Error: ' + err;

    snowflake.execute({sqlText:`ROLLBACK;`});

    throw result;

}

$$;

CREATE OR REPLACE TASK SCD1_DB.PUBLIC.CUSTOMER_TASK

WAREHOUSE = COMPUTE_WH

SCHEDULE = '1 MINUTE'

WHEN SYSTEM$STREAM_HAS_DATA('SCD1_DB.PUBLIC.CUSTOMER_STREAM')

AS CALL SCD1_DB.PUBLIC.CUSTOMER_SP();

ALTER TASK SCD1_DB.PUBLIC.CUSTOMER_TASK RESUME;

SHOW TASKS;

Glue job sample data load python script

 import boto3

import requests

import base64

from botocore.exceptions import NoCredentialsError, PartialCredentialsError

# Example usage

github_url = "https://raw.githubusercontent.com/deacademygit/project-data/refs/heads/main/country_details.json"  # Raw file URL from GitHub

bucket_name = "sampleS3bucketname"  # The S3 bucket where the file will be uploaded

s3_key = "data/country_details.json"  # The key (path) in the S3 bucket

def fetch_data_from_github_and_upload_to_s3(github_url, bucket_name, s3_key):

    try:

        # Fetch the file content from GitHub (raw URL or GitHub API URL)

        response = requests.get(github_url)

        # Check if the request was successful

        if response.status_code == 200:

            # Get the content from the response (for raw URL, it's already plain text or binary)

            file_content = response.content

            # Initialize the S3 client

            s3_client = boto3.client('s3')

            # Upload the file content to S3

            s3_client.put_object(Bucket=bucket_name, Key=s3_key, Body=file_content)

            print(f"File uploaded successfully to s3://{bucket_name}/{s3_key}")

        else:

            print(f"Error: Failed to fetch file from GitHub. Status code {response.status_code}")

    except (NoCredentialsError, PartialCredentialsError) as e:

        print(f"Error: AWS credentials are missing or incomplete. {e}")

    except requests.exceptions.RequestException as e:

        print(f"Error fetching the file from GitHub: {e}")

    except Exception as e:

        print(f"Error uploading file to S3: {e}")

fetch_data_from_github_and_upload_to_s3(github_url, bucket_name, s3_key)

python script sample

 # Create a new DataFrame that includes only the title_name and genres columns, and remove the square brackets and spaces from the genres column

df = dim_titles_roku_python[['title_name', 'genres']].copy()  # make a copy to avoid modifying the original DataFrame

df.loc[:, 'genres'] = df['genres'].str.replace('[', '', regex=True).str.replace(']', '', regex=True).str.replace(' ', '', regex=True)

# df['genres'] = df['genres'].str.replace('[', '').str.replace(']', '').str.replace(' ', '')

# Use the `apply` function to split the genres column into a list of individual genres

df = df.apply(

    lambda x: pd.Series(x['genres'].split(',')),

    axis=1).stack().reset_index(level=1, drop=True)

# Use the `reset_index` function to create a new DataFrame where each genre is a separate row

df = df.reset_index(name='genre')

# Use the `groupby` and `size` functions to count the number of occurrences of each genre

df = df.groupby('genre').size().reset_index(name='frequency')

# Sort the resulting DataFrame by the frequency column in descending order

df = df.sort_values(by='frequency', ascending=False).head(10)

print(df)

Top & Bottom Categories each Year for Men and Woman based on Product Reviews

 df = dim_category_nike_python.copy()

#print(df)

product = dim_product_nike_python.copy()

product = product.merge(df,on='category_id',how='inner')

product = product[['order_date','gender','category_name','product_reviews']]

product['order_date'] = pd.to_datetime(product['order_date']).dt.year

product=product.groupby(['order_date','gender','category_name'])['product_reviews'].mean().reset_index()

product['top_rank']=product.groupby(['order_date','gender'])['product_reviews'].rank(ascending=False)

product['bottom_rank']=product.groupby(['order_date','gender'])['product_reviews'].rank(ascending=True)

new_df = product[(product['top_rank']==1)|(product['bottom_rank']==1)]

new_df = new_df.copy()

new_df.loc[:,'Category_type'] = new_df.apply(lambda row: 'Top' if row['top_rank']== 1 else 'Bottom',axis=1 )

grouped  =new_df.groupby(['order_date','gender','Category_type'])['category_name'].first().reset_index()

pivot_df = grouped.pivot(index='order_date',columns=['gender','Category_type'],values='category_name')

pivot_df.columns = [  f"{gender}_{abc}_Category"  for gender,abc in pivot_df.columns ]

pivot_df = pivot_df.reset_index()

column = {'men_Bottom_Category':'Men_Bottom_Category',

    'woman_Top_Category':'Woman_Top_Category',

    'woman_Bottom_Category':'Woman_Bottom_Category',

    'men_Top_Category':'Men_Top_Category',

    'order_date':'order_year'

    

}

pivot_df=pivot_df.rename(columns = column)

#print(product[product['order_date']==2020])

print(pivot_df)

python script to aggregate , merge, sort etc

 df= fact_orders_grubhub_python.merge(dim_delivery_person_grubhub_python, on='Delivery_person_ID')

# Use the `groupby` and `agg` functions to compute the average age and rating for each city_type

df = df.groupby('city_type').agg({

    'Delivery_person_Age': 'mean',

    'Delivery_person_Ratings': 'mean'

}).reset_index()

# Use the `round` function to round the age and rating columns to 2 decimal places

df['Delivery_person_Age'] = df['Delivery_person_Age'].round(2)

df['Delivery_person_Ratings'] = df['Delivery_person_Ratings'].round(2)

# Rename the columns to match the names in the SQL query

df.rename(columns={

    'Delivery_person_Age': 'avg_age',

    'Delivery_person_Ratings': 'avg_rating'

}, inplace=True)

print(df.head(5))

Qlik connection string switch by environment

 

 Environment var.

LET v.SYS.ServerName = UPPER(ComputerName());

 

//storing environment name into a varibale

LET v.SYS.EnvironmentName = if(index('$(v.SYS.ServerName)','DV'),'dev connection string name', 

                            if(index('$(v.SYS.ServerName)','UA'),'UAT connection string name'                   ));

LIB CONNECT TO  $(v.SYS.EnvironmentName);

Customer rating using SQL

 WITH CustomerRatings AS (

    SELECT

        customer_id,

        AVG(rating) AS avg_rating

    FROM

        wlmt_reviews

    GROUP BY

        customer_id

)

SELECT

    CASE

        WHEN avg_rating >= 4 THEN 'Promoter'

        ELSE 'Neutral' END AS segment,

    COUNT(*) AS customer_count,

    ROUND(COUNT(*) * 100.0 / (SELECT COUNT(*) FROM CustomerRatings),2) AS percentage

FROM

    CustomerRatings

GROUP BY

    segment

ORDER BY

    percentage DESC;

    

    

    

   

Find top 3 rated Netflix series using CTE

 WITH FiveRatedseries AS

    (

    SELECT

            c.series_id,

            series_name,

            COUNT(DISTINCT r.profile_id) AS TotalProfileCount,

            AVG(rating) AS AvgRating

    FROM            content c 

    INNER JOIN      series s ON s.series_id=c.series_id

    INNER JOIN      reviews_ntf r   ON r.content_id=c.content_id

    GROUP  BY c.series_id,

            series_name

   

    )

SELECT series_name,TotalProfileCount,ROUND(AvgRating,2)AS AvgRating

FROM FiveRatedseries

WHERE TotalProfileCount >=  5

ORDER BY AvgRating DESC

LIMIT 3

 

Find maximun difference in the order for the same customer SQL

 WITH CustomerData AS(

SELECT      customer_id,

            oz.order_id, 

            product_id,

            SUM(quantity)  AS purchase_quantity, 

            SUM(quantity) - LAG(SUM(quantity)) OVER(PARTITION BY customer_id,product_id ORDER BY order_id) quantity_diff

FROM        orders_amz oz

INNER JOIN  orderdetails o ON o.order_id=oz.order_id

GROUP BY customer_id,oz.order_id, product_id)

SELECT      c.customer_id, 

            p.product_id, 

            product_name

FROM        products_amz p 

INNER JOIN  CustomerData c on c.product_id=p.product_id

ORDER BY    quantity_diff DESC

LIMIT 1

 

CTE query to find top selling products

 WITH MonthlySales AS (

    SELECT

        p.product_name,

        s.store_id,

        SUM(s.units_sold) as units_sold,

        SUM(s.units_sold) OVER (PARTITION BY p.product_id) as total_units,

        RANK() OVER (PARTITION BY p.product_id ORDER BY SUM(s.units_sold) DESC) as store_rank

    FROM

        product p

    JOIN

        sales s ON p.product_id = s.product_id

    WHERE

        YEAR(p.launch_date)=2022

        AND s.sale_date BETWEEN p.launch_date AND DATE_ADD(p.launch_date, INTERVAL 1 MONTH)

    GROUP BY

        p.product_name, s.store_id

)

SELECT

    ms.product_name,

    st.location_city,

    st.location_country,

    ms.units_sold,

    ms.total_units,

    ROUND(100.0 * ms.units_sold / ms.total_units, 2) as store_influence_percentage

FROM

    MonthlySales ms

JOIN

    stores st ON ms.store_id = st.store_id

WHERE

    ms.store_rank = 1

ORDER BY

    store_influence_percentage DESC;

Year part extract in SQL

 SELECT 

p.product_name,

p.category,

SUM(s.units_sold) AS total_quantity_sold,

SUM(s.units_sold*s.unit_price) AS total_revenue

FROM product p

INNER JOIN sales s ON s.product_id=p.product_id

WHERE EXTRACT(YEAR FROM s.sale_date) = 2023

GROUP BY p.product_name

ORDER BY total_quantity_sold DESC 

LIMIT 5

Find the top-performing employee in terms of both revenue generated and projects completed in January 2023.

 SELECT e.employee_id, e.role

FROM employee_dimension e

JOIN employee_performance_fact ep 

ON e.employee_id = ep.employee_id

WHERE ep.month = 'January' AND ep.year = 2023

AND (ep.revenue_generated, ep.projects_completed) = (

    SELECT MAX(revenue_generated), MAX(projects_completed)

    FROM employee_performance_fact

    WHERE month = 'January' AND year = 2023)

Customer with the Highest Total Spending in Each City

 SELECT city, name, SUM(amount) AS TotalSpending

FROM customers_prc c

JOIN orders_prc o ON c.customerid = o.customer_id

GROUP BY city, name

HAVING TotalSpending = (SELECT MAX(TotalSpending) 

FROM (SELECT city, name, SUM(amount) AS TotalSpending 

        FROM customers_prc c JOIN orders_prc o ON c.customerid = o.customer_id GROUP BY city, name) AS Subquery 

WHERE Subquery.city = c.city);

Qlik filter values into comma separated values

 

 

Replace(

Concat({<[Aggregation Level 1] = p([Aggregation Level 1])>} [Aggregation Level 1], ',')

,'Region','region_name') 

to use the list as values/column name in the expressions

$(=$(Aggregationlist))

chiclet slicer to keep drill through selection for more than one page

Here is the video to update /keep drill through selection on for as many as pages we want. 

https://www.youtube.com/watch?v=uq0FozPBidE&t=273s

Date format with time zone in Power BI

Choose the date field and enter below expression in the dropdown. it will accept and change the format accordingly. 

my sample data is in EST and no hours difference and I just hardcoded EST at the end. conversion needed if the data tracked in the different time zone. 

mm/dd/yyyy hh:mm:ss AM/PM "EST"

Power BI slicer or filters are not working in another page

 

Sync filters by choosing a filter in any page and update as needed. 

Power BI Rank & Dense Rank

 Rank = RANKX(all(Sheet1),Sheet1[Column1],,DESC)

DenseRank = RANKX(all(Sheet1),Sheet1[Column1],,DESC,Dense)

Power BI DAX measure to lookup value from other table without Merging two tables

 Lookup_Name = LOOKUPVALUE(details[Name],details[SId],Dataset1[SId])

Here we are creating this measure under Dataset1 to pickup Name field from details table.

Note: If we have more than one matching row for the same ID, either one or second row value will be populated.

Power BI Measure - DAX query to count string values

 Total =

VAR Count1= [field1]

VAR Count2=[field2]

VAR T1= SWITCH(TRUE(), SELECTEDVALUE(dataset1[IsRequired])="Yes",Count1, SELECTEDVALUE(dataset1[IsRequired])="NA",Count2)

Return T1