mm Good morning/ Afternoon,

My name is Pawan Bhise. I have Completed my Graduation from Pune University. I

have Total 4.2 years of experience in Data Engineering. During this period, I have gained
hands-on experience with various technologies like SQL, Python, PySpark, Snowflake . I
have relevant experience on various Azure Services like (ADF,ADB,ANH,Azure monitory,
Azure Logic app, Azur active directory, Azure Functions etc)
I Worked At Freekyat Systems, for Allen and hanbury Client. It is a pharmaceuticals
and biotechnology company, Basically on a pha
rma domain project. Subdomain is Drug manufacturing and Production in that in
quality control module
I am working on the Pharma Manufacturing Department and in Pharma
Manufacturing Department I Working on Quality Assurance Node. [to reduce production cost
while increasing product quality and operational efficiency through formulation
optimization].
For this project, we were using various Azure Cloud services. We use ADLS Gen2 for
data storage, ADF for development and Testing purposes, ADF for ETL and Data
Transformations, Snowflake for Data warehousing & analysis, Azure monitor for Monitoring,
ANH for the Notification and ADF for automating and Managing The workflow.

My Roles includes:

 ADF ETL Script Development: ETL Processes for Data consistency, Quality and
Integrity.
 Analyzing and Converting Business Requirement into Technical Requirement.
 Developing and Troubleshooting Data Pipeline: Extracting Da ta from Various
Sources, transforming and cleansing the data and loaded it into Datawarehouse’s.
 Created and Managed ADF using ETL Pyspark Scripts for efficient data
transformations
 Developed data processing workflow using PySpark, significantly improving
processing speed and efficiency.
 Data Validation and Testing: Data Validation using SQL Queries, scripting Languages,
Data Quality Checks includes Data Uniqueness, Completeness, identify missing
values, data duplication and using regular expressions and validation script to perform
these checks.
 Ensuring data quality and monitoring
 Writing Transformations Script.
 Automated ETL Pipelines using ADF, Ensuring smooth batch job orchestration and
data flow

Snowflake Roles and Responsibilities:

• Write, optimize, and troubleshoot SQL queries within the Snowflake environment.
• Integrate Snowflake with various data sources like ADLS GEN2.
• Loading data from ADLS GEN2 to Snowflake Datawarehouse.
• Ensure data security and compliance with industry standards.
The Aim of This Project Is to Analyze the Data for Different Purposes-(Bussiness
requirement) ---- (client requirement) --------
 Reduce production costs while increasing product quality and operational efficiency
through formulation optimization.
 Improve quality checks for dissolution rate, hardness, impurity levels, and strength.
 Increase Efficiency of a Medicines
 To track post Treatment efficiency of medicines.
 To split the result on different criteria like age, gender, etc.
 To find out least recovery rate and Drugs effectiveness.
 To Reduce the time Required for a medicines

Should I explain my high level project flow ??

Project workflow-

There is injection team who Gathers the data from different data sources (like oracle
database, INTRANET applications like LIMS,CDS,MES) and dump the data into ADLS
GEN2 CONTAINER, we Get that data into csv format on daily basis.
Actually, my works start with ADLS GEN2 CONTAINER where I Get data in csv format
on daily basis.
Then I create a ADB for preprocessing like cleansing and data validation like (null
validation, schema validation, remove duplicate, filling missing value, special character
validation).
Here I segregate with Good and bad records. Then I move this record into respective
CONTAINER.
As per client requirement we go with good record and do transformation like filtering,
grouping, mapping, joins, aggregate function IN ADB.
Then move the records to the another ADLS GEN2 CONTAINER that is gold layer. And
then by using Snowpipe I loaded data ADLS GEN2 CONTAINER to snowflake
Datawarehouse.
First of all, implementing SCD type2, insert current data and update historical data, if data
exists then update if not insert.

Example:

Task: Clean and filter the raw pharmaceutical data to focus on good records. Specifically:

 Filter out records where Batch_Status = 'Failed'.

  Handle missing values in key columns (e.g., dissolution_av, impurities_total).
 Correct the data types for specific columns (e.g., hardness to Newtons).
 Aggregate the cleaned data to produce insights on drug batches.
 Calculate the average impurity levels (AvgImpurity) and tensile strength
(TensileStrength) for all valid batches.
 Use Snowpipe to automate the loading of aggregated data from the Gold Layer in S3
into Snowflake for analysis.

Step-by-Step Implementation:

1. Clean and Filter the Raw Data:

First, we load the raw data from the Raw Layer and apply transformations to clean and filter
the dataset.

 Filtering out records: We remove batches where Batch_Status = 'Failed'.

 Handling missing values: Fill missing values in critical columns like
dissolution_av and impurities_total.
 Correcting data types: Convert hardness values to Newtons.

python
Copy code
from [Link] import col

# Load raw data from S3 (Raw Layer)

raw_df = [Link]("csv").option("header",
"true").load("s3://pharma-project/raw/laboratory_data.csv")

# Step 1: Filter out records where Batch_Status = 'Failed'

filtered_df = raw_df.filter(col("Batch_Status") != 'Failed')

# Step 2: Handle missing values by filling them with defaults

clean_df = filtered_df.fillna({"dissolution_av": 0, "impurities_total": 0})

# Step 3: Convert hardness to Newtons

clean_df = clean_df.withColumn("hardness_newtons", col("tbl_av_hardness") *
9.81) # Assuming hardness was in kgf

2. Aggregate the Cleaned Data:

After cleaning the data, we aggregate it to compute insights such as average impurity levels
and tensile strength for all valid batches.

 Average Impurity Levels (AvgImpurity)

 Tensile Strength Calculation: Use the formula σ=2∗Fπ∗t∗d\sigma = 2 * \frac{F}{\
pi * t * d}σ=2∗π∗t∗dF where F is the hardness in Newtons, t is the tablet thickness,
and d is the diameter.

python
Copy code
from [Link] import avg, expr

# Step 4: Calculate average impurity levels and tensile strength

aggregated_df = clean_df.groupBy("Batch_ID") \
 .agg(
avg("impurities_total").alias("AvgImpurity"),
expr("2 * tbl_av_hardness / (PI() * tablet_thickness *
diameter)").alias("TensileStrength")
)

3. Save the Aggregated Data to S3 (Gold Layer):

After calculating the metrics, we save the results to the Gold Layer in S3 for further analysis.

python
Copy code
# Step 5: Write the aggregated data to S3 (Gold Layer)
aggregated_df.[Link]("csv").option("header",
"true").save("s3://pharma-project/gold/aggregated_data.csv")

4. Automate Loading to Snowflake with Snowpipe:

Now that the cleaned and aggregated data is in the Gold Layer (S3), we use Snowpipe to
automate the loading of this data into Snowflake. Snowpipe continuously listens to the S3
bucket and automatically loads the new data into the appropriate Snowflake tables.

 Configure Snowpipe: You would configure Snowpipe to listen to the Gold Layer in
S3 for new data and automatically load the aggregated data into Snowflake.

Here's an example of how Snowpipe can be configured (assuming it's already set up in your
environment):

sql
Copy code
-- Step 6: Create Snowpipe to automate loading from S3 to Snowflake
CREATE OR REPLACE PIPE pharma_pipe
AUTO_INGEST = TRUE
AS
COPY INTO pharma_aggregated_data_table
FROM @my_s3_stage/gold/aggregated_data.csv
FILE_FORMAT = (TYPE = 'CSV' FIELD_OPTIONALLY_ENCLOSED_BY = '"');

Summary of Example:

 Data Cleaning: Filter out failed batches, handle missing values, and convert hardness
to Newtons.
 Aggregation: Compute the average impurity levels and tensile strength for valid
batches.
 Automated Loading: Save the aggregated data to the Gold Layer in S3 and automate
loading into Snowflake using Snowpipe for further analysis by end-users.
 Adls gen2 RAW Layer
Oracle Database
(CSV Format)
(Using JDBC-ODBC Driver and
Azure Credentials, Write SQL Daily Basis
Query to select data you want)

ADB

Preprocessing like Data Cleansing and Data validation (Null Validations, Schema Validations,
Remove Duplicates, Filling Missing Values, Special Character Validation)

Adls gen2 Silver Layer

Good Records and Bad Records

ADB

Transformations (filtering, grouping, mapping, joins,

aggregate functions)

Snowflake

[Link] Storage Integration Object

[Link] a Stage object using storage

integration object
Adls gen2
[Link] a copy command to load the
Gold Layer
data

[Link] a pipe by using copy command

[Link] event notifications at cloud

storage provider end (Azure notification
hub)
Flow of Project:

1. Initially the data is ingested from Oracle databases to adls gen2 which we called as
Raw layer.
[There was a migration team which handles migration of data from oracle to ADLS
GEN2 CONTAINER using JDBC and ODBC drivers. We received the data in CSV
format in adls gen2 on a Daily basis.]

2. This data further goes in pre-processing stages like data cleansing and validations
process.
[I used to do several validations like schema validation, null validations, filling
missing values, Special character validation.]

3. After this process we get good records and bad records. Then we dump the good
records into a ADLS GEN2 CONTAINER which we also called as Silver Layer.
[bad records are dumped into the another ADLS GEN2 CONTAINER which will go
further to get clarity from the client and convert those bad records into the Good
records.]

4. For further transformations on good records, we are using ADB.

Then we store the processed data into the S3 Bucket which we also called as Gold
Layer.
[In ADB, using Pyspark, we perform transformations on data frames to clean the data.
These transformations include operations such as grouping, mapping, joins, aggregate
functions, filtering, and other operations.]

5. Loading Transformed Data into Snowflake: After writing the validated data to the
new ADLS GEN2 CONTAINER, we used the Storage Integration in Snowflake to
load the data into the target Snowflake table for the End User. [also we can load data
using the Snowpipe, copy command, STORAGE connection, ADB script]

Loading data to snowflake by using snow pipe

STEPS:
1. Create Storage integration Object
2. Create stage object using storage integration object
3. Create and test COPY command to load data
4. Create PIPE by using COPY command
5. Setup event notification at cloud storage.

Steps for Loading Data to Snowflake using Snowpipe with Azure:

1. Create Storage Integration Object:
Create a Storage Integration object that allows Snowflake to interact
with your Azure Blob Storage securely.
 CREATE OR REPLACE STORAGE INTEGRATION demo_azure_int
TYPE = EXTERNAL_STAGE
STORAGE_PROVIDER = AZURE
ENABLED = TRUE
STORAGE_AZURE_TENANT_ID = '<azure-tenant-id>'
STORAGE_AZURE_CLIENT_ID = '<azure-client-id>'
STORAGE_AZURE_SECRET = '<azure-secret>'
STORAGE_ALLOWED_LOCATIONS = ('azure://<your-storage-
account>.[Link]/<container-name>')
COMMENT = 'Integration with Azure Blob Storage';
Explanation:
 STORAGE_AZURE_TENANT_ID: Azure Active Directory tenant ID.
 STORAGE_AZURE_CLIENT_ID: Client ID for your service principal.
 STORAGE_AZURE_SECRET: The secret (password) for your service
principal.
 STORAGE_ALLOWED_LOCATIONS: The path to the Azure Blob
container.

2. Create Stage Object Using Storage Integration:

Next, create a stage object that points to the Azure Blob Storage
location, using the storage integration you created in Step 1.

CREATE OR REPLACE STAGE azure_stage

URL =
'azure://<your-storage-account>.[Link]/<container-
name>'
STORAGE_INTEGRATION = demo_azure_int;
Explanation:
 URL: The path to the Azure Blob Storage container where your data
is stored.
 STORAGE_INTEGRATION: This refers to the integration object
created in Step 1.

3. Create and Test COPY Command to Load Data:

Create a table in Snowflake where the data will be loaded and use the
COPY command to manually test loading the data from the stage.

CREATE OR REPLACE TABLE my_table (

column1 VARCHAR,
column2 INT
);
Test COPY Command:
COPY INTO my_table
FROM @azure_stage/data_file.csv
FILE_FORMAT = (TYPE = CSV);
Explanation:
 The COPY command loads data from Azure Blob Storage (defined in
the stage) into the Snowflake table.
 data_file.csv is the file you're loading from the Azure container.

4. Create PIPE Using COPY Command:

Now, create a Snowpipe to automate the data loading process. The
Snowpipe listens for new files in Azure Blob Storage and automatically
loads them into Snowflake.

CREATE OR REPLACE PIPE my_pipe

AUTO_INGEST = TRUE
AS
COPY INTO my_table
FROM @azure_stage
FILE_FORMAT = (TYPE = CSV);
Explanation:
 AUTO_INGEST = TRUE: Enables Snowpipe to automatically load data
as new files arrive in Azure Blob Storage.

5. Setup Event Notification at Azure Cloud Storage:

To automate file ingestion, set up Event Grid Notifications or Azure
Functions to notify Snowflake when new files are added to the Blob
Storage container.
Steps:
 1. In Azure Blob Storage, configure Event Grid to send notifications
when a new file is added to the Blob container.
2. The Event Grid can send a message to a Webhook in Snowflake,
which triggers the Snowpipe to start loading the new data.
You can also configure an Azure Function that monitors the storage
container for new blobs and sends a request to Snowflake's REST API
to trigger Snowpipe.

Summary of the Process:

1. Create a Storage Integration that connects Snowflake to your Azure
Blob Storage.
2. Create a Stage object pointing to your Blob Storage location.
3. Test the COPY command to load data from the Blob container into
Snowflake.
4. Create a Snowpipe using the COPY command to automate the data
loading.
5. Set up Event Grid or Azure Functions to notify Snowflake when new
files are added to the Blob container.

Data Sources:
Data sources
 LIMS (laboratory information management system)
1. Raw data lab report
2. Intermediate lab report
3. Final product lab report

 DMS (documentation management system)

 Process data
 Process time series
 CMS (content management system)
1. API data
2. Excipient data

Requirement:

 Quality checks:
1. Dissolution check:
 each batch of drug should qualify dissolution range between 90-110
(table_name: [Link]
Column name: dissolution_av)

2. Hardness check:
Standardised unit of hardness column converting to Newton
1 dyne = 0.000010.00001 Newton
1lbf=4.448 Newton
1kgf= 9.81 Newton
(table name: [Link]
Column name: tbl_av _hardness)

3. Impurity level threshold check (data from HPCL test)

Impurities_total column should be > 0.2 to qualify batch.
(table name: [Link]
Column name: impurities_total)

4. Strength: strength is content of API per tablet

i. Cleaning the column and standardization of unit (SI)

Tensile Strength computation for data enrichment.

2∗f
σ=
( Pi∗t∗d )
F=Hardness (Newton)

T=Tablet thickness (in mm)

D= diameter (in mm)

(table_name: [Link]
Column name: tbl_av_hardness, av_hardness)

5. TimeFormat Standardization:(Process time series file)

Convert timestamp in any common form
6. Addressing data gaps: remove
7. Find total number of waste Total_waste:
a. Computed from waste column IN PROCESS TABLE

 Accomplishments:
 Reduce testing time and reduce product lead time
 Reduce cost of manufacturing

Data Transformations: Perform transformations like data type conversions, string manipulations, and Mathema
tical computations to prepare the data for analysis.

Challenges :
  Data Volume and Scalability: Processing large volumes of data can pose
challenges in terms of resource management, performance optimization,
and ensuring efficient utilization of ADF.
 Data Quality and Consistency: Dealing with data quality issues,
inconsistent data formats, and variations in data sources may require
additional effort in data pre-processing and cleansing stages.
 Performance Optimization: Optimizing query performance in Azure
Synapse Analytics requires careful consideration of data partitioning,
indexing, and using appropriate file formats like Parquet or ORC.
 Security and Access Control: Ensuring secure data transfer, data storage,
and proper access control to protect sensitive pharmaceutical data
throughout the data pipeline.
 Error Handling and Logging: Implementing robust error handling
mechanisms and comprehensive logging to capture and address any
failures or issues during data processing.
 Cost Optimization: cost of ADF and optimizing data storage costs in
ADLS GEN2 while maintaining performance and meeting business
requirements.

One challenge I encountered was with a dataset that had duplicate column
names. When I created a dataframe from this data, it understandably showed
duplicate column names. To resolve this, I used the columns function to fetch
the first occurrence of each column, effectively getting rid of the duplicates.
Then, I renamed this initial column to something unique. With this corrected
dataframe, I proceeded to perform further transformations and analyses as
needed.

The challenge you encountered with duplicate column names is common in scenarios where data is
extracted from multiple sources, such as in your project where data from Oracle databases or
Salesforce cloud is ingested into ADLS GEN2 in CSV format. This can happen due to various reasons,
like improper joins or merging datasets where column names are identical but hold different
meanings.

Here’s a deeper explanation of how you resolved this challenge:

The Issue:

When you created a DataFrame, it included duplicate column names. In Spark (or PySpark, which
you’re using), a DataFrame is expected to have unique column names. If it encounters duplicate
column names, Spark might not handle the data correctly or may throw errors during subsequent
transformations. This duplication could lead to ambiguity and data loss if not managed carefully.

Resolution Approach:

1. Identifying Duplicate Column Names: The first step was to recognize that duplicate column
names existed in the DataFrame, which can happen when two datasets are joined or when
data is ingested from inconsistent sources.

2. Fetching the First Occurrence: Using the columns attribute of the DataFrame, you can access
the list of all column names in the DataFrame. The idea here was to identify the first
occurrence of each column and keep it, effectively removing the duplicates.

Here’s how this could be done in PySpark:

python

Copy code

# Get the list of all column names in the DataFrame

column_names = [Link]

# Identify duplicates (if any)

unique_columns = []

duplicate_columns = []

for col_name in column_names:

if col_name not in unique_columns:

unique_columns.append(col_name)

else:

duplicate_columns.append(col_name)

print("Duplicate Columns:", duplicate_columns)

3. Renaming Duplicate Columns: Once you identified the duplicate columns, you renamed
them to something unique. This ensures that every column has a distinct name, preventing
conflicts during further transformations.

Example of renaming:

python

Copy code

# Rename the duplicate column with a suffix

for idx, dup_col in enumerate(duplicate_columns):

 df = [Link](dup_col, dup_col + "_dup" + str(idx))

This approach preserves the original data but ensures that all columns in the DataFrame are uniquely
named, avoiding issues with ambiguous references.

4. Continuing with Transformations: After renaming the duplicate columns, you proceeded
with your usual transformations and analyses. With each column now uniquely identified,
you could apply the necessary filtering, joins, aggregations, and calculations without Spark
throwing any errors or ambiguities.

Why This Is Important:

 Avoiding Ambiguity: Duplicate column names create confusion when trying to access or
manipulate specific columns. Renaming ensures you know exactly which column you're
referring to.

 Accurate Transformations: PySpark requires columns to have unique names to apply

transformations correctly. Without unique names, operations like filtering, grouping, and
joining might behave unpredictably.

 Maintaining Data Integrity: Renaming the columns allowed you to maintain the integrity of
the dataset while eliminating any confusion between columns that shared names but might
have contained different data.

Example Workflow:

1. Initial DataFrame with Duplicate Columns:

python

Copy code

+------------+------------+---------+

| batch_id | dissolution| dissolution|

+------------+------------+---------+

| 101 | 92 | 95 |

| 102 | 87 | 90 |

Here, the dissolution column appears twice due to data being merged or joined from different
sources.

2. Rename Duplicates:

python

Copy code

df = [Link]("dissolution", "dissolution_1") \

.withColumnRenamed("dissolution", "dissolution_2")

3. Resulting DataFrame with Unique Names:

python
Copy code

+------------+------------+------------+

| batch_id | dissolution_1 | dissolution_2 |

+------------+------------+------------+

| 101 | 92 | 95 |

| 102 | 87 | 90 |

This approach ensured that you could carry out further quality checks, transformations, and analyses
on your pharmaceutical data without errors related to duplicate column names.

Transformations:

# Quality checks
# Dissolution rate check
qualified_batches_dissolution = laboratory_df.filter((col("dissolution_av") >= 90) &
(col("dissolution_av") <= 110))

# Hardness check and unit standardization

laboratory_df = laboratory_df.withColumn("hardness_newtons", col("tbl_av_hardness") *
0.00001) # Convert to Newtons

# Impurity level threshold check

qualified_batches_impurity = laboratory_df.filter(col("impurities_total") > 0.2)

# Tensile Strength computation

laboratory_df = laboratory_df.withColumn("tensile_strength", 2 * (col("tbl_av_hardness") /
(3.14159 * col("tablet_thickness") * col("diameter"))))

# Time format standardization and data gaps

process_df = process_df.withColumn("timestamp", expr("cast(timestamp as timestamp)")) #
Convert to timestamp datatype
process_df = process_df.filter(col("waste").isNotNull()) # Remove rows with null values in
waste column

# Compute total waste

total_waste = process_df.agg({"waste": "sum"}).collect()[0][0]

# Output results or further processing

Print ("Qualified Batches based on Dissolution Rate:")
qualified_batches_dissolution.show()

# Convert timestamp to a common format

from datetime import datetime
 def convert_timestamp(timestamp):
try:
# Convert timestamp to a common format (e.g., "YYYY-MM-DD HH:MM:SS")
converted_time = [Link](timestamp, "%Y-%m-%d %H:%M:%S")
return converted_time.strftime("%Y-%m-%d %H:%M:%S")
except ValueError:
return None

convert_timestamp_udf = [Link](convert_timestamp, StringType())

process_df = process_df.withColumn("timestamp",
convert_timestamp_udf(col("timestamp")))

# Compute total waste

total_waste = process_df.selectExpr("sum(waste) as total_waste").collect()[0]["total_waste"]

# Output results
Print ("Converted Timestamps:")
process_df.show()

------------------------- L2, CLIENT, H3 QUESTIONS ----------

6. your team size?

Company Size : 150 - 200 employees

Team Size : 11
1 Manager
1 Team Lead
1 Solution Architect
3 Data Engineers ( 1 Senior and 2 of us )
3 BA
2 QA
---------------------------------------------------------------------------------------------------------------------------
----------------
7. how you deploy your code or have you worked on production environment?

No. I am working in development environment.

We develop the code locally using pycharm then we push that code to github.
after that we raise one PR (pull request) to review our code by seniors.
if its fine according to buissiness requirement then we merge with master branch.

-----
---------------------------------------------------------------------------------------------------------------------------
-----------
[Link] are the AZURE services you used in your project?
ADF,ADB,AZURE ACTIVE DIRECTORY, AZURE LOGIC APP, AZURE FUNCTION
---------------------------------------------------------------------------------------------------------------------------
----------------
[Link] is pii data?

Personally Identifiable Information.

it is sensitive information of user like adhar no, pan no, phone no.
data is alredy masked.

---------------------------------------------------------------------------------------------------------------------------
----------------
[Link] of tables and columns in your data?

40 to 50 tables. but my work is evolved around 12 to 15 table

5-6 tables name columns- around 25 to 30 column

---------------------------------------------------------------------------------------------------------------------------
----------------
[Link] of your data in your project?

Two times in a week from Monday to Friday. usually size is around 1 to 5 mb in json format.

---------------------------------------------------------------------------------------------------------------------------
----------------
[Link] versions (from documentation)

ADF ADB --
PYspark -- 3.3.2
python -- 3.8
snowflake - 7.30

---------------------------------------------------------------------------------------------------------------------------
----------------
[Link] is your upstream and downstream file format?

Upstream stream – ADLS GEN2 CSV file

Down stream – ADLS GEN2 PARQUET file

---------------------------------------------------------------------------------------------------------------------------
----------------

[Link] data size (mbs or gbs) (processing time?) 30-40?

we get data file twice in a week.

that file size is 10 to 25 mb.
after transforming that it will become 20 to 30 gb.

---------------------------------------------------------------------------------------------------------------------------
----------------
 [Link] of data have you handled or have you handled streaming data?

upto now, i handled only batch data but if i get chance to work in streaming data i will

---------------------------------------------------------------------------------------------------------------------------
----------------

[Link] is your cluster size?

- in our cluster we have 9 node with 3 namenode and 6 datanode.

- each namenode have 512 GB RAM and 3.4 TB HDD

- so RAM -- 512*3 = 1.5 TB RAM
HDD -- 3.4*3 = 10.2 TB HDD

- each datanode have 256 GB RAM and 9.4 tb HDD

- so RAM -- 256*6 = 1.5 TB RAM
HDD -- 9.3*6 = 55.8 TB HDD

we are not using any external cluster like emr.

we are using ADB so everything is in-built.

G1X (4 cpu , 16 GB ram,84 GB disk, 1 executor/node)

suppose we assign 2 node then our power is 2 DPU.

suppose we assign 10 node then our power is 10 DPU. (40 cpu,160 GB ram,840 gb disk)

--------------------------------------------------------------------------------------------------------------------------------------
-----

[Link] is data mapping?

Data mapping is the process of matching data fields from one source to data fields in another
source.

---------------------------------------------------------------------------------------------------------------------------
----------------

[Link] of your sprint?

Duration of sprint 15 [Link] Monday to friday only.

---------------------------------------------------------------------------------------------------------------------------
----------------

[Link] is waterfall model and agile methodology?

- Waterfall is a sequential model it contain some phases for information gathering to develop
the software
- agile is a continuous iterative model to develop the software.
It follows incremental approach agile allows to make changes in development where as in
 waterfall model is difficult to make changes in the development.

---------------------------------------------------------------------------------------------------------------------------
----------------

[Link] is your engagement platform?

- microsift teams
- stand up meating
- synk up call

---------------------------------------------------------------------------------------------------------------------------
----------------

[Link] did you get your project task?

Jira tool
Jira is a software which is agile project management tool. It supports agile methodology
We use scrum process
Agile support – scrum, kanban, xp ( xtream programming)

---------------------------------------------------------------------------------------------------------------------------
----------------

[Link] to dealing with corrupt records?

there ar e two ways to handle bad records

[Link] set mode to DROPMALFORMED

df=[Link]("csv").option("header","true").option("mode","DROPMALFORMED").lo
ad(data)
[Link]()

option("mode","PERMISSIVE") -- print corrupt record (default)

option("mode","FAILFAST") -- show error even 1 corrupt record also occur
option("mode","DROPMALFORMED") -- ignore corrupt record (use for corrupt record
handeling)

[Link] way is define schema manually and then set option

"columnNameOfCorruptRecord" and cache it.
pk_schema=StructType([StructField("id",IntegerType(),True),StructField("name",StringType(),
True),StructField("corrupt_records",StringType(),True)])

df=[Link](pk_schema).format("csv").option("columnNameOfCorruptRecord","c
orrupt_records").option("header","true").load(data).cache()

[Link]("corrupt_records").filter(col("corrupt_records").isNotNull()).show(truncate=False)

df=[Link]("id int,name string,corrupt_records

string").format("csv").option("columnNameOfCorruptRecord","corrupt_records").option("he
ader","true").load(data).cache()

[Link]("corrupt_records").filter(col("corrupt_records").isNotNull()).show(truncate=False)

---------------------------------------------------------------------------------------------------------------------------
----------------

[Link] you dealing with null values?

1. [Link]().show(2) or [Link]().show(2)

2. [Link]("SHASHIKANT").show(2) or [Link]("SHASHIKANT").show(2)

[Link] have coalesce,nvl functions in spark sql.

------------------------------------------------------------------------------------------------------------
------------------------------------------------------------------------------------------------------------

What is CICD PIPELINE ?

CI/CD (Continuous Integration and Continuous Deployment/Delivery) refers

to the automated processes used to ensure the efficient development,
testing, and deployment of data pipelines and systems.

Continuous Integration (CI): Automatically merges code changes into a

shared repository and runs automated tests to catch integration errors
early. It ensures that new code doesn't break existing code and maintains
quality.

Continuous Deployment/Delivery (CD): Automates the deployment of

validated code changes to production or other environments. After passing
tests, code is automatically deployed, enabling fast and reliable delivery of
updates.