Intro :-

Hello, my name is Shraddha Gangadas Amrutwar.

I completed my Bachelor of Technology.

From MIT, Aurangabad, and I have 4+ years of experience.

working with Google Cloud Platform (GCP).

I started my career in Jan 2020.

at Ventrek Solutions as a Data Engineer,

where I worked for one year on

BigQuery data warehousing,

focusing on data extraction,

transformation, and running queries.

In 2022, I joined Infosys Ltd as a system engineer.

where I took on the role of a data engineer.

Over the years, I have gained expertise.

in building ETL pipelines using

Dataflow, Dataproc, BigQuery, Airflow, and GCS,

where I have written and optimized scripts

for large-scale data processing. Additionally,

I have worked with Git and ServiceNow.

I have hands-on experience designing and implementing

scalable data solutions for the retail industry,

I have hand on SQL, Python, and PySpark,

leveraging Google Cloud Platform (GCP) services

such as Google Cloud Storage (GCS),

Dataproc, Dataflow, Cloud Composer (Apache Airflow),

and BigQuery to build scalable and efficient data pipelines.

I am excited about opportunities where

I can leverage my expertise in GCP and data engineering

to solve complex business problems.

my role is to transform and process data

received from the data migration team.

I clean, validate, and optimize data using PySpark,

handling duplicates, skewed data, and schema evolution.

I apply business rules(Apply a 5% markup on high-demand regions., Exclude products with

latest price using window functions .), implement window

functions for aggregations, and ensure efficient

data storage in BigQuery. Additionally, I optimize transformations

using partitioning, bucketing, and Airflow monitoring

to maintain smooth pipeline execution.

Retail Projects : Client ::-- IGA ,US

What is the business : -

The business was a retail company implementing

region-based pricing to optimize revenue and

market competitiveness. Prices varied across regions

due to factors like tax differences, transportation costs,

and local demand trends. However, incorrect pricing in

some stores led to revenue loss, compliance issues,

and frequent refunds. My role was to process and transform

pricing data, ensuring accurate adjustments, stock validation,

and effective discount application using

PySpark and BigQuery

I build and optimize data pipelines

that collect, clean, and process

pricing data from various sources, such as sales,

trends, and customer demand.

This data is then analyzed to adjust prices

dynamically across different regions.

Business Impacted by :-

Revenue Loss :

Customer Trust Issues :

Operational Overhead :

Frequent refund requests due to incorrect charges

After the Pipeline :-

By implementing an automated

pricing adjustment system, we improve

profit margins, increase sales in competitive

regions, and prevent revenue loss in

low-demand areas. The key KPIs I track include

revenue impact per price change,

profit margin by region, and stock turnover rate.

Recently I was working for the retail domain

We have the separate data migration team who

were responsible for loading data from

client to shared GCS

locations, and some files need to be extracted from the

their database using Cloud SQL,

So there are some main applications that are offered.

to data received from different applicatins

like SAP CRM, ORACLE, SQUARE POS,

SAPS/4 HANA Retail, etc.

We had 4 Layer architecture.

There are different terms like

Atomic layer.

Data Refined Layer,

Data Storage Layer,

TemporarystageLayer

Atomic layer :

So basically we have the access of

Data extraction from shared bucket to our project

Bucket with the help of gcp commands

gsutil -m cp -r gs://shared-bucket-name/path-to-files

gs://your-bucket-name/

we were able to read the data.

by using the pyspark spark. read. Read options

Access :- [Link].

We had got 9 different files like customer data file,

There are the different types of file

format Like csv,parquet,json,avro etc

Data Refined Layer

read from Atomic layer and

processed data like standardizing the date,

handling missing values, schema evaluation,

special characters, data validation, etc.

and segregate the bad record and

good records after that

Write the file into the parquet format

and uploaded into the ETL layer of the GCS bucket

From there we can read the data.

Data Storage Layer

Did transformations like

average data

using window functions denserank, rownumber,

revenue for each inventory,

and load into the storage layer in parquet format

and created a dataset and table in BigQuery.

TemporaryStageBucket :-

We were created the temporary bucket

For storing the temp bucket before uploading

Dataframe to the bigquery .

So there is the reason behind it that is

When we process the data it partitioned

Into chunks so before the loaded into

Bigquery we need the collect and convert into

Structured format.
KPI :-
Daily Routine :-

As a Data Engineer, my day starts by

checking Airflow DAGs and monitoring ETL

pipelines for failures. If any job fails, I debug logs

and re-run tasks. In stand-up meetings, I discuss

progress, blockers, and upcoming tasks. My core work

involves developing and optimizing PySpark

pipelines on Dataproc, transforming data for

BigQuery, and ensuring data quality.

For example, I recently optimized a skewed dataset

by implementing bucketing, reducing job runtime

by 40%. Before logging out, I review deployments,

update documentation, and plan the next steps.

Opeator and use cases :-

For security purpose in GCS ,

We had used roles like Storage Admin,

Viewer, and Editor.

And assigned access at project, bucket,

or object level.

-- Signed URLs

-- GCS Security = IAM + Signed URLs + ACLs +

Encryption + Logging.

In Bigquery

(Policy tag = Finance and Marketing teams

,Data Masking on card_number)

Scenario:
-- A retail company stores

customer data (customer _name,card_number,

email, phone,

purchase history) in BigQuery.

Only the Finance and Marketing teams

should access certain fields

Using policy tag.

--Supports roles like Viewer, Editor,

and Owner

BigQuery Security = IAM + Policy Tags +

Row-Level Security + Encryption + Logging

[Link] (Reducing Partitions to Avoid Small Files) :- Region

2. Bucketing (Optimizing Joins and Shuffle) :- Store_ID, Product_ID

3. Partition By :- Sale_Date

4. Cluster By :- Region, Product_ID

5. Window Functions (Ranking & Aggregation) :-

Finding most sold products → dense_rank().

Calculating revenue impact → sum() over partition.

SCD Type 2 history tracking → lag(), lead().

6. Pivot (Reshaping Data):- Region, Sale_Date, Revenue.:--

7. to show monthly revenue per region.

8. Rollup :- Total revenue at different levels (Region, Store, Product):-

[Link]("Region", "Store_ID", "Product_ID").sum("Revenue")

9. Listagg (String Aggregation):- Combining customer purchase

history into a single row.

df = [Link]("Customer_ID").agg(F.concat_ws(", ", F.collect_list("Product_Name")))

For autoscaling need to create the policy

Master Node (1 Node)

Machine Type: n2-standard-4 (4 vCPUs, 16 GB RAM)

Disk Type: SSD (faster performance)

Disk Size: 100 GB

Worker Nodes (4-6 Nodes)

Machine Type: n2-standard-4 (4 vCPUs, 16 GB RAM)

Number of Workers: 4-6 (based on concurrency needs)

Disk Type: SSD

Disk Size: 200 GB per worker (to handle shuffle and intermediate storage)

Preemptible Workers (Optional)

Add 2-3 preemptible workers (n2-standard-4) to reduce cost

While transformations :-

Handling Skewness :- Use BucketingBy,Coalecse,Serialization

Use Bucketing, Coalesce.

Duplicate Records :- Use DISTINCT, ROW_NUMBER(),

or Deduplication with Window Functions.

Data Type Mismatch :- Convert data types during transformation

Data Quality Issues :- Replace NULLs with default values,

Convert all dates to a standard format during ingestion

Late-arriving Data :- late arrivaing separate bucket

Failed Job :- Implement Retry Logic, Checkpoints,

and Failure Alerts.

While Data Loading :-

Schema Evolution :- Enable ALLOW_FIELD_ADDITION while loading data

Partitioning Strategy :- Partition by frequently used filter columns.

PYSPARK :-3.2x

BEAM :-

AIRFLOW :-

GOOGLE_CLOUD :-

SQL:-plsql

PYTHON :-3.9

TEAM SIZE :- 6 (1 Manager,1 BA,1 lead,2 DE,1 TE)

FILE SIZE :- 500MB-5GB

Numebr of incoming table TABLE :- 09

[Link] Column :-21()

Bigquery :

Table :- SCD1,SCD2,

Airflow Operator :-

ExternalSensorOperator

DataProcCreateClusterOperator

DataProcSubmitJobOperator

DataProcDeleteJobOperator

EmailOperator

TriggerDagrunOperator
Optimizations :

Optimize Data Processing (PySpark)

Use Column Pruning – Select only the

required columns instead of SELECT *.

Use Partitioning & Bucketing – Partition by Sale_Date

and Region to improve query speed.

Avoid Wide Transformations – Reduce expensive

operations like groupBy() and orderBy().

Use MERGE Instead of Overwrite

for SCD1 (Only update changed records).

Use LAG and LEAD for Efficient SCD2 Processing.

Use PARTITION BY Sale_Date

for time-based queries.

Use CLUSTER BY Region, Product_ID to

improve lookup performance.

Use SUM() with Window Functions

to reduce groupBy() operations.

Filter Data Before Aggregation – Avoid unnecessary

joins before transformation.

SCD Type 2 Algorithm (Simplified Steps)

[Link] if the product exists in the SCD2 table.

If not, insert a

new record with Effective_Start_Date = CURRENT_DATE() and Is_Current = TRUE.

[Link] the new price with the existing price.

If the price is unchanged, do nothing.

[Link] the price has changed,

update the old record by setting

Effective_End_Date = CURRENT_DATE() and Is_Current = FALSE.

[Link] a new record

with the updated price, Effective_Start_Date = CURRENT_DATE(),

Effective_End_Date = NULL, and Is_Current = TRUE.

[Link] for each incoming record,

ensuring only one active record (Is_Current = TRUE)

per product while keeping historical data.

from [Link] import SparkSession, Window

import [Link] as F

from [Link] import *

# Initialize Spark Session

spark = [Link].

appName("RetailPricingETL").getOrCreate()

# BigQuery Configuration

PROJECT_ID = "your-gcp-project"

DATASET = "retail_pricing"

BQ_TABLE_SILVER = f"{PROJECT_ID}.{DATASET}.silver_pricing_data"

BQ_TABLE_SCD1 = f"{PROJECT_ID}.{DATASET}.historical_pricing_scd1"

BQ_TABLE_SCD2 = f"{PROJECT_ID}.{DATASET}.historical_pricing_scd2"
# =================== 1. Load Data from Various Sources ===================

pos_sales = input_path = "gs://your-bucket-name/region_pricing.csv"

df = [Link]("header", True).csv(input_path)

# =================== 2. Data Cleaning ===================

cleaned_data = pos_sales.dropDuplicates().

[Link]({"Base_Price": 0, "Sale_Date": "2024-01-01"})

# Convert Sale_Date to Date Type

cleaned_data = cleaned_data.

withColumn("Sale_Date", F.

to_date("Sale_Date", "yyyy-MM-dd"))

# ==============3. Join Tables for Complete Business Data ===================

full_data = cleaned_data \

.join(pricing_rules, ["Store_ID"], "left") \

.join(inventory, ["Product_ID"], "left") \

.join(promotions, ["Product_ID"], "left") \

.join(logistics, ["Product_ID"], "left")

# =================== 4. Calculate KPIs ===================

full_data = full_data.withColumn("adjusted_price", F

.col("Base_Price") * 1.05) # 5% markup

full_data = full_data.withColumn("revenue", F.

col("adjusted_price") * [Link]("Available_Stock"))

full_data = full_data.withColumn("cost", F.

col("Available_Stock") * 0.1 + [Link]("Transportation_Cost"))

full_data = full_data.withColumn("profit_margin", ((F.

col("revenue") - [Link]("cost")) / [Link]("revenue")) * 100)

# Revenue Impact per Price Change

window_spec = [Link]("Region")
 full_data = full_data.withColumn("revenue_impact", F

.sum([Link]("adjusted_price") – F

.col("Base_Price")).over(window_spec))

# Customer Retention Rate

full_data = full_data.

withColumn("customer_retention_rate", (F

.col("Discount_Percentage") / 100) * 100)

# Stock Turnover Rate

full_data = full_data.withColumn("stock_turnover_rate", F

.col("Available_Stock") / [Link]("Reorder_Threshold"))

# Discount Effectiveness

full_data = full_data.withColumn("discount_effectiveness", (F

.col("Discount_Percentage") / 100) * 100)

# =================== 5. SCD Type 1 - Overwrite Data ===================

df_scd1 = [Link]("Updated_Timestamp",

current_timestamp())

df_scd1.[Link]("bigquery") \

.option("table", "your_project.

your_dataset.region_pricing_scd1") \

.mode("overwrite") \

.save()

# =================== 6. SCD Type 2 - Maintain History ===================

window_spec = [Link]("Store_ID", "Product_ID").orderBy(col("Updated_Timestamp"))

# Add LAG and LEAD columns for

previous & next record comparison

df_scd2 = scd1_df.withColumn("Prev_Recommended_Price", lag("Recommended_Price").over(window_spec))

.withColumn("Next_Recommended_Price", lead("Recommended_Price").over(window_spec)) \

.withColumn("Effective_Start_Date", col("Sale_Date")) \
.withColumn("Effective_End_Date", lead("Sale_Date").

over(window_spec)) \

.withColumn("Is_Current",

when(col("Next_Recommended_Price").

isNull(), lit(1)).otherwise(lit(0)))

# Set End Date for Historical Records

df_scd2 = df_scd2.

withColumn("Effective_End_Date",

when(col("Is_Current") == 1, lit(None)).

otherwise(col("Effective_End_Date")))

#Upload SCD2 to BigQuery

df_scd2.[Link]("bigquery") \

.option("table", "your_project.your_dataset.

region_pricing_scd2") \

.mode("append") .sav