PySpark is a Python API for Apache Spark, which is a fast and general-purpose

cluster-computing system. Here are some insights and use cases related to PySpark:
1. Data Processing and Job Scheduling:
o PySpark jobs can be scheduled to run at specific times to update
databases with new data. For example, a PySpark job might run daily
to update a PostgreSQL table with new item registrations and order
details 1 .
o PySpark can be used to process out-of-stock data by reading from
Kafka topics, joining data frames, and calculating impacts and
opportunities 1 .
2. Schema Evolution and Checkpointing:
o PySpark does not inherently support schema evolution in streaming
queries, which can lead to checkpoint mismatches. Solutions include
using versioned checkpointing or Delta Lake for automatic schema
handling 2 .
3. Performance and Optimization:
o Transitioning from Koalas to PySpark can improve performance in data
pipelines. PySpark offers better performance for large datasets and
complex transformations 3 4 .
4. Integration with EMR Serverless:
o PySpark can be run on Amazon EMR Serverless by packaging Python
libraries in a virtual environment and uploading them to S3. This setup
allows for the execution of PySpark jobs with specific configurations 5 .
5. Testing and Validation:
o PySpark transformations and SQL queries can be tested using Pytest
and Spark testing utilities. This ensures that the logic in PySpark
applications is correct and reliable 6 .
6. Feature Serving and Real-time Data:
o PySpark is used in feature-serving platforms, allowing data engineers
and scientists to work efficiently. It supports REST APIs for feature
serving and can handle real-time data processing 7 .

These insights highlight the versatility and power of PySpark in handling large-scale
data processing tasks, real-time data streaming, and integration with cloud services.

[Link] Access Request Comment

APP-PROD-EDP-CORE- I am working for the project Data Elixir and I will

1
DEVELOPER be requiring above access to continue working on
the data lake platform
APP-EDP-CORE-DEV-
2 Team needed access for UC
DEVELOPER
Oracle Payables India Expense
Kindly approve the access request for Oracle
3 Entry : 7-Eleven Internet Expense
Expense Entry
Entry - IND
I will require the access to this group in order to
4 MSA-RG-Dev-C360-Contrib
develop and orchestrate via Airflow
5 DIST-7EYP-GSC I am a new joiner, please approve
env = create_or_get_text_widget("env", "")

global_constants = EDPConstants(env)

constants = RDMConstants(env)

catalog_name = global_constants.BRZ_CATALOG

schema_name = global_constants.BRZ_SERVICENOW_API.split('.')[1]

table_name = f'{constants.BRONZE_SNOW_METRIC_INSTANCE_TBL}'

audit_table_name = f'{constants.BRONZE_SNOW_METRIC_INSTANCE_AUDIT_TBL}'

delta_path = f'{constants.BRONZE_PATH}'

[Link]('abfss://edp-
brz@[Link]/skipcart_sqlserver_dbo/
payment_ac_htransfers',True)

sql

DROP TABLE edp_brz_dev.skipcart_sqlserver_dbo.paymentachtransfers

env = create_or_get_text_widget("env", "")

global_constants = EDPConstants(env)

constants = RDMConstants(env)

catalog_name = global_constants.BRZ_CATALOG

schema_name = global_constants.BRZ_SKIPCART_SQLSERVER_DBO.split('.')[1]

bonus_log_table_name = f'{constants.BRONZE_BONUS_LOG_TBL}'

carrier_orders_log_history_table_name =
f'{constants.BRONZE_CARRIER_ORDERS_LOG_HISTORY_TBL}'

configuration_data_table_name =
f'{constants.BRONZE_CONFIGURATION_DATA_TBL}'
driver_pay_order_rates_table_name =
f'{constants.BRONZE_DRIVER_PAY_ORDER_RATES_TBL}'

driver_vehicle_table_name = f'{constants.BRONZE_DRIVER_VEHICLE_TBL}'

order_scans_table_name = f'{constants.BRONZE_ORDER_SCANS_TBL}'

configuration_log_history_table_name =
f'{constants.BRONZE_CONFIGURATION_LOG_HISTORY_TBL}'

payment_ach_transfers_table_name =
f'{constants.BRONZE_PAYMENT_ACH_TRANSFERS_TBL}'

delta_path = f'{constants.BRONZE_PATH}'

[Link](f"""

insert into {catalog_name}.{schema_name}.{config_table_name}

values ('order_scans','abfss://edp-brz@[Link]/
skipcart_sqlserver_dbo/order_scans','abfss://edp-
brz@[Link]/skipcart_sqlserver_dbo/
order_scans/_checkpoint/','10
minutes','/mnt/RAW_DATA/SKIPCART/skipcart_schema/order_scans.json','[Link]
[Link]','[Link]','e
arliest','false','edp_brz_dev.skipcart_sqlserver_dbo.order_scans',current_timestamp(),
current_timestamp(),'CreatedOn,UpdatedOn,CallbackRecdTimestamp,source_kafka_t
imestamp,kafka_timestamp') """)

%sql

select * from edp_brz_dev.skipcart_sqlserver_dbo.config

-------

1) delete the payment_ach record from config and insert it again with correct name

2) orderscans DDL re run it after changing the timestamp and reload the history data for that
file and change the timestamp columns ijn the config table

3) in the config table update the ConfigurationLogHistory record name correctly from listory

#sql

[Link]("DROP TABLE edp_brz_dev.skipcart_sqlserver_dbo.order_scans")-- drop a

table
[Link]('abfss://edp-brz@[Link]/
skipcart_sqlserver_dbo/order_scans',True)-- remove path from dbutils

display(final_dataframe)-- to view final data frame

%sql

select max(date(load_timestamp)) from

edp_brz_dev.skipcart_sqlserver_dbo.payment_ach_transfers

%sql

desc history edp_brz_dev.skipcart_sqlserver_dbo.order_scans

%sql

desc history edp_brz_dev.skipcart_sqlserver_dbo.configuration_log_history

-------

%sql

delete from edp_brz_dev.skipcart_sqlserver_dbo.config where inserted_timestamp="2024-

12-10T16:29:38.496+00:00" ---- delete timestamp from table

%sql

select config_id,count(*) from edp_brz_dev.skipcart_sqlserver_dbo.configuration_data group

by 1 having count(*)>1

%sql

desc history edp_brz_dev.skipcart_sqlserver_dbo.promotion_deliveries---- table

history

%sql

restore table edp_brz_dev.skipcart_sqlserver_dbo.promotion_deliveries version as of

2 -- restoring of a table

abfss://skipcart-7ep-inbound@[Link]/
Streaming/skipcartprod_publisher.[Link]/
abfss://skipcart-7ep-inbound@[Link]/
Streaming/skipcartprod_publisher.[Link]/
abfss://skipcart-7ep-inbound@[Link]/
Streaming/skipcartprod_publisher.[Link]/

drop the table

remove the table from the loc

load the history

start the incremental

create the schema - abbrevate the column name, and keep the columns in order

once the incremental data is inserted

start the silver workflow in your local

--------

have loaded payment_batch_history and also validated the data its looks good. You can use
below code snippet to validate the data.

%python

from [Link] import lit

# Select the specific row from each table

df2 = [Link]("SELECT * FROM

edp_brz_dev.skipcart_sqlserver_dbo.payment_batch_history WHERE driver_id = 1640")

df1 = [Link]("SELECT * FROM

edp_slv_dev.digital.fact_skipcart_payment_batch_history WHERE driver_id = 1640")

# Cast all columns to string

df1 = [Link]([df1[col].cast("string").alias(col) for col in [Link]])

df2 = [Link]([df2[col].cast("string").alias(col) for col in [Link]])

for col in [Link]:

 if col not in [Link]:

df2 = [Link](col, lit(None).cast("string"))

df2 = [Link]([Link])

# Perform a union of the sorted DataFrames

union_df = [Link](df2)

# Display the result

display(union_df)

validations silver--

select count(distinct offer_id) from edp_brz_dev.skipcart_sqlserver_dbo.driver_offers

where operation != 'd' and date(created_on) between '2025-01-15' and '2025-01-21' --- bronze

select count(*) from edp_slv_dev.digital.fact_skipcart_driver_offers

where delete_flag = false

and date(created_on) between '2025-01-15' and '2025-01-21' --- silver

select count(*) from edp_slv_dev.digital.fact_skipcart_driver_offers --- total count

select * from edp_slv_dev.digital.fact_skipcart_driver_offers

where date(created_on) between '2025-01-15' and '2025-01-21' --- checking the data

select count(distinct id) from edp_brz_dev.skipcart_sqlserver_dbo.payment_batch_history

--where operation != 'hist'

--date(created_on) between '2025-01-17' and '2025-01-22' ----

select count(*) from edp_slv_dev.digital.fact_skipcart_payment_batch_history

--where date(created_on) between '2025-01-16' and '2025-01-22'

%sql

select IsRetailerCancelled from ord where orderid="3679836" --- to get that colums results

%sql

select JobId, count(*) from test where date(CreatedOn) < '2025-02-05' group by JobId -- to
check count with gives PK values

%sql

select * from test where jobid in ('1243859', '1264785','1306984')

%sql

select * FROM edp_slv_dev.digital.fact_skipcart_jobs where job_id in ('1243859',

'1264785','1306984')

%sql

SELECT count(JobId) JobCount, COUNT(Distinct DriverId) DriverCount, Status FROM

test where date(CreatedOn) < '2025-02-06'GROUP by Status;

%sql

SELECT count(Job_Id) JobCount, COUNT(Distinct Driver_Id) DriverCount, Status FROM

edp_slv_dev.digital.fact_skipcart_jobs where

date(Created_On) < '2025-02-06' GROUP by Status;

%sql

select area_id, Area_Name, created_on, updated_on, source_kafka_timestamp, operation

from edp_brz_dev.skipcart_sqlserver_dbo.areas
--where area_id = 4 and source_kafka_timestamp = '2025-02-12 9:17:05.731+00:00'

where operation = 'u'

check PK must not be null

count of PK must be 1 not more than 1

do the rest of the validation what u did in the mrg

use excel to see the creatwed_on column

sum for the metric columns

PII columns must be encrypted

And finally our created_on,updated_on,kafka_json coliumns must be crt

Please check these also

1) source schema that is columns from source must be in brz with string datatypes and
timestamp as timestamps.

2) brz schema should be in order

3) Brz schema and slv schema should be same and the datatype of silver must be same as
source

4) the silver schema must be same in the cold_schema.txt files whatever we hv uploded in the
volume check that also order must not mismtach

1) after drivers again check all the DDL and schema and order and datatype of the table in brz
slv

2) And for all tables check if PII is there if it is encrypted or not. And check the config file of
slv and brz see whether the PII columns and timestamp columns are mentioned correctly or
not

3) if an PII column is timestamp that column must be in string in both brz and dev

validation on below rules:

1) %sql

select * from test where RegionareaId is null -- pk sholud not be null check

2) %sql
select distinct(RegionareaId), count(*) from test group by 1 having count(*)>1 -- Pk should
only one , not more than 1

3) check the sum of metrics of source and slv tables., take scrnshot. --- make all in a doc

from [Link] import col

df = [Link]("multiline", "true").json('abfss://skipcart-7ep-
inbound@[Link]/Streaming/
skipcartprod_subscriber_30Sept23.[Link]/')

df = [Link](col("data.*"), col("data").alias("kafka_json"))

display(df)

host = [Link](scope="key-vault-secrets", key=host_key)

user = [Link](scope="key-vault-secrets", key=user_key)

password = [Link](scope="key-vault-secrets", key=password_key)

port = [Link](scope="key-vault-secrets", key=port_key)

service = [Link](scope="key-vault-secrets", key=service_key)

# JDBC URL for Oracle DB connection

url = f"jdbc:oracle:thin:@//{host}:{port}/{service}"

jdbc_options = {

"driver": driver,

"url": url,

"user": user,

"password": password,

"query": "SELECT * FROM apps.onesource_use_tax_details",

"numPartitions": [Link]

}
df = [Link]("jdbc").options(**jdbc_options).load(). ---- display(df) --

# %sql

# truncate table edp_brz_dev.erp_l2dv_apps.onesource_use_tax_details

# %sql

# insert into edp_brz_dev.erp_l2dv_apps.onesource_use_tax_details

# select * from edp_brz_dev.erp_l2dv_apps.onesource_use_tax_details version as of

1

dQuoteHandling STOP_AT_DELIMITER Defines how the CsvParser will handle values

with unescaped quotes.

 STOP_AT_CLOSING_QUOTE: If unescaped
quotes are found in the input, accumulate
the quote character and proceed parsing
the value as a quoted value, until a
closing quote is found.
 BACK_TO_DELIMITER: If unescaped
quotes are found in the input, consider
the value as an unquoted value. This will
make the parser accumulate all characters
of the current parsed value until the
delimiter is found. If no delimiter is
found in the value, the parser will
continue accumulating characters from
the input until a delimiter or line ending
is found.
 STOP_AT_DELIMITER: If unescaped
quotes are found in the input, consider
the value as an unquoted value. This will
make the parser accumulate all characters
until the delimiter or a line ending is
found in the input.
 SKIP_VALUE: If unescaped quotes are
found in the input, the content parsed for
the given value will be skipped and the
value set in nullValue will be produced
instead.
  RAISE_ERROR: If unescaped quotes are
found in the input, a
TextParsingException will be thrown.

Sets a single character used for escaping quoted values where the separator can
be part of the value. For reading, if you would like to turn off quotations, you
read/write
" need to set not null but an empty string. For writing, if an empty string is set, it
uses u0000 (null character).

####### Oracle Credential

def getEbsData(queryFinal):

env = 'DEV'

try:

driver = "[Link]"

host = [Link](scope = "key-vault-secrets", key = "ED-{ENV}-ED-L2DV-

ORACLE-DB-HOST".format(ENV=env))

user = [Link](scope = "key-vault-secrets", key = "ED-{ENV}-ED-L2DV-

ORACLE-DB-USER".format(ENV=env))

password = [Link](scope = "key-vault-secrets", key = "ED-{ENV}-ED-L2DV-

ORACLE-DB-PASSWORD".format(ENV=env))

port = [Link](scope = "key-vault-secrets", key = "ED-{ENV}-ED-L2DV-

ORACLE-DB-PORT".format(ENV=env))

service = [Link](scope = "key-vault-secrets", key = "ED-{ENV}-ED-L2DV-

ORACLE-DB-SERVICE-NAME".format(ENV=env))

url = "jdbc:oracle:thin:@//{0}:{1}/{2}".format(host,port,service)
 srcDf = ([Link]("jdbc").option("driver", driver)

.option("url", url)

.option("user", user)

.option("password", password)

.option("query", queryFinal)

.option("numPartitions",[Link])

.load())

return srcDf

except Exception as e:

err = str(sys.exc_info()[0]) + " : "+ str(sys.exc_info()[1])

[Link]("Exception while connecting and reading data from oracle")

[Link](f"{e}")

raise err ---- L2DV

for i in [Link](scope = "key-vault-secrets", key = "ED-DEV-ORACLE-OFAP-

DB-USER"):

print(i,end=" "). ---- to retrieve secrets from KVS

------

Overview
Databricks is an all-in-one Data Platform, providing many core Data
Engineering and Data Science services. It is the core of EDP’s data
infrastructure, and we use many services within Databricks itself.

Specifically, Databricks has the following managed services in its Platform:

 Data Lakehouse with Delta Lake - All of your enterprise data in one
location
 Apache Spark - Distributed compute engine for DE and DS
workloads
  Python Jupyter Notebooks - For writing DE and DS workloads, and
submitting them to Spark
 Unity Catalog - A centralized catalog and governance layer for the
Lakehouse
 Mosaic AI - Hosted machine-learning models, with auto-generated
REST APIs

New Databricks WorkSpace Setup



Owned by Anjali Sinha, created with a template

Last updated: May 09, 2025
2 min read

13 people viewed Karma Page Builder Request approval

This document explains how can we set up a new Databricks workspace

for our Infrastructure.
Pre-Requisite
[Link] a request with CSS team to create a new Resource Group.

1.1Follow the naming convention for the new resource group.

1.2Provide the subscription details.

1.3Define the Tags:

APPOWNER: UmaMaheswaraRao.Dacharla2@[Link]
BUSOWNER: Shahmeer Ali Mirza, Ganesh Susarla
COSTCTR: 0000209
APPADMIN: APP-Azure-DDE-Admins
APPNAME: EDP

1.4Add the Roles

Add contribute role to APP-Azure-DDE-Admins AD group
and owner role to UmaMaheswaraRao.Dacharla2@[Link]

SNo Resource Reference Ticket Request Type Link

1 RG-DEV-ED-MERCH RITM1721590 Special Request special requ

[Link] the resource group is created drop an email to Cloud engineering

team<cloud-engineering@[Link]> for provisioning the Vnets/Subnets.

2.1 Share the Resource group details and the Subscription details with
Cloud Engineering team to proceed with the Vnet/Subnet Provisioning.
2.2Also share the Vnet/subnet address spaces that is required to be
provisioned.

Open Screenshot 2025-05-09 at 11.23.09 [Link]

For DEV the Vnet/Subnet can be provisioned in3 days from Cloud team.

For UAT/PROD Vnet/Subnet provisioning request is completed via CR.

Validation for the Vnets/Subnets:

[Link] the subscription of the new Vnet.

[Link] The Subnet Name, Security Group , Route Table.

Open Screenshot 2025-05-09 at 11.51.11 [Link]

[Link] the Service endpoints.

Open Screenshot 2025-05-09 at 11.53.13 [Link]

Once the Vnets and Subnet are

provisioned
[Link] a ticket with CSS team to create a new databricks workspace.

1.1Follow the naming convention for the new workspace.

1.2Share the Resource Group Details, Subscription Details,Vnet/Subnet

Details and Tags to be added to the new Workspace to CSS team .

1.3Tags should be similar to the Resource Group Tags.

SNoResource Reference Request Type Link

Ticket

1 ed-merch-dbks-dev- RITM1727333 Special Request special r

 01

After the new Databricks workspace is created raise a new firewall ticket
to enable all the firewall rules to the new workspace.

[Link] the InfoSec Jira Ticket .

[Link] the Source IP or FQDN.

[Link] the Destination IP.

[Link] the Destination Port/Destination URL/Firewall Application Name if

any.

Once the request is raised , we need to reach our to our Manger to

approve this request before it is taken care by Firewall team.

Open Screenshot 2025-05-09 at 12.17.51 [Link]

SNO Item Reference Ticket Request Type Link

1. Firewall Request SCTASK1765693 Firewall Request-NetSec Firewall req

After the Firewall request is completed , Launch the newly created

workspace from Azure.

Try to spin up a test-cluster , run few Adhoc commands to validate that

the workspace is functioning as expected.

Adhoc Command:

!telnet [Link] 443

 EDA Recommendation - Data Integration
Tools

Owned by Benjamin Sivoravong

Last updated: May 07, 2025
3 min read

13 people viewed Karma Page Builder Request approval

 Overview
 When to use a Data Integration Tool?
 When not to use a Data Integration Tool?
 Data Integration Tools
o Azure Data Factory
o Dell Boomi
o Striim
o Confluent Kafka Connect
o Oracle Golden Gate
o MFT

Overview
This doc provides EA recommendations on selecting a Data Integration
tool for any specific project.
NOTE: EA recommendations are just starting points. Any specific solution
architecture should be reviewed by peers.

When to use a Data Integration Tool?

The main use-case for Data integration tools in EDP is sourcing data from
OLTP systems. For example, an app team might have a SQL Server
running a core API service, which contains data that you want.

Instead of connecting to the SQL Server directly via JDBC or similar, you
can use a Data Integration tool to source records from that database into
a file or data lake. Most data integration tools use the underlying file
system of the database as the source of data, instead of the SQL
interface, making them much faster and more efficient than querying data
directly.

When not to use a Data Integration Tool?

Although data integration tools are very quick and efficient mechanisms
for sourcing data, all they do is directly copy OLTP data from the source
system into the analytics data lake. If you can get the application team to
directly produce analytical events from their app, that method would
almost always be preferred to reusing OLTP datasets for Analytical
workloads.

Data Integration Tools

Azure Data Factory

Pros:

 Baked into Azure natively, so it’s often the easiest path forward
when sourcing data from Azure datasources

Cons:

 May have issues connecting to non-azure datasources

 Not very good for orchestration or data transformation, just direct
data sourcing.

When to Use:
  Batch ingest from source systems that are in Azure (or accessible
from the Azure network)

Dell Boomi

Dell Boomi is a cloud-based integration platform-as-a-service (iPaaS) that

enables organizations to connect applications, data, and systems. It
provides a low-code interface for building integrations, automating
workflows, and managing APIs, making it easier to streamline data
exchange and business processes.

Pros:

Cons:

When to Use:

When not to Use:

Note:

 SNOW request for for new Boomi integrations - Requests |

ServiceNow

Striim

Striim is a real-time data integration platform that enables streaming data

pipelines with support for both historical and incremental (CDC) data
loads. It provides seamless integration across heterogeneous
environments and supports low-latency data movement and
transformation.

Pros:

 Cloud-Agnostic: Not tied to any specific cloud provider — unlike ADF

(Azure) or GoldenGate (Oracle).
 Versatile Data Movement: Capable of both Change Data Capture
(CDC) streaming and bulk historical data loads.
 Flexible Integration: Supports various databases and messaging
systems, allowing broader applicability in hybrid and multi-cloud
environments.
 Flexible Deployment: Striim can be deployed in a self-hosted VM,
enabling it to be used in network-restricted environments with no
ingress.
Cons:

 Self-Hosted Deployment: At 7-Eleven, Striim is deployed on self-

managed VMs, which can add infrastructure overhead. This could be
a pro for those wanting full control or a con due to maintenance
complexity.
 Licensing & Cost: May be more expensive compared to native cloud
options, especially for simpler use cases.
 Learning Curve: Requires a specialized skill set to manage pipelines
and handle error scenarios effectively.

When to Use:

 When both historical and incremental CDC loads are needed from
the same platform.
 When working in a hybrid or multi-cloud setup where cloud-agnostic
solutions are preferred.
 When needing low-latency, real-time streaming across systems.

When Not to Use:

 If the source system is in Azure, and the integration needs are

straightforward — Azure Data Factory (ADF) might offer a more
cost-effective and fully managed alternative.

Confluent Kafka Connect

Description:

Pros:

Cons:

When to Use:

 CDC from a cloud-based database, which could be accessed by

Kafka Connect.
 Streaming / real-time use cases for data

When not to Use:

 On-prem or network-restricted systems

 Historical Loads (though confluent claims this is changing)
 Oracle Golden Gate

Description:

Pros:

Cons:

 Licensing costs can be very expensive for Golden Gate

When to Use:

 Sourcing data from Oracle databases in OCI

When not to Use:

NOTE: Look into Oracle Fusion in the future, also look into Confluent
Kafka’s XTreme processor

MFT

Description:

Pros:

Cons:

When to Use:

 Flat file transfers (especially internal to 7-Eleven)

When not to Use:

 Live, streaming data integration

EDA Recommendation - Data Sharing
Protocols

Owned by Benjamin Sivoravong

Last updated: May 07, 2025

3 min read

15 people viewed Karma Page Builder Request approval

 Overview
 Data Sharing Protocols
o Azure Storage via SFTP
o Azure Storage via SAS URI
o Delta Share (Databricks to Databricks)
o Delta Share (Databricks to non-Databricks)

Overview
This doc outlines the current recommendation from EDP’s Enterprise
Architecture team on which Data Sharing protocols to use, based on the
requirements of any specific project.
These “EA Recommendation” documents are starting-points, each project
should still be reviewed by peers during the Solution Architecture process.

Data Sharing Protocols

Azure Storage via SFTP

Description:

Create an Azure Storage account, with a storage container. Then, enable

access to that storage container with Azure Storage’s SFTP feature
([Link]
transfer-protocol-support).

Pros:

 SFTP is the most ubiquitous data-sharing protocol today. Almost all

vendors and technologies support SFTP.

Cons:

 SFTP credentials need to be closely guarded. Especially if using

user-password logins.
 SFTP is not very fast. For very large datasets, it will be slow and
expensive to copy the entire dataset into a separate SFTP location
for sharing.

Use when:

 The client uploading or downloading the dataset does not support

any other sharing protocols.
 The dataset is relatively small in size (anything in the 10s of GB),
and does not need to refresh frequently (no more than once a day).

Don’t use for:

 Very large datasets, or datasets who refresh more than once a day.

Notes:

 Have a plan to rotate your SFTP credentials regularly, every 90 days

is preferred
 If the SFTP user is not within the 7-Eleven network, you will need to
know their public-facing IP address range, so we can whitelist if in 7-
Eleven and Azure firewalls.
 Azure Storage via SAS URI

Description:

Create an Azure Storage account, with a storage container. Then, create

short-lived SAS URIs for each user accessing that storage container
(setting permissions appropriately). Clients will need to access the storage
container using the Azure SDK and the given SAS URIs.
[Link]
overview

Pros:

 SAS URIs have baked-in expiration dates.

 SAS URIs can be generated based on a user permissions in Entra ID,
allowing different users to access different locations in the storage
account

Cons:

 SAS URIs are essentially passwords. If they are leaked, anyone who
obtains them can access the shared location.
 SAS URIs are specific to Azure Storage. Many data-sharing tools
don’t natively support SAS URI uploads / downloads.

Use when:

 You need user-specific RBAC. Especially if those users are 7-Eleven

employees, and therefore already in our Entra ID directory.
 Your consumer is an engineer, and can use the Azure SDK to access
the location in the SAS URI

Notes:

 Follow all the best practices mentioned in Azure’s docs here:

[Link]
sas-overview

Delta Share (Databricks to Databricks)

Description:

Delta sharing is a protocol based on the Delta Lake table format, which
allows external users to query your Delta Tables without copying the
entire dataset. It’s similar to giving users a direct connection-string to
your data lakehouse, they can make queries directly on your dataset
(assuming they have permissions to).

[Link]

Pros:

 If both parties are using Azure Databricks, Delta Sharing is very

easy. Probably the easiest of all these options.
 If both parties happen to use the same region of Azure, there is no
egress / ingress fee at the network level.
 Delta sharing is a “live” connection, meaning if the shared data is
updated, the consumer will receive the updates immediately, there
is no syncing task to orchestrate.
 Because of the zero-copy nature, Delta sharing can have massive
bandwidth savings on very large datasets, especially if those
datasets are refreshed frequently.

Cons:

 Delta Sharing requires that both parties use the Delta Lake Open
Table Format, which is not really an industry standard yet.
 Delta Sharing requires deploying a dedicated delta-sharing server
(the next option) if one party does not use Databricks.
 Delta Sharing requires that both parties are using Unity Catalog

Delta Share (Databricks to non-Databricks)

Description:

If one of the involved parties is not using Databricks, we can still use Delta
Sharing, but there is a little more setup that has to happen. Namely, the
provider of the data (Databricks) must create credentials for the
consumer, then the consumer must install the Delta Sharing libraries into
their Apache Spark cluster, and use them to access the data

[Link]

[Link]

Pros:

 All the benefits of delta sharing above

Cons:
  More setup than Databricks-to-Databricks, especially by the party
which is not using Databricks.

Use When:

 You need all the benefits of Delta Sharing, but one of the 2 parties
does not use Databricks
 The pros of Delta Sharing outweight the manual engineering effort
of adopting the Delta Table format, and installing Delta Sharing
libraries.
Name Description

MSA-RG-Dev-DigitalDataEG-Contrib Contributor access for De

MSA-DDE-DL-LR-ProdSupport Read access to Productio

DIST-DIGITAL-DATA-ENGINEERING Email Distribution List - T

ACC-ATLASSIAN-JIRA Jira Parent AD Group

ACC-ATLASSIAN-CONFLUENCE Confluence Parent AD Gro

APP-ATLASSIAN-ENTERPRISE-DATA-DEVELOPER Jira/Confluence Access

APP-ATLASSIAN-ENTERPRISE-DATA-SCRUMMASTER Jira/Confluence Access Sc

APP-EDP-CORE-DEV-DEVELOPER Databricks Development

Environment

APP-EDP-CORE-TEST-DEVELOPER Databricks Test Environm

APP-EDP-CORE-UAT-DEVELOPER Databricks UAT Environm

APP-PROD-EDP-CORE-DEVELOPER Databricks Production En

Note: production AD group differs from other
environments

APP-EDP-CORE-DATA-ENGINEER-MLLAB Databricks ML Lab Enviro

APP-EDP-CORE-DATA-SCIENTIST-MLLAB
2024-09-16 - Data Import / Export -
Current State

Owned by Benjamin Sivoravong

Last updated: Sep 17, 2024

5 min read

13 people viewed Karma Page Builder Request approval

Overview
This meeting will serve to answer these questions:

 What are the most common ingestion patterns used in EDP right
now?
 What are the exceptional cases? Places where ingestion is not
following the norms. For each of these, why are the normal pattens
not being used?
o These exceptional cases could be indicators of technical gaps
in our architecture, or they could point out challenging use-
cases we need to support better.
 Are there any specific issues related to our ingestion work today?
Including technical issues, or just process inefficiencies.
 Do we have a list of all ingestion pipelines, and their underlying data
sources?
The outputs of this meeting will include:

 Documentation for all the questions discussed above

 A detailed outline of the current state of ingestion at EDP, including:
o Standard ingestion patterns
o Non-standard ingestion cases
 Justification for each one
o List of all ingestion pipelines, including data-source and owner
for each one
o Any technical issues, or missing features of our current
systems

These will be used to help guide us in the design of a new streamlined

ingestion framework in the future.

Notes:
What are the most common ingest patterns we use right now?

redshift, mongo db, postgres, sql db , oracle , blob storage, aurora db, ftp,
sftp, streaming (kafka, kinesis)

For all of these, we’re just using the built-in databricks connectors for
these services.

Types of ingest:

 Push / Pull
 Batch / Incremental (streaming)
 From internal sources (within 7Eleven, e.g. from 7ep or other
internal teams)
o Streaming
 AWS Kinesis
 Azure Event Hub
 Confluent Kafka
o Direct DB connections (msft sql, oracle, aurora, postgres,
mongodb, redshift, )
 For each db, we’re just use “select *” queries, using
database checkpoints.
 We’re not doing any true CDC on these databases.
 (ram) 7EP is doing a lot of this for us. They have a
CDC tool (strim), that they can configure.
o Files (Batch)
 ftp / sftp
 Azure blob storage
 S3 (earlier we used to have, but we moved out)
  From external sources
o Files (Batch)
 sftp
 Azure blob storage
 S3 (earlier we used to have, but we moved out)

Exceptional Cases:

 MixPanel - We deployed a python application to Kubernetes, to hit

the API and then write to Azure blob storage.
o Why? - there were no connectors between the source and
databricks. So we just use the k8s app to translate from API to
a file.
 MDM - Python app that queries from SQL and then write to blob.
Might be decommissioned.
 Sharepoint - Because there are no specific connectors between
sharepoint and dbks, we also deploy an app

Ingest Pipelines:

 All of our ingest pipelines read directly from a data-source, and write
to bronze with minimal transformation
o Flattening, deduplication (some job, ideally we wouldn’t even
deduplicate at the bronze layer)

Outbound / Data Export:

 Files
o sftp
o Azure Blob
 Databases
o Write to cosmodb
 API (altria, mixpanel)
 Boomi (vendor, integrates with azure blob)
 CShopper - here, we are exporting tons of data to CShopper, but we
want to bring them to the datalake instead, instead of copying the
data.
o (ram) this would be a great use-case for delta-sharing with
external vendors. It saves us from the ETL / reverse ETL
o (ekta) what’s our strategy for actually rearchitecting our
outbound data export? CShopper might be a non-standard use
case. So we don’t really want to use them as the reference.
They don’t want to run their jobs in our infra, to protect their
IP and logic.
Raw / Landing folder:

 Question: What are our standards for when to use “raw” vs when to
go directly to bronze:
o (ram): For push, we use raw, and for pull, we can directly write
to bronze.
o (sundar): SFTP, etc, do need a raw folder to copy to, then the
pipeline
 Question: What’s the interface for app teams to write to the raw
folder:
o Boomi does this
o For app teams, we usually create a separate container for
each app, and them let them use that to write data to.
 This app-specific container is separate from raw. They
write to their container, then we copy to raw, then to
bronze.
 We have a similar setup for outbound data.

Exceptional Export use-cases:

 Fuels: We connect to a NAS server, and upload files. We want ot

move away from this, but we need to have th econversaion wiht the
teams.

Gaps, and things that we need to improve in Ingest:

 Each data-source is implemented fully e2e, so there’s no shared

libraries or functionality.
o Ingest jobs are application-specific
 It would be nice to have a framework for ingest, based on different
data-sources
o Reduce code duplication between all the different ingest jobs
 If we add a new ingest-source, it could affect all of the other
applications:
 Because we don’t have a framework, it is a little hard to track where
things come from.
 Question: Is it better to organize by application, or by type of data?
o My Suggestion (Ben) : create standards around each data-
source type first, then organize your jobs by application, and
enforce those standards.
o (ram): There will be thousands of jobs to orchestrate, but each
job will have its own dependency graph and requirements.
Pipelines will be app-specific, but we can have a framework for
 the code itself. DE team would deliver the pipelines per table,
and ops team would orchestrate the pipelines.
o (Sundar): Lots of things depend on the source. Some teams
provide data from mongo, kinesis, etc. so we do have to
design ingestion based on source.
o (ram): we want to decouple data ingest from data processing.
o (Frank and Ram): Many people use a “raw” or “landing”
folder, which is where the data-sources write to first (in native
format), then there is a pipeline to write to bronze (which does
have a schema)
 From dbks, this is seen as a volume, not a catalog /
schema / table
o (ram) after X time, we also want to move data from raw to
“archive”, which is something like glacier storage
 Make sure that our Ingestion pipelines ONLY write to bronze with no
transformation (or as little processing as possible)
 We need some clear published interfaces for import and
export to/from EDP’s data lake.
o These interfaces can be used by 7EP, by external vendors,
and we can write utilities for different data-types and data-
sources.
 Open Question: Do we want to support direct-connections to RDBMS
systems, like we do now? Or do we want to standardize on CDC, or
get the app teams to push data to a “raw” folder?
o This will involve talking with the upstream teams, to see if
they would be able to do this. This is something to discuss
with Sunil’s team.
o We need to talk to Sunil before making a decision here, Sunil
and Lakshmi N.
 We want to make greater use of delta-sharing for data ingest and
data export with external teams.
o (vinod): who will bear the cost of this if we go on Delta Share
(esp wrt ingress costs). We should do a POC.
o (ram): Delta sharing is slowly becoming a standard for data
exorts across the industry, especially because it works across
all cloud providers.
 Ideally we’d have secondary DBKS accounts in AWS and
GCP, then we can use those to manage data-export with
teams that are in those providers.
 (sundar): we did a poc sharing data from azure dbks to
aws dbks.
 (vinod) same note on outbound files, we want a standard for
encryption, esp if we’re sending data to external vendors.
 (ram) we also want standards for anonymization and
deidentification
o (ekta) Much of our data that goes to CShopper, there are
contracts saying there will be no PII. Standard we use with
teams is PGP
 o (vinod) Ganesh suggested using Azure KeyVault managed
keys
o We should definitely have a standard utility for encryption, but
it stopped being used because not everyone knew it existed.
 (ekta) We also need to make sure there’s no PII within the datalake
(or if it is, it’s encrypted). We want all customer data to be in MDM.
 (veera) We should also discuss standardizing data-compression
techniques.
o We usually gzip, or tar
 (vinod) We also need some kind of standard for data retention, data
lifecycles.
o (ram) The data lifecycle for each application will depend on
the use-case. Enterprise 7 years is really the standard, but hot
and warm layers will depend on how much history app teams
will need regularly.
o (ben) If we built a framework for data lifecycle, it should have
app-specific policies.
2024-10-09 - Federated Workspace
Discussion

Owned by Benjamin Sivoravong

Oct 10, 2024

1 min read

7 people viewed Karma Page Builder Request approval

Attendees:
 Ram
  Ben
 Uma
 Priyanshu

Goal:
Define which applications should be mapped to which domains, so that we
can provision access on a domain-level, and also to help answer any
future questions of routing and RBAC.

Notes:
Domains that have workspaces right now:

 Merch (priority)
 Marketing (priority)
 Fuel 360 (priority)
 Fuel Pricing
 FPNA
 DCL
 EDP / Core / DDE

For many domains, federated nodes will have diamond and platinum
layers. EDP will have the bronze, silver, gold layers.

Domains that exist in UC:

 There are 13 domains in UC, owned by Priyanshu

o sales
o digital
o marketing
o accounting
o inventory
o fuel
o merch
o itops
o operations
o restaurant
o ev
o asset_protection
o supply_chain
  These domains were originally given to Priyanshu from Smitha. They
were created with the organization’s hierarchy chart in mind, each
domain has a VP at its head.'
 We should probably have THESE domains as our federated
workspaces.
o For fuels, there’s some history there between fuel 360 and
fuel pricing, they didn’t want to have a shared workspace. We
need to figure this out.

Current state of RBAC:

 Uma has a list of permissions scripts that are maintained in

Databricks itself, but these are done on an application level, not
assigned to specific domains.
Open [Link]

 Ben is compiling a list of projects that exist in EDP, for the Gitlab
migration. It’s not fully done, but it would be the best source-of-
truth that exists for a master project list
2024-05-21 EDP Platform - Current
Deployment Process


Owned by Benjamin Sivoravong

Last updated: May 22, 2024

3 min read

16 people viewed Karma Page Builder Request approval

👥 Attendees:
 Ben
 Sundar
 Uma
 Ram
 Somesh
 Bharath

📚 Notes:
 We have lots of repositories in Azure Git
 10-15% of process is automated, 80% is manual
 Right now there’s a UC migration happening. They’re trying to
create 60-70 new repositories, but there’s going to be lots of
deployments for that.
o There is a team working on some automation, but they’re
separate from UC migration, so that work is not going to be
ready in time.
o Ideally the UC team would do their own deployments, but we’ll
see.
 DevOps team has experienced a ton of issues with the post-
deployments steps not being super clear, and causing issues in the
deployments. That lead over time, to the creation of this
deployment process.

Deployment Process:

 Every thursday / friday, devs reach out to the DevOps teams to get
approval:
o No remove commands
o No passwords
o Decent code
o Devops checklist, deployment steps
 After teams do the PR, everything is on the responsibility of the
DevOps team
 Project onboarding:
o Devops team creates a one-time devops pipeline for that
project
o This is more of a minor annoyance, it’s not a huge blocker at
all
 o Biggest concern - when we have to do this 70 times for
somethign like the UC catalog. And the fact that the
developers have to depend on Devops to scaffold a new repo
 Post-deployment steps - there’s a template - this is where most of
our time is taken.
a. NOTE: this is live, in a teams call
b. Cherry-pick changes into prod branch
c. Build pipeline and release pipeline triggers - deploys
notebooks
d. Developers verifies that their changes are properly reflected
in the prod environment
e. Data-copy - from dev to prod
i. often times, initial creation of the tables are done in dev,
then instead of re-creating the tables in prod, they just
copy it
ii. we know this is bad, this process is slowly going away,
but not 100%
iii. process involves taking a backup if it’s a new table
(which is automatically cleaned up)
iv. Typically they’d copy the bronze, silver layers. Gold
layer definitely needs approval
v. requires Ganesh approval
f. Developer verify that their changes are properly copied
g. (optional) create a new databricks job
i. One-time, no airflow or anything
1. Give dev jobid, name, parameters. DevOps has a
script for copying a job from dev to the prod
environment. (we just assume the notebooks is in
the same place. sometimes this is a cause of
failure)
2. Currently we have ~2000 jobs in production. So
this takes a lot of time.
ii. Scheduled job
1. In the past, DevOps would Create a databricks
workflow
2. Now, the devs include an astronomer DAG in the
PR, and the Devops team will deploy the DAG to
prod astronomer
a. DAGs let developers do orchestration,
notebook dependencies
b. DAGs let the mindtree support team monitor
all the jobs, they’re all in one place, instead
of being split up by workspaces
c. This was around 1.5 years ago that we got
astronomer
h. Developer verifies their job ran in prod
i. Notebook cell execution
 This deployment process has existed for around 3-4 years
  This entire deployment process can take several hours. So each
dev-ops engineer takes a few deployments and they just juggle
them all. A lot of this process is done async, hopping on and off
calls.
o It’s like a war room, every tuesday and thursday
o These deployments completely take out engineers for
tuesdays and thursdays
 Tredence is currently working on the Bronze, Silver, Gold medallion
architecture for all of EDP. We need to solidify and stabilize our
devops architecture, so that we can support that effort easier.

Feedback and Discussion:

 ideally we would not have DevOps involved in every single commit

and deployment. Ideal process:
o automatic rollbacks in dev
o unit testing
o uat testing
o product-driven signoff for when to push to prod
 There’s a culture-change compeonetn of this, we need applicaiton
teams to be better about peer-review
 For data-copy, if we need to refresh a data. We shoudl just re-create
the entire table from the raw data-sources in prod. Not copying the
tables from dev.
o In the pipeline, you could include a bulk-load script that
executes one time. And maybe devs have access to run these
on-demand in prod
 Step 5 is really bad. Devs cannot have access to real prod data.
 We need to create a new process, and any greenfield projects
should use the new process, and we need to create backlog items
for migrating old jobs to the new process.
 Even for one-time jobs, they should still be Airflow DAGS. Airflow
DAGs can be invoked from a script too, so we just put that in the
script.
a. Deploy notebooks
b. Deploy DAGs
c. (for one time jobs) invoke DAGs within the deployment script
 A key feature of our data infrastructure - liquibase - version control
for data pipeliens and the data itself. Lets you rollback your data
warehouse to a previous, functional state.
Notes
There are so many things being managed by the EDP Platform team right
now. Some are for operations, some are improvements, some are etc. We
need to categorize all the items, to help organize the work, resourcing,
and status reporting.

This will help us understand:

 Do we need additional resources? We can only justify new resources

once we know all the work in progress, and we show that we don’t
have enough people
 Who is in charge of what item of work? This will mostly just help
with communication and making delivery go quicker.

Platform team:

Architecture: Ben, Sundar

Delivery (Onshore): Uma, Somesh, Bharath

Delivery (Offshore): Anjali (GSC), Amit (GSC), Rohan (LTI), Riya (GSC),
Rashmi (LTI), Bharath R. (LTI)

Part-time (Offshore): Arvind (LTI)

Workstreams:

Tasks:

 Administration
o setup (product, env, etc)
o upgrades (Change management)
o maintenance
o access control
o software provisioning
o capacity provisioning
o migrations
o production issues (shared)
o cost optimization
o maintain security posture
o platform readiness
o availability (backups, ha/dr)
o Change Management (C)
o Platform Project Implementation, Execution
  Deployments
o copy-paste notebooks
o promote code to prod pipelines
o security scanning
o CR reviews
 Architecture
o Product evaluations (POCs, POVs)
o Network design
o Data provisioning
o Use-case scoping and planning
o Platform Posture
o DAG Templates
o DAB templates
o data lifecycle management (vision)
o Change Management (C)
o InfoSec Point of Contact
o Cost assessment
o Platform Project Design
 Automation
o Job Scheduler, Orchestration
o Data Pipeline as Code (Notebooks, DABs, DAGs)
o Svc principal automation
o Platform setup with Terraform scripts
o SAS renewals (for blob sas urls)
 Product Management
o Process Engineering
o SLAs
o Intake automation (service now, jira, email)
o Dist, Inbox management
o Change Mangement (R,A)
o Training and Enablement for user teams
o Communication and PR
 Support
o Job/Alerts monitoring 24/7
o External incidents

Side conversation on Incident Management:

 We have a big issue with incident management right now, our

reports are showing a huge number of incidents in production, >100
per week. We need to look at how our incidents are created and
triaged, so that our numbers are actually accurate.
 We need:
o Better definitions of what makes an incident
 o Ability to group incidents together, separate instances of the
same incident
o Better automation for creating incidents?
o Better reporting for incidents, all Platform team members
should see the incidents, platform team should have more
visibility on incidents than management (currently only
management sees the reports, but the platform team doesn’t
at all)
Coremark AR Gap Analysis
Documentation


Owned by Durgesh Yadav

May 26, 2025

6 min read

2 people viewed Karma Page Builder Request approval

📌 Overview
This documentation outlines the Coremark AR Gap, a critical metric used
to identify discrepancies between the payments claimed by our vendor
Coremark and the actual value of inventory received by the stores.
These discrepancies are broken down into Price Variance and Quantity
Variance and are analyzed across different data layers—Bronze, Silver,
and Gold—with dedicated data quality checks at each stage.

🔄 Business Flow

Participants Involved:
 Vendor: Coremark
 Distribution Center (CDC)
 Store
Process Flow:
1. Store places order to Coremark.
2. Coremark fulfills the order directly or via a Distribution Center
(CDC).
3. Discrepancies (AR Gaps) arise between:
o What Coremark claims to have delivered,
o What the CDC confirms was received, and
o What the Store ultimately acknowledges receiving.

Example:

 Coremark claims: $100 delivered

 CDC says: only $90 received
 Store says: only $80 received
 AR Gap = $100 - $80 = $20

⚙️Key Definitions
Term Description

AR Gap Difference between amount invoiced by Coremark and the value of

inventory received at the store

Price Difference due to changes in unit price between order placement and
Variance actual delivery

Quantity Difference due to mismatch in number of units ordered vs. received

Variance

OTP Other Tobacco Products; items where PSA starts with ‘32’

PSR Table Source of Store data (dcl_db.ar_gap_slv_psr)

Coremark Source of Vendor data (dcl_db.ar_gap_slv_coremark_weekly_sac_details)

Table

📈 Price Variance Calculation

➕ Scenario
 Store orders item at $10
  Coremark delivers/invoices item at $11
 Store receives 5 units
 Price Variance = (11 - 10) * 5 = $5

✅ Required Inputs:
 Vendor Cost: From net_charge / shipped_qty in Coremark data
 Store Cost: From (cost_amount - distribution_fee_amount +
excise_tax_amount) in PSR data

🧮 Logic for Non-OTP Price Variance

sql

CopyEdit

WITH coremark_data AS ( SELECT *, net_charge / shipped_qty AS

Coremark_Unit_Price FROM dcl_db.ar_gap_slv_coremark_weekly_sac_details ),
psr_data AS ( SELECT *, (cost_amount - distribution_fee_amount +
excise_tax_amount) AS psr_unit_price, CAST(invoice_number AS INT) AS psr_invoice
FROM dcl_db.ar_gap_slv_psr ), final_data AS ( SELECT DISTINCT *,
ROUND((b.true_CM_unitcost - a.cdc_unit_price) * a.net_received_quantity, 2) AS
calculated_ar_gap FROM psr_data a INNER JOIN coremark_data b ON a.item_id =
b.item_number AND a.store_id = [Link] AND a.psr_invoice = b.po_number WHERE
a.vendor_id = 36827 ) SELECT * FROM final_data WHERE psa <> '32' AND
calculated_ar_gap > 0 AND original_delivery_date >= '2025-01-01' AND shipped_qty
= net_received_quantity AND store_received_quantity = net_received_quantity AND
crv_deposit = 0;

🔁 For OTP Variance: Replace psa <> '32' with psa = '32'.

📦 Quantity Variance Calculation

➕ Scenario
 Store orders 100 units
 Vendor claims delivery of 100 units
 Store confirms receiving only 95
 Quantity Variance = 100 - 95 = 5 units

✅ Required Inputs:
 Coremark Quantity: shipped_qty
 Store Quantity: net_received_quantity
 CDC Price: Derived from PSR as (cost_amount - distribution_fee_amount
+ excise_tax_amount)
🧮 Logic for Quantity Variance (Non-OTP)
sql

CopyEdit

WITH coremark_data AS ( SELECT * FROM

dcl_db.ar_gap_slv_coremark_weekly_sac_details WHERE stnm = 1 ), psr_data AS
( SELECT *, (cost_amount - distribution_fee_amount + excise_tax_amount) AS
cdc_unit_price, CAST(invoice_number AS INT) AS psr_invoice FROM
dcl_db.ar_gap_slv_psr ), final_data AS ( SELECT DISTINCT *, ROUND((unit_cost *
shipped_qty) - (cdc_unit_price * net_received_quantity), 2) AS calculated_ar_gap
FROM psr_data a INNER JOIN coremark_data b ON a.item_id = b.itm_num AND
a.store_id = [Link] AND a.psr_invoice = b.po_number ) SELECT * FROM final_data
WHERE psa != '32' AND ar_gap > 0.99 AND shipped_qty <> net_received_quantity
GROUP BY ALL;

🔁 For OTP Quantity Variance: Replace psa != '32' with psa = '32'.

Data Pipeline Architecture

Layer Description Source Tables Purpose

Bronz Raw ingestion layer PSR & Coremark Tables Raw data load
e

Silver Cleansed & Joins + Calculated Adds business logic

transformed columns

Gold Analytics-ready layer Filtered & Aggregated Used for dashboards &
reports

✅ Data Quality Checks

Each layer includes rigorous DQ checks to ensure consistency and
accuracy:

Bronze Layer Checks:

 Null checks on primary keys (e.g., store_id, item_id)
  Data type validation

Silver Layer Checks:

 Join key validation between PSR and Coremark
 Range validation on cost & quantity fields

Gold Layer Checks:

 Consistency checks between calculated and reported AR gaps
 Variance threshold filters (e.g., calculated_ar_gap > 0)

📊 Output and Usage

 Final outputs from the Gold Layer are consumed by:
o Visualizations & dashboards
o Monthly reporting
o Exception handling for Finance/Operations
 Data is segmented by OTP and Non-OTP to align with
compliance requirements.

📍 Conclusion
The Coremark AR Gap process provides transparency into discrepancies in
inventory transactions. By categorizing gaps into Price and Quantity
Variance and validating data across Bronze, Silver, and Gold layers,
we ensure:

 Accurate financial reporting

 Improved vendor accountability
 Reliable store-level operations tracking

✅ Overview of the Dataset

Column Name Description

RULE_ID Unique identifier for the rule

RULE_NAME Human-readable rule name

RULE_DESC Description of what the rule checks

 RULE_TYPE Type (e.g., CUSTOM)

SCHEMA_NAME Schema where the table resides

TABLE_NAME Target table name

COLUMN_NAME Column(s) involved in rule (can be "ALL_COLUMNS" or "Major

Granularities")

RULE_LOGIC SQL logic applied to enforce the rule

SCAN_TYPE SCAN type: e.g., INCREMENTAL

INCREMENTAL_COL_NA Column used for incremental logic (typically date)

ME

ACTIVE_FLAG Y/N - whether the rule is active

EXTRA_CONFIGS Placeholder for any additional configs

CREATE_DDTM Rule creation timestamp

UPDATE_DDTM Rule update timestamp

RULE_CATEGORY High-level grouping (e.g., COMPLETENESS, UNIQUENESS,

CONSISTENCY)

SEVERITY Severity level (likely 1–5 scale)

DOMAIN Business domain (e.g., COREMARK_PRICE_VARIANCE_NON_OTP

LAYER Data layer (e.g., BRZ, SLV, GLD)

MAIL_RECEIVER Email recipient for rule violations

DQ_TYPE Specific data quality type (e.g., NOT_NULL, DUPLICATE_CHECK,

DATE_FORMAT, etc.)

📌 Insights / Metrics You Can Derive

Here are ideas for analyzing and visualizing this information:

1. Rule Distribution
 Count of rules by RULE_CATEGORY (e.g., COMPLETENESS vs
UNIQUENESS)
 Count of rules by DQ_TYPE
  Rules per TABLE_NAME

2. Layer-wise & Domain-wise Coverage

 Number of rules implemented for BRZ, SLV, GLD
 Rules per domain: COREMARK, COREMARK_PRICE_VARIANCE_NON_OTP,
etc.

3. Column-wise DQ Check Coverage

 How many columns have NOT_NULL or NON_NEGATIVE checks?
 Columns frequently checked for data quality

4. Severity Heatmap
 Number of rules per severity (e.g., Severity 1, 2) for each
TABLE_NAME or RULE_CATEGORY

5. Date-wise Audit
 Rules created or updated over time (using CREATE_DDTM /
UPDATE_DDTM)

📊 Suggested Visualizations (Power BI / Dash / Tableau)

Visualization Idea Chart Type

Rules by RULE_CATEGORY Bar / Pie chart

Rule count by LAYER and Stacked Bar Chart

DOMAIN

Active vs Inactive Rules Donut chart

Rule count trend over time Line chart using CREATE_DDTM

Table of rules with filters Interactive table with search, filter by TABLE_NAME, DQ
etc.

Duplicate Checks with long Collapsible table or code preview box

SQL

Possible Enhancements
1. Add Status Column: Flag rules as Validated, Failed, Pending Fix, etc.
 2. Add Execution Logs: Link each rule to a job run result for full
tracking.
3. Store DQ Result Stats: Record number of records failed per rule.
4. Tag Rules with Owners: Who owns fixing the rule failures?

📌 Overview

This page documents the Data Quality (DQ) rules applied to the Coremark
datasets across various layers such as GLD (Gold), BRZ (Bronze), and
SLV (Silver). These rules are critical for ensuring data completeness,
consistency, and uniqueness in downstream data products and
analytics.

Each rule includes:

 Target Schema and Table

 Column being validated (or combined columns)
 Logic used to check quality
 Rule Type and Severity
 Who receives notifications
 Type of validation like NOT_NULL, NON_NEGATIVE, DUPLICATE_CHECK,
etc.

🔶 Section 1: GLD Layer —

ar_gap_gld_coremark_non_otp_price_variance
Rul Rule Column Rule Logic Summary Rule Seve DQ Typ
e ID Name Type Category rity

1 Store store_id CUST Records where COMPLETE 2 NOT_N

4 ID Not OM store_id IS NULL NESS
6 Null
Check

1 Order ordered_qty CUST Records where COMPLETE 2 NON_N

4 ed OM ordered_qty < 0 NESS VE
7 Qty
Non-
Negat
ive

1 Net net_received_quanti CUST Records where COMPLETE 2 NON_N

ty net_received_quanti
 4 Recei OM ty < 0 NESS VE
8 ved
Qty
Non-
Negat
ive

1 Cost cost_amount CUST Records where COMPLETE 2 NON_N

4 Amou OM cost_amount < 0 NESS VE
9 nt
Non-
Negat
ive

1 Futur original_delivery_dat CUST Records where CONSISTE 2 DATE_

5 e e OM original_delivery_dat NCY AT
0 Date e > current_date()
Check

1 Duplic Combination of CUST Flags duplicate UNIQUENE 2 DUPLIC

5 ate 27+ fields OM rows using full SS CHECK
1 Recor record
d concatenation
Check and HAVING
COUNT(*) > 1

📍Mail Receiver: [Link]@[Link]

Note: All rules are scanned incrementally using today’s load_date.

🔷 Section 2: BRZ Layer —

ar_gap_brz_coremark_weekly_sac_details
Rule Rule Column Rule Logic Rule Category Severi DQ Type
ID Name Type Summary ty

11 Duplicat Major CUSTO Detects UNIQUENESS 2 DUPLICATE

2 e Check Granulariti M duplicate CK
es full-row
records
using string
concatenati
on
 11 CSPON csponm CUSTO Records COMPLETENE 2 NOT_NULL
3 M Null M where SS
Check csponm IS
NULL

🟡 Section 3: SLV Layer —

ar_gap_slv_coremark_weekly_sac_details
Rule Rule Column Rule TypeLogic Rule Category Severi DQ Type
ID Name Summary ty

12 PO po_number CUSTO Records COMPLETENE 2 NOT_NULL

7 Number M where SS
Null po_number
Check IS NULL

12 Duplicat Major CUSTO Identifies UNIQUENESS 2 DUPLICATE

8 e Check Granulariti M duplicate CK
es records
across ~26
fields using
CONCAT +
HAVING
COUNT(*) >
1

📄 Rule Logic Examples (GLD Layer)

🔍 Rule 146: Store ID Not Null Check

sql

CopyEdit

SELECT * FROM dcl_db.ar_gap_gld_coremark_non_otp_price_variance WHERE

store_id IS NULL

🔍 Rule 150: Future Delivery Date Check

sql
CopyEdit

SELECT * FROM dcl_db.ar_gap_gld_coremark_non_otp_price_variance WHERE

original_delivery_date > current_date()

🔍 Rule 151: Duplicate Check (Simplified)

sql

CopyEdit

SELECT * FROM dcl_db.ar_gap_gld_coremark_non_otp_price_variance a WHERE

CONCAT(all columns...) IN ( SELECT CONCAT(all columns...) FROM
dcl_db.ar_gap_gld_coremark_non_otp_price_variance GROUP BY all columns HAVING
COUNT(*) > 1 )

📬 Notification & Ownership

 All alerts/reports triggered from these DQ checks will be emailed
to:
o [Link]@[Link]
 Scan Type: All rules are run incrementally (load_date =
current_date()), ensuring new/updated records are always validated.
 Extra Configs: Currently none, but extensible to include tolerances
or thresholds in future.

📊 Summary
Domai Table Total Categories
n Rules
Domai Table Total Categories
n Rules

GLD ar_gap_gld_coremark_non_otp_price_va 6 Completeness, Consistency

riance Uniqueness

BRZ ar_gap_brz_coremark_weekly_sac_detai 2 Completeness, Uniqueness

ls

SLV ar_gap_slv_coremark_weekly_sac_detai 2 Completeness, Uniqueness

ls
✅ Future Scope
 Add threshold-based rules (e.g., count drop > x%)
 Implement automated ticketing integration (e.g., Jira) for failed
rules
 Create Power BI dashboards for DQ monitoring
Payment Analysis


Owned by Akanksha Kumari

May 22, 2025

3 min read

1 person viewed Karma Page Builder Request approval

1. Vendor Payment Analysis

a. Here we are checking the payment trend of stores to CDC.
Like as per the terms after how many days Stores are paying
to CDC.
b. Once we got the payment from stores SEI is doing payment in
how many days and what is the amount.
c. We are receiving invoices from Vendor (McLane) and in the
particular time period what is the amount we are paying to
vendors.
d. Every week McLane used to send a Account Receivable Gap
(AR Gap) report to SEI.
e. After we got this we are taking that amount and checking with
the vendor analysis so we got to know is there any flaw in that
data.
f. Currently we are doing it for 10 Vendors but our goal is to
make it for all vendor in coming days.
g. Attached screenshot for reference.
Open Vendor Analysis [Link]

2. CAR Details
a. We are doing this for charging back the stores for the
shortage amount
b. We are receiving file from the stores and we are creating
formulation for checking the stores shortage amount whether
it has to send a notice to store for asking the shortage
 payment or we have to leave it as it is falling under the
threshold amount.
c. If the total amount is falling under 30 Dollar we will leave that
case as coming under threshold amount.
d. If the total shortage amount is coming above 30 dollar then
we will send a notice to stores for the shortage amount. Store
has the liberty to challange us within 7 days if their challenge
approves then they do not have to pay any shortage amount
or else they have to send the amount to SEI within a time
frame.
e. We are receiving this shortage cost (CAR details) on a weekly
basis from the stores and we are working with some
formulation for the analysis.

3. Paid vs Unpaid Claims

a. Claim cases is raised by stores.
b. When the stores notice that they has to receive refund
amount from SEI, they used to raise a claim case through 7-
Help incident.
c. Those incident used by us to check the correctness of the data
(format, amount, duplicate).
d. Once it is done we will do the analysis for the claim amount.
e. We are having threshold for above 15 dollar and 15 dollar or
below it.
f. If claim amount is 15 dollar or less than that it will go for auto
approval case and those claims will be settled.
g. But if the amount is coming greater than 15 dollar then we
have to review it.
h. There are different calculation involved for checking the
amount and quantity.
i. Once our review compeleted we will approve the claims.
j. We received the claims files in a weekly manner and
consolidating them into a single file.

4. Oracle Payment History

a. This is used for getting the payment track of the payment
from SEI to Vendors.
b. We are using Oracle EBS and Oracle OFAP for getting payment
details through complex query.
c. Oracle RISP is used to get the item level Payment details
through COmplex query.
d. We are used this Oracle OFAP payment history in DASH for
showing it to end users.
e. Different team uses this payment file for doing their analysis.
 f. We are updating this particular file on a regular basis so daily
analysis work will not hamper.

5. NACHA Payment History

a. Once we received the invoices from Vendor we are going to
send a invoices to the stores on next day.
b. Vendor send this file with item level amount details to SEI.
c. Once we received the amount from the stores as per the
invoices which is done by SEI, we will do the payment to the
vendors.
d. On next file vendors send a updated file with the received
amount confirmation which is falling under NACHA category.
e. We are showing this table into our DASH with the updated
data on a daily basis.
Layered Architecture & DQ
Framework AR GAP


Owned by Durgesh Yadav

Last updated: May 29, 2025

8 min read

5 people viewed Karma Page Builder Request approval

In Data Engineering, we often hear about Bronze, Silver, and Gold

layers as part of a layered architecture—especially in Data Quality
(DQ) and Data Lake systems.

This concept helps organize data in stages based on its quality,

cleanliness, and readiness for business use, similar to how raw
materials go through stages before becoming finished products.

Let’s break this down in simple, non-technical terms, using an

example and detailed explanation for each layer.
🎯 Why Use Layered Architecture in Data Engineering (DQ
context)?
 Clean, Reliable Data: Each layer improves data quality step-by-
step.
 Separation of Concerns: Raw data is handled differently than
cleaned or business-ready data.
 Auditability & Traceability: Easy to trace data back to its original
source.
 Efficiency: You don’t need to clean everything at once—do it in
stages.
 Reusability: Intermediate data (like Silver) can be reused in
multiple reports.
 Error Handling: Catch errors early before reaching the business
reports.

Graphic Diagramme to Understand in High

Level

Open [Link]

🧱 Let's Understand the Layers: Bronze → Silver →

Gold

🥉 1. Bronze Layer (Raw Data)

What is it?
 Raw, uncleaned data directly ingested from source systems (e.g.,
databases, APIs, files).
 No transformation, no filtering—just raw dump of what came in.
Why is it important?
 Keeps original data intact for auditing.
 Helps recover or reprocess in case of downstream errors.
Real-life analogy:
Imagine a grocery warehouse just received vegetables from farms.
They’re dusty, unwashed, and maybe even mixed with rotten items—but
they’re the real, original goods.
Example:
  A file received from an e-commerce system with missing customer
names, null prices, and some corrupt records.
text

Data fields: order_id, customer_name, product_id, price Some rows: 101, John, P001,
100 102, , P002, 103, Sarah, , 90

🥈 2. Silver Layer (Cleaned & Validated Data)

What is it?
 This layer cleans, filters, and formats the data.
 Implements Data Quality rules: removes duplicates, fixes nulls,
standardizes columns.
Why is it important?
 Makes data trustworthy and usable for analytics.
 Helps detect data quality issues like missing values, invalid entries,
etc.
Real-life analogy:
Back to our grocery example: now, vegetables are washed, sorted, and
rotten ones are discarded. They're still not cooked but ready for the
kitchen.
Example:
From the Bronze layer:
 Remove rows with missing customer_name or price.
 Standardize column formats (e.g., make price numeric).
text

Cleaned data: 101, John, P001, 100 103, Sarah, P003, 90

🥇 3. Gold Layer (Business-Ready Data)

What is it?
 Aggregated, enriched, and business-context data ready for
reporting, dashboards, or ML models.
 Combines multiple cleaned sources.
Why is it important?
 Enables decision-makers to get accurate insights.
 Performance-optimized for business use cases.
Real-life analogy:
Now the cleaned vegetables are turned into a delicious meal—ready to be
served to the customers (i.e., business users).
Example:
 Combine customer data with orders and product data.
 Calculate metrics like total sales per customer or region.
text

Customer_Sales_Report: Customer | Total Orders | Total Sales John | 5 | $500 Sarah |

3 | $270

📊 Putting It All Together: A Simple Story

Imagine you're working in a company like Amazon.

1. 🔁 Every second, data flows in from customer purchases, returns,

warehouse updates, etc.
2. 📥 All that goes into the Bronze layer — untouched, raw.
3. 🧹 Then, you apply data cleaning rules in the Silver layer — filter
invalid orders, fix prices, and ensure data is correct.
4. 📈 Finally, the Gold layer summarizes insights like “Top 10 products
sold last week” or “Sales by region” — presented in dashboards for
decision-makers.

✅ Summary Table
Layer Purpose Data Quality Role Example

Bronze Raw Ingestion Keep original data Messy order data from s

Silver Clean & Validate Apply DQ checks (nulls, types) Filtered, formatted order

Gold Ready for BI Aggregated, business logic Sales report for dashboa

Open [Link]
Flow chart to Explain in Basic Terminology

🧠 Why This Helps in Data Quality?

 Each layer lets you focus on a specific aspect of quality (format,
completeness, accuracy).
 Easier to identify and isolate issues (Was the problem in Bronze?
Or in Silver?).
 Avoids rework — Gold layer doesn’t have to worry about cleaning.

Our Framework that we have created for

DQ & Location
So currently we have multiple Tables used for our Dash & powering our
Visualisation. Giving the complete Structure of Tables.

Below are the Examples of the Table that we have created in our Layered
Architecture. If you go to table name section you will see substring of
GLD, BRZ, SLV.

SCHEMA_NA TABLE_NAME LAYE DOMAIN

ME R

dcl_db ar_gap_slv_psr SLV PSR

dcl_db ar_gap_slv_psr SLV PSR

dcl_db ar_gap_slv_psr SLV PSR

dcl_db ar_gap_slv_psr SLV PSR

dcl_db ar_gap_gld_mclane_price_variance_otp GLD PRICE_VARIANCE_OTP

dcl_db ar_gap_gld_mclane_price_variance_otp GLD PRICE_VARIANCE_OTP

dcl_db ar_gap_gld_mclane_price_variance_otp GLD PRICE_VARIANCE_OTP

dcl_db ar_gap_gld_mclane_price_variance_otp GLD PRICE_VARIANCE_OTP

dcl_db ar_gap_gld_mclane_price_variance_otp GLD PRICE_VARIANCE_NON

dcl_db ar_gap_brz_mcl_cdc_weekly_otp BRZ MCLANE_OTP

dcl_db ar_gap_brz_mcl_cdc_weekly_non_otp BRZ MCLANE_NON_OTP

dcl_db ar_gap_brz_mcl_cdc_weekly_non_otp BRZ MCLANE_NON_OTP

dcl_db ar_gap_brz_mcl_cdc_weekly_otp BRZ MCLANE_OTP

dcl_db ar_gap_brz_mcl_cdc_weekly_otp BRZ MCLANE_OTP

dcl_db ar_gap_brz_mcl_cdc_weekly_non_otp BRZ MCLANE_NON_OTP

dcl_db ar_gap_brz_mcl_cdc_weekly_otp BRZ MCLANE_OTP

dcl_db ar_gap_gld_coremark_non_otp_price_v GLD COREMARK_PRICE_VA

ariance _OTP

dcl_db ar_gap_gld_coremark_non_otp_price_v GLD COREMARK_PRICE_VA

ariance _OTP

dcl_db ar_gap_gld_coremark_non_otp_price_v GLD COREMARK_PRICE_VA

ariance _OTP

dcl_db ar_gap_gld_coremark_non_otp_price_v GLD COREMARK_PRICE_VA

ariance _OTP

dcl_db ar_gap_gld_coremark_non_otp_price_v GLD COREMARK_PRICE_VA

ariance _OTP

dcl_db ar_gap_gld_coremark_non_otp_price_v GLD COREMARK_PRICE_VA

ariance _OTP

dcl_db ar_gap_gld_mclane_price_variance_inp GLD MCLANE_PRICE_VARIA

ut_file FILE

dcl_db ar_gap_gld_mclane_price_variance_inp GLD MCLANE_PRICE_VARIA

ut_file FILE

dcl_db ar_gap_gld_mclane_price_variance_inp GLD MCLANE_PRICE_VARIA

ut_file FILE

dcl_db ar_gap_gld_mclane_price_variance_inp GLD MCLANE_PRICE_VARIA

ut_file FILE

dcl_db ar_gap_gld_mclane_price_variance_inp GLD MCLANE_PRICE_VARIA

ut_file FILE

dcl_db ar_gap_gld_mclane_price_variance_inp GLD MCLANE_PRICE_VARIA

ut_file FILE
How these tables are designed in Layers ?

Additional to these Layers every table DDL structure is added with three
new columns-

Unique_ID , Load_Time , Load_Timestamp

What is the Reason of Adding Unique_ID in All the Layered Tables ?

So we have Included Unique_ID to identify the Rows that will fail in our DQ
Check as we don't have a Primary Key in our Table so we tried to create
Unique_ID in such a way that every time new data gets loaded into the
Table it gets a distinct Unique_ID.

This Unique_ID is a auto incremented Integer numbers that get assigned

to each row in all the Layers.

NOTE - Unique ID is mandatory on all the Tables that are part of DQ

Check.

A sample DDL Structure of a Table with Unique_ID

[Link]("""

CREATE TABLE dcl_db.ar_gap_brz_coremark_weekly_sac_details (

unique_id BIGINT GENERATED ALWAYS AS IDENTITY,

brnm STRING,

strnum STRING,

cusnum STRING,

Seitnr STRING,

seitds STRING,

itmnum STRING,

cnsinv STRING,

csponm STRING,

trnm STRING,

stnm STRING,
 invdte STRING,

orddte STRING,

ciwdyn STRING,

seorup STRING,

qtyord STRING,

qtyshp STRING,

unitcost STRING,

crvdep STRING,

ordcost STRING,

crunits STRING,

`credit$$` STRING,

salestax STRING,

netchg STRING,

StreRcvQty STRING,

PaidAmt STRING,

Argap STRING,

load_date DATE,

load_timestamp TIMESTAMP

USING DELTA

PARTITIONED BY (invdte)

LOCATION
'dbfs:/user/hive/warehouse/dcl_db.db/brz/ar_gap_brz_coremark_weekly_sac_det
ails'

""")

Why load_date & load_timestamp is added ?

Load_Date & Load_Timestamp - Load_Date is added to put the Parameter

in DQ Check that Every time when the rule runs “load_date =
current_data” . How this parameter works ? So whenever our Automation
works a current_date as load_date and current_time as load_timestamp is
added to the Table and now this help us in our DQ Automation as we run
our DQ Check on same date when the data gets loaded.

Three Major Tables responsible for Storing DQ Checks, Data & Affected
Rows ?
Table 1 - dcl_db.ar_gap_dq_rule_config

Columns in this DQ Table & their meaning -

Column Value in Query Description

Name

rule_id 2 Unique identifier for the Data Quality (DQ

rule_name 'store column length Name of the rule. It ensures that the stor
check' column has exactly 5 digits.

rule_descripti 'store column should be of Detailed explanation of the rule. The stor
on length 5 digits' column must always have exactly 5
characters.

rule_threshol 'CUSTOM' Threshold type for validation. 'CUSTOM' m

d the validation rule is not a standard chec
NULL or UNIQUENESS) but a user-defined

database_na 'dcl_db' The database where the rule is applied.

me

table_name 'ar_gap_slv_ The table in which the rule applies.

mcl_cdc_weekly_otp'

column_nam 'store' The column being validated (store).

e

rule_logic 'select * from dcl_db. Custom Rule Logic: This rule checks that
is exactly 5 characters long. - len(store) >
Fails if the column has more than 5 digits
len(store) < 5: Fails if the column has few
ar_gap_slv_mcl_ than 5 digits. - load_date = current_date(
Ensures the rule applies only to the lates
load.
cdc_weekly_otp where
 (len(store)>5 or
len(store)<5) and
load_date =
current_date()'

scan_type 'INCREMENTAL' Defines how often the scan should run: -

INCREMENTAL: Checks only new data for
current date. - FULL: Scans the entire dat
each time.

filter_conditio 'load_date = Condition that restricts the check to only

n current_date()' today's data (current_date()).

active_flag 'Y' Indicates if the rule is active (Y) or inactiv

comments '' Space for additional comments (empty in

case).

created_at current_timestamp() Timestamp indicating when the rule was

inserted into the table.

updated_at current_timestamp() Timestamp for when the rule was last

modified.

rule_category 'CUSTOM' Type of Data Quality rule: - UNIQUENESS

Checks for unique values. - NULL CHECK:
Ensures values are not null. - CUSTOM: U
defined validation logic.

rule_priority '2' The priority level of the rule: - 1: High - 2

Medium - 3: Low This determines how urg
a failed rule should be investigated.

domain_nam 'FPNA' The business domain under which this ru

e (FPNA → Financial Planning & Analysis).

layer_name 'SLV' The data processing layer where the rule

enforced: - RAW: Initial ingestion - SLV:
Standard Layer View - CURATED: Final cle
data used for reporting.

receiver_ema '[Link]@7- The email of the person responsible for

il [Link]' monitoring rule failures.
Table 2 - dcl_db.ar_gap_dq_rules

Field Name Data Meaning / Description

Type

result_id bigint Unique identifier for each result record

rule_id int ID of the data validation rule applied

rule_desc string Description of the rule used to validate the data

run_id string Identifier for the data quality run or batch

table_name string Name of the table being evaluated

column_name string Name of the specific column checked within the ta

total_records_count bigint Total number of records in the dataset or table

good_records_count bigint Number of records that passed the validation rule

error_records_count bigint Number of records that failed the validation rule

good_records_percent double Percentage of records that passed validation

age

error_records_percent double Percentage of records that failed validation

age

table_health string Overall health status of the table (e.g., Good, Aver
Poor) based on rule checks

result string Outcome of the rule check (e.g., Passed, Failed)

load_date date Date when the data was loaded or validated

load_timestamp timesta Exact timestamp of when the data was loaded or

mp validation executed

domain string Business domain to which the data belongs (e.g.,

Finance, Insurance, Retail)

Table 3 - dcl_db.ar_gap_dq_affected_rows

Field Name Data Meaning / Description

Type

result_id bigint Unique identifier linking to the result of a data validation or

 quality check

affected_row_ bigint Identifier of the specific row in the dataset that was affecte
id flagged as erroneous

load_date date Date when the data was loaded into the system or when th
check was performed

load_timesta timesta Exact timestamp indicating when the data was loaded or
mp mp validation check was executed

These three Tables when merged and joined together are used for
the DQ Summary Dashboard.

TIMELINESS CHECK :

Included the timeliness check logic in DQ framework , where we can

ingest the rules in the dq config table to check whether particular table is
loaded as per it’s schedule or not. If it’s not loaded as per schedule, then
since how many days data is not loaded in the table , that information will
be loaded to dq results table.

Team can ingest the rules in dq table as below for timeliness check

-- insert into dcl_db.ar_gap_dq_rule_config

-- values

-- (24,'timeliness check for maclane otp table','maclane otp table should be updated
every
tuesday','TIMELINESS','dcl_db','ar_gap_slv_mcl_cdc_weekly_otp','','','WEEKLY','','Y','TUESD
AY',current_timestamp(),current_timestamp(),'TIMELINESS','2','MACLANE','SLV','pranilpra
[Link]@[Link]','TIMELINESS')

RULE_TYPE - should be always ‘TIMELINESS’ for timeliness check

SCAN_TYPE - should the frequency of the table load (DAILY,WEEKLY)

EXTRA_CONFIGS - if the table load frequency is WEEKLY, then mention on

which day of week table is getting loaded in EXTRA_CONFIGS column
(Monday,Tuesday etc)

DBKS_AR_GAP_DQ_CHECKS - scheduled this job to check timeliness check

on daily basis at 6 PM IST. Team can add their rule id for timeliness check
in ids parameter in the job (comma seperated values) as below and save
the task.

Open Screenshot 2025-05-21 at 7.51.08 [Link]

Once this job runs successfully, then timeliness check data will be loaded
to ar_gap_dq_results table as below.

Open Screenshot 2025-05-21 at 7.53.29 [Link]

Here, result column explains since how many days data is not loaded. if
it’s 0 then data is loaded as per it’s schedule and timeliness check is
passed. if it’s 1,2,3 etc that means data is not loaded in the table since
1,2,3 etc. days or there is delay in data load.

Enhancement & UI Review –

Dashboard Components


Owned by Akanksha Kumari

May 20, 2025

1 min read

3 people viewed Karma Page Builder Request approval

1. CDC Detailed Check-In

· Graph Enhancement: Improve the visual representation of the Daily

Credit graph to enhance readability and insights.

· Collapse Functionality: Verify that collapse/expand functionality is

working correctly.

· Navigation Link: Include a "View Details" link for users to navigate to the
full CDC Detailed Check-In view.

2. Price Variance Charges to Stores: McLane and Core-Mark

· Graph Adjustment: Remove the Total line/series from the graph for
clearer comparison and visualization.
· Collapse Functionality: Validate collapse/expand behaviour.

· Navigation Link: Add a "View Details" link to provide access to the

complete breakdown of store-level charges.

3. AR GAP – Lump Sum Payments

· Table Enhancement: Add a Total Row at the top of the data table for
immediate visibility.

· Collapse Functionality: Check the collapse/expand interaction for

usability.

· Navigation Link: Provide a "View Details" link to access in-depth payment

information.

4. AR GAP – McLane Reported vs SEI Expected Payment

· Table Enhancement: Insert a Total Row at the top of the table for quick
access to aggregate figures.

· Collapse Functionality: Confirm that the collapsible panel works as

intended.

· Navigation Link: Ensure a working "View Details" link is in place for

deeper exploration.

5. Money Recovery Program and SCIS

· Collapse Functionality: Check that the module’s collapse/expand feature

functions properly.

· Navigation Link: Include a "View Details" link redirecting to the Money

Recovery Program page for full content access.
Skipcart Business Usecase

Owned by Veera Vissa

Last updated: Apr 25, 2025

1 min read

5 people viewed Karma Page Builder Request approval

Skipcart is a last-mile delivery platform that partners with 7-Eleven to

power fast, reliable, and scalable on-demand deliveries.

Business Justification:

Here is the business case to bring the live streaming data:

a) Real-Time Monitoring of Deliveries: Enabling operations teams to

monitor deliveries in progress and track potential delays or route
deviations.

Order at risk - based on location of drivers, and timing of deliveries,

detecting when it would be impossible to fulfill a delivery within the
desired time-window. Skipcart needs to have relatively real-time driver
locations and order information to be able to determine this (within 30
seconds).

b) Predictive Alerts and Notifications: Using event-based triggers

(e.g., unexpected driver delays or merchant delays) to send alerts to
partners/customers to take proactive steps.
c) Data-Driven Decision-Making: Analyzing real-time trends in peak
order times, demand hotspots, or high-cancellation zones to optimize
resource allocation.

Data Lake Jobs and Data Dictionary Details

Environment : DEV

 Data Lake Load Job details

o DBKS_Skipcart_INGESTION_Streaming_Priority_Bronze -
Databricks
o DBKS_Skipcart_INGESTION_Streaming_Bronze - Databricks
o DBKS_SKIPCART_INGESTION_STREAMING_PRIORITY_SILVER -
Databricks
o DBKS_SKIPCART_INGESTION_STREAMING_SILVER - Databricks

Attached are the 12 Bronze and Silver Data Dictionaries for Bronze and
Silver Streaming tables.

 Bronze
o Catalog Name : edp_brz_dev
o Schema Name : skipcart_sqlserver_dbo
o
 Silver
o Catalog Name : edp_slv_dev
o Schema Name : digital
Skipcart - Upsert Logic in Azure Data
Factory


Owned by Anandhan M R

Last updated: Jun 02, 2025

3 min read

2 people viewed Karma Page Builder Request approval

•Pipeline Setup: Parameters are used to dynamically initialize table

names.

•Lookup Activity: Connects to the source SQL Server and runs a query
to retrieve tables from the parameters.

•For Each Activity: After successful Lookup, each table is processed in a

Data Flow activity.

Open [Link]

LookUp Activity:

Connects to the source to retrieve the primary key for each table being
iterated over from the previous Lookup activity. The query uses dynamic
code to fetch the primary key for each table.

Open [Link]

Checkpoint Declaration:

•Checkpoint Key: In the dataflow activity a checkpoint key is initialized

(or an auto-generated key can be used).

•Data Processing: The Data Flow activity is configured to process data

from the respective tables based on the defined operations.

•Dynamic Parameters: Parameters are initialized to dynamically fetch

the table and schema names from the source server, ensuring
adaptability based on the tables and schemas being processed.

Open [Link]

Source Activity:
In the Data Flow activity, the Source uses these key components:

•Source Settings: A dataset is created with the linked service to connect

the pipeline to the source database.

•Source Options: A dynamic query is used to insert audit columns into

the Delta Lake tables for consistency and traceability.

•Incremental Column: The checkbox is selected to enable upsert logic,

allowing updates to existing records or insertion of new ones.

•Upsert Setup: The column for upsert logic is chosen, with the option
"Full on the first run, then incremental" to load all data initially and apply
incremental updates thereafter. Alternatively, only incremental changes
can be pulled from the start.

Open [Link]

Open [Link]

If a full load needs to be performed for the tables, the checkpoint key can
be overridden or manually changed using the pipeline expression builder.
Once the checkpoint key is adjusted, the pipeline should be published and
triggered to initiate the full load process for the respective tables. This
allows for a fresh load of data, bypassing the checkpoint mechanism, and
ensuring that the entire dataset is loaded as needed.

Open [Link]

Partition Key Declaration:

•Partitioning: Select “Set Partitioning” to define the partition key for

Delta tables.

•Partition Type: Choose "Key" where each distinct column value

becomes a partition (ideal for a smaller number of distinct values).

•Alternative Options: Other partitioning types can be selected based on

the use case.

Open [Link]
Alter Row Activity:

•Alter Row Activity: Conditions are initialized based on the

requirements.

•Upsert Logic: The Upsert If condition is selected, with the expression

true() in the expression builder, which considers all incoming records and
checks for the upsert condition.

•Other Options: Alternatives like Insert If, Update If, and Delete If
can be used depending on different scenarios.

Open [Link]

Open [Link]

Sink Activity:

•Sink Activity: Under the Settings tab, the folder path for the Delta
table data is specified.

•Update Method: Set to Allow Insert and Allow Upsert to handle data
from the incoming stream.

•Key Column: Custom Expression is selected for the primary key, with
an expression provided to match the primary key accordingly.
Alternatively, the primary key can be selected manually from the list of
columns. Select “Set Partitioning” to define the partition key for Delta
tables under the optimize tab.

Open [Link]
Skipcart - One Time Load and Column
Refinement

Owned by Anandhan M R

Last updated: Jun 02, 2025

3 min read

3 people viewed Karma Page Builder Request approval

Pipeline Setup: Parameters are used to dynamically initialize table

names.

•Lookup Activity: Connects to the source SQL Server and runs a query
to retrieve tables from the parameters.

•For Each Activity: After successful Lookup, each table is processed in a

Data Flow activity.

Open [Link]

LookUp Activity:

Connects to the source to retrieve the primary key for each table being
iterated over from the previous Lookup activity. The query uses dynamic
code to fetch the primary key for each table.
Open [Link]

Source Activity:

In the Data Flow activity, the Source uses these key components:

•Source Settings: A dataset is created with the linked service to connect

the pipeline to the source database.

•Source Options: A dynamic query is used to insert audit columns into

the Delta Lake tables for consistency and traceability.

Open [Link]

Partition Key Declaration:

•Partitioning: Select “Set Partitioning” to define the partition key for

Delta tables.

•Partition Type: Choose "Key" where each distinct column value

becomes a partition (ideal for a smaller number of distinct values).

•Alternative Options: Other partitioning types can be selected based on

the use case.

Open [Link]

If there is no partition is needed or it is not available as per the usecase

“Use current partitioning” option can be used to load the data without any
partition into the delta lake.

Sink Activity:

•Sink Activity: Under the Settings tab, the folder path for the Delta
table data is specified.

•Update Method: Set to Allow Insert to handle data from the incoming
stream.

•Optimize: Select “Set Partitioning” to define the partition key for

Delta tables under the optimize tab. If there is no partition is needed or it
is not available as per the usecase “Use current partitioning” option can
be used to load the data without any partition into the delta lake.

Open [Link]

Refinement Pipeline:

 Pipeline Structure Consistency: The pipeline follows the same

structure as the other existing pipelines.
 Use of Select Activity: A Select activity is incorporated to map
columns accurately.
 Column Standardization: It processes certain columns to align
with data standards.
 Data Transformation: Modifies columns from the source to target
(e.g., removing special characters or extra spaces).
 Improved Data Quality: Ensures clean and standardized data is
inserted into the data lake.

Open [Link]

In the Skipcart use case, the pipeline leverages the expression true() to
check all incoming columns from the source. This ensures that any extra
spaces in the columns are trimmed using the expression replace($$, ' ', ''),
which removes spaces from the data. Additionally, the use of expressions
is highly customizable; depending on the specific use case and the
characteristics of the incoming columns, other expressions can be applied
to perform necessary transformations or modifications to the data. This
flexibility ensures that data is processed according to the required
standards before being loaded into the target system.

Once the Select activity is completed and the necessary column

modifications are applied, the transformed data is loaded into the Delta
Lake using the Sink activity. This ensures that the cleaned and
standardized data is written to the target storage in the desired format.
ADF

Owned by Veera Vissa

Jun 03, 2025

3 people viewed Karma Page Builder Request approval

Pros:
 Can connect to all major sources - ADF-Connectors.
 Bulk Data Copy from SQL Server to ADLS Gen2.
 Built-in retry, failure handling, and alerting mechanisms.
 Simplifies pipeline creation with a drag-and-drop visual
environment.

Cons:

 Can’t Encrypt PII data on the fly. Need to call Databricks notebook
for encryption
 Limited Transformation Logic
 Optimizations, schema enforcement, versioning, and rollback
features of Delta Lake needDatabricks.
 Less flexible for advanced conditional or event-driven workflows
compared to tools likeAirflow.

Digital Data Engineering

/

Unity Catalog -Airflow Step by Step

 

Share

Copy link

More actions

Unity Catalog -Airflow Step by Step

Owned by Praveen Kumar (Deactivated)

Last updated: Aug 22, 2024

2 min read
47 people viewed Karma Page Builder Request approval

Airflow

Contributor Praveen Kumar

Problem Statement Step by step instruction for “create new DAG in Airflow”

Apps Fuel360

What is Airflow?

Apache Airflow is an open-source platform for developing, scheduling, and

monitoring batch-oriented workflows.

Step-by-step guide for creating, scheduling, and monitoring a

workflow in Apache Airflow

[Link] Ivanti Secure Access Client (VPN):

Install the Ivanti Secure Access client on local machine to establish a VPN
connection.

[Link] Databricks Server to Docker/Sandbox (Container):

Establish a connection between Databricks server (RDM folder) and

Docker/Sandbox container. This ensures that Airflow can interact with
Databricks for workflow execution.

[Link] Airflow Workflow:

 Plan and design workflow, identifying the stages, tasks, and

dependencies.
 Decide on the naming convention for tasks and DAGs to maintain
consistency.
 Determine the parameters and configurations required for each task
in the workflow.
  Create or gather any necessary scripts, notebooks, or code that will
be executed as part of the workflow tasks.

4. Create Python File:

 Open a text editor or integrated development environment (IDE) to

create Python file.
 Import all Python libraries or modules that are required for workflow.
 Set up required environment variables or configurations required for
workflow to run successfully.
 Define DAG using the DAG class provided by Airflow, specifying
parameters such as DAG id, schedule interval, default args, etc.
 Define each task within the DAG using the appropriate Airflow
operators (e.g. Bash Operator, Python Operator etc.).
 Specify the dependencies between tasks using the set upstream
and set downstream methods or the >> and << operators.

5. Define DAG Structure:

 In Python file, create the DAG structure, including Bronze, Silver and
Gold stages.
 Ensure that task dependencies are correctly defined so that Silver
triggers after Bronze completes, and Gold triggers after Silver
completes.

6. Obtain Job IDs from Databricks Workflow UI:

 Log in to the Databricks Workflow UI.

 Navigate to Jobs and locate the specific job details for each stage of
workflow (Bronze, Silver, Gold).
 Note down the Job IDs, as they will be used in Airflow for task
execution.

7. Upload JSON Files as Variables in Airflow:

 Prepare JSON files containing task and dependency configurations.

 Log in to the Airflow UI and navigate to Admin > Variables.
 Use the "Import" feature to upload the JSON files as variables into
Airflow.

8. Validate JSON Files:

Before uploading JSON files to Airflow, validate them using a JSON online
validator and formatter (e.g., JSON Lint) to ensure they are correctly
formatted and free of errors.

[Link] and Schedule the Workflow:

  Once DAG and variables are set up in Airflow, you can trigger the
execution of the workflow.
 Run the DAG manually from the Airflow UI to verify that it executes
as expected.
 Set up scheduling parameters to define when and how often the
workflow should run automatically.
 Monitor the execution of the workflow from the Airflow UI, checking
task statuses, logs, and any error messages.
 Adjust the DAG or task configurations as needed based on the
results of the execution and any feedback received.

Introduction
What is Unity Catalog?

Unity Catalog provides centralized access control, auditing, lineage, and

data discovery capabilities across Databricks workspaces. Key features of
Unity Catalog include: Define once, secure everywhere: Unity Catalog
offers a single place to administer data access policies that apply across
all workspaces and personas

 Provides a unified platform for administering data access policies

across workspaces and user personas.
 Offers a standards-compliant security model based on ANSI SQL,
allowing familiar syntax for granting permissions.
 Automatically captures detailed audit logs and lineage data to track
data access and usage.
 Facilitates data discovery through tagging, documentation, and an
intuitive search interface for easy access to relevant data assets.

Unity Catalog is a powerful tool for managing and securing data within
Databricks workspaces, providing centralized access control, auditing,
lineage tracking, and data discovery functionalities.

Unity Catalog works with your existing data catalogs, data storage
systems and governance solutions so you can leverage your existing
investments and build a future-proof governance model without expensive
migration costs.

Purpose and Benefits

Unity Catalog helps teams efficiently manage metadata, improve data

discovery, and enhance data governance. It enables collaboration and
ensures that data assets are well-documented and secure.

As shown in picture above Unity Enabled cluster will have mark

on Compute
Note : While migrating RDM or any other apps/project following point need
to be taken care

1 : All write/read operation will be on abfss/adls location (“abfss://data-

warehouse@*******************.[Link]/**********/*******“)

Mount Point not allowed to dump data ( read or write ) .

Lineage will not maintained while reading or writing data to/from mount
Point

2 : While reading or writing data table , follow 3 tier architecture as given

below example

Example :

%sql

select * from [Link].batch_master_data

where Catalog Name = test

Schema Name = digital

Table_name = batch_master_data

3 : Use Volume for writing logs only ( In next point mentioned how to
create volume and its drawbacks)

Here is an example of what not to do:

log_base_path='/dbfs/mnt/BATCH_RESULTS/FIREBOLT/
LM_Loyalty_Daily_Batch_Logs/'

Instead of mount point , you should use volume For example:

log_base_path='/Volumes/test/digital/uc_vol_test/'

4 : Here is an example of what not to do while creating table

[Link]("CREATE EXTERNAL TABLE my_table USING DELTA LOCATION

'dbfs:/mnt/my-data/my-table'")

Instead of DBFS, you should use direct paths to your external storage
locations, such as S3 or ADLS Gen 2 URIs. For example:
[Link]("CREATE EXTERNAL TABLE my_table USING DELTA LOCATION
'abfss://data-warehouse@[Link]/SPARK_TABLES/
test/batch_master_data'")

Use DBFS while launching Unity Catalog

clusters with Single User access mode
Databricks recommends using DBFS mounts for init scripts,
configurations, and libraries stored in external storage. This behavior is
not supported in shared access mode.

Do not use DBFS with Unity Catalog external

locations
Unity Catalog secures access to data in external locations by using full
cloud URI paths to identify grants on managed object storage directories.
DBFS mounts use an entirely different data access model that bypasses
Unity Catalog entirely. Databricks recommends that you do not reuse
cloud object storage volumes between DBFS mounts and UC external
volumes.

3. Volumes on Unity Catalog

Access, store, govern, organize and process non-tabular data *

 Volume types: managed or external

 Files organized in a 3-level namespace centrally in Unity Catalog
<catalog>.<schema>.<volume_name>
 Hadoop distributed file system implementation and FUSE support
 Paths
Hadoop
[dbfs:]/Volumes/<catalog>/<schema>/<volume>/<path>/<file_na
me>
FUSE
/Volumes/<catalog>/<schema>/<volume>/<path>/<file_name>
 Governance model based on ANSI SQL GRANT and REVOKE
commands - volume-level
 User interface for browsing and managing files in the Data Explorer
and Notebooks
 Content accessible via APIs (Spark, dbutils, REST, SQL, local files
system) and the Databricks CLI
Migration from Dremio to Databricks:


Owned by Shailav Sinha, created with a template

Last updated: Oct 15, 2024
3 min read

10 people viewed Request approval

Overview
As discussed with DataLake team, we have been notified that Dremio
license will be decommissioned by end of Nov 24 (Datalake team yet to
come with an ETA). As part of this plan most of the Merch Tech Power BI
reports are consuming data from Dremio and we wanted to be ready with
data source change from Dremio to Databricks.

Problem Statement: Migration from Dremio to Databricks

The objective is to successfully migrate our existing data analytics and

data processing workflows from Dremio to Databricks.

Proposed Solution
Background:

Our organization currently uses Dremio for data virtualization, query

acceleration, and analytics. Dremio license will expire in Nov’24. DataLake
team will not be renewing the license for next year. As our data needs
evolve, we have identified Databricks as a more suitable platform due to
its robust support for Apache Spark, collaborative features, and scalable
data lakehouse architecture. The migration is necessitated by the need to
enhance performance, scalability, and integration with our machine
learning and data engineering workflows.

Implementation Plan
Migrating a data source from Dremio to Databricks involves several steps,
including data transfer, schema mapping, and adjusting queries. Here’s a
high-level overview of how you can approach this migration:

1. Understand the Existing Environment

 Data Sources: Identify and document the data sources you are
currently using with Dremio.
 Schema: Review the schema definitions and data types in Dremio.
 Queries: Collect any existing queries and transformations being
used in Dremio.

2. Prepare Databricks Environment

 Set Up Databricks Workspace: Ensure you have a Databricks
workspace set up with the necessary clusters and configurations.
 Data Storage: Decide where you will store the data in Databricks.
You can use Databricks’ built-in Delta Lake or another storage
option compatible with Databricks.

3. Transfer Data

A. Export Data from Dremio

 Export Methods: Use Dremio’s tools to export data. You can

export data to formats like CSV, Parquet, or ORC, which are
commonly supported in Databricks.

B. Import Data into Databricks

 Upload Data: Use Databricks’ file upload tools or Databricks CLI to

upload the exported files to DBFS (Databricks File System) or your
cloud storage (e.g., AWS S3, Azure Blob Storage).
 Create Tables: Use Databricks SQL or PySpark to create tables
from the uploaded files. You might use commands like CREATE TABLE
or CREATE OR REPLACE TEMPORARY VIEW to define your data in
Databricks.

4. Migrate Queries and Transformations

 Rewrite Queries: Dremio uses its SQL dialect which might differ
slightly from Databricks SQL. Review and adapt your queries to
match Databricks SQL syntax.
 Adjust Transformations: If you have complex transformations or
data pipelines, you might need to rewrite these using Databricks’
Spark SQL or PySpark.

5. Validate Data
 Data Integrity: Ensure that the data in Databricks matches the
data from Dremio. Perform data validation checks and compare
sample results.
  Performance Testing: Test the performance of queries and
transformations in Databricks to ensure they meet your
requirements.

6. Update Applications and Dashboards

 Reconfigure Connections: Update any applications, dashboards,
or BI tools that were connected to Dremio to now connect to
Databricks.
 Testing: Test the connectivity and functionality of these
applications and dashboards to ensure everything works as
expected.

7. Monitor and Optimize

 Monitor Performance: Keep an eye on the performance of your
queries and data processes in Databricks. Use Databricks’
monitoring tools to track performance.
 Optimize: Based on the performance and feedback, optimize your
queries and data storage configurations.

Describe what an outer join in sql and give me example

Assistant
An outer join in SQL is a type of join that returns all rows from one table and the matched
rows from the other table. If there is no match, the result is NULL on the side of the table
that does not have a match. There are three types of outer joins: LEFT OUTER JOIN,
RIGHT OUTER JOIN, and FULL OUTER JOIN.

 LEFT OUTER JOIN: Returns all rows from the left table, and the matched rows
from the right table. If there is no match, NULL values are returned for
columns from the right table.
 RIGHT OUTER JOIN: Returns all rows from the right table, and the matched
rows from the left table. If there is no match, NULL values are returned for
columns from the left table.
 FULL OUTER JOIN: Returns all rows when there is a match in one of the
tables. If there is no match, NULL values are returned for columns from both
tables.

Here is an example of each type of outer join:

LEFT OUTER JOIN

SELECT
[Link],
[Link],
departments.department_name
FROM
employees
LEFT OUTER JOIN
departments
ON
 employees.department_id =
[Link];

RIGHT OUTER JOIN

SELECT
[Link],
[Link],
departments.department_name
FROM
employees
RIGHT OUTER JOIN
departments
ON
employees.department_id =
[Link];

FULL OUTER JOIN

SELECT
[Link],
[Link],
departments.department_name
FROM
employees
FULL OUTER JOIN
departments
ON
employees.department_id =
[Link];
These queries will return all employees and their corresponding department names, with
NULL values where there is no match between employees and departments.
s
spra3049@[Link]
What is more likely to change, fact or dimension?
Assistant
In a data warehouse, dimensions are more likely to change than facts.

 Dimensions: These are descriptive attributes related to the facts, such as

customer information, product details, or geographical data. Dimensions can
change over time as new products are introduced, customer information is
updated, or organizational structures evolve.
 Facts: These are the measurable, quantitative data points, such as sales
amounts, quantities, or transaction counts. Facts are typically historical
records and do not change once they are recorded.
Therefore, dimensions are more likely to change due to updates in descriptive attributes,
while facts remain static once they are recorded.
s
spra3049@[Link]
I have a dimension that has 3 columns: Department ID, Department Name, Employee
name, Employee Salary Your source is a CSV. Write code to provide the following output:
3rd highest salary by department.
Assistant
To find the 3rd highest salary by department from a CSV file, you can use PySpark. Here
is the code to achieve this:

%python
# Load the CSV file into a DataFrame
df = [Link]("/path/to/your/
[Link]", header=True,
inferSchema=True)
# Register the DataFrame as a
temporary view
[Link]
("employee_data")
# SQL query to find the 3rd highest
salary by department
query = """
SELECT DepartmentID, DepartmentName,
EmployeeName, EmployeeSalary
FROM (
SELECT
DepartmentID,
DepartmentName,
EmployeeName,
EmployeeSalary,
DENSE_RANK() OVER (PARTITION
BY DepartmentID ORDER BY
EmployeeSalary DESC) as rank
FROM employee_data
) ranked_data
WHERE rank = 3
"""
# Execute the query
result_df = [Link](query)
# Display the result
display(result_df)
s
spra3049@[Link]
Let’s assume you are loading transaction data with the following scenario: It requires a
historical load of 2 years worth of data. Incremental data needs to be updated using a
kafka topics Question: Describe how you would model and deploy given the scenario
without loosing any data records or without duplicating records.
Assistant
To model and deploy a data pipeline that handles both historical and incremental data
without losing or duplicating records, you can use a combination of batch processing for
the historical load and stream processing for the incremental updates. Here’s a step-by-
step approach:
Step 1: Historical Load

1. Load Historical Data: Use a batch process to load the 2 years worth of
historical data into a Delta Lake table.
2. Schema Definition: Define the schema for your transaction data, ensuring it
includes necessary fields such as transaction ID, timestamp, and other
relevant attributes.

%python
# Load historical data into a
DataFrame
historical_df = [Link]
("csv").option("header", "true").load
("/path/to/historical/data")
# Write the historical data to a
Delta table
historical_df.[Link]("delta").
mode("overwrite").save("/path/to/
delta/table")

Step 2: Incremental Data Ingestion

1. Kafka Integration: Set up a Structured Streaming job to read from Kafka

topics for the incremental data.
2. Schema Enforcement: Ensure the schema of the incoming data matches the
schema of the historical data.

%python
# Read from Kafka
kafka_df = [Link]
("kafka") \
.option("[Link].
servers",
"your_kafka_server:port") \
.option("subscribe",
"your_kafka_topic") \
.load()
# Define the schema for the Kafka data
schema = StructType([
StructField("transaction_id",
StringType(), True),
StructField("timestamp",
TimestampType(), True),
# Add other relevant fields
])
# Parse the Kafka data
parsed_df = kafka_df.selectExpr("CAST
(value AS STRING) as json") \
.select(from_json("json", schema).
alias("data")) \
.select("data.*")

Step 3: Merge Incremental Data with Historical Data

1. Upsert Logic: Use the MERGE operation in Delta Lake to upsert the
incremental data into the Delta table. This ensures that new records are
inserted and existing records are updated without duplication.

%python
from [Link] import DeltaTable
# Load the Delta table
delta_table = [Link]
(spark, "/path/to/delta/table")
# Define the merge condition and the
update/insert actions
delta_table.alias("tgt").merge(
parsed_df.alias("src"),
"tgt.transaction_id = src.
transaction_id"
).whenMatchedUpdateAll().
whenNotMatchedInsertAll().execute()

Step 4: Ensure Exactly-Once Processing

1. Checkpointing: Use checkpointing to ensure exactly-once processing

semantics in your streaming job.

%python
# Write the stream to the Delta table
with checkpointing
parsed_df.[Link]("delta")
\
.outputMode("append") \
.option("checkpointLocation", "/
path/to/checkpoint") \
.start("/path/to/delta/table")

Step 5: Monitoring and Maintenance

1. Monitoring: Set up monitoring to track the health and performance of your

data pipeline.
 2. Maintenance: Periodically optimize the Delta table by
running OPTIMIZE and VACUUM commands to maintain performance.

-- Optimize the Delta table

OPTIMIZE delta.`/path/to/delta/table`
-- Vacuum the Delta table to remove
old files
VACUUM delta.`/path/to/delta/table`
RETAIN 168 HOURS
By following these steps, you can ensure that your data pipeline handles both historical
and incremental data efficiently, without losing or duplicating records.
s
spra3049@[Link]
You are loading transaction data that contains various fields including a product id. Let’s
assume that the product id is part of a product hierarchy. This data flow supports a
dashboard that analyses last year’s sales versus current year’s sales of product category
B. Last year the same category used be A. Describe how you would model your dataflow
to seamlessly view product sales for these categories without needing to reload your
data.
Assistant
To model your data flow to seamlessly view product sales for categories that have
changed names without needing to reload your data, you can use a combination of a
dimension table for the product hierarchy and a fact table for the transaction data.
Here’s how you can achieve this:

Step 1: Create a Product Dimension Table

1. Product Dimension Table: Create a dimension table that includes the

product hierarchy and maps old category names to new category names.

CREATE TABLE product_dimension (

product_id STRING,
product_name STRING,
category_name STRING,
old_category_name STRING
);

Step 2: Populate the Product Dimension Table

1. Insert Data: Insert data into the product dimension table, ensuring that you
map the old category names to the new category names.

INSERT INTO product_dimension

(product_id, product_name,
category_name, old_category_name)
VALUES
('P1', 'Product 1', 'B', 'A'),
('P2', 'Product 2', 'B', 'A'),
-- Add more products as needed
;

Step 3: Create a Transaction Fact Table

1. Transaction Fact Table: Create a fact table that includes transaction data
with a reference to the product ID.

CREATE TABLE transaction_fact (

transaction_id STRING,
product_id STRING,
transaction_date DATE,
sales_amount DOUBLE
);

Step 4: Populate the Transaction Fact Table

1. Insert Data: Insert historical and current transaction data into the transaction
fact table.

INSERT INTO transaction_fact

(transaction_id, product_id,
transaction_date, sales_amount)
VALUES
('T1', 'P1', '2022-01-15', 100.0),
('T2', 'P2', '2023-01-15', 150.0),
-- Add more transactions as needed
;

Step 5: Create a View for Seamless Analysis

1. Create a View: Create a view that joins the transaction fact table with the
product dimension table to map old and new category names.

CREATE OR REPLACE VIEW sales_analysis

AS
SELECT
t.transaction_id,
t.product_id,
t.transaction_date,
t.sales_amount,
COALESCE(p.category_name, p.
old_category_name) AS
category_name
FROM
transaction_fact t
JOIN
 product_dimension p
ON
t.product_id = p.product_id;

Step 6: Query the View for Analysis

1. Query the View: Use the view to query and analyze sales data for the
desired categories.

SELECT
category_name,
YEAR(transaction_date) AS year,
SUM(sales_amount) AS total_sales
FROM
sales_analysis
WHERE
category_name = 'B'
GROUP BY
category_name, YEAR
(transaction_date);
By following these steps, you can seamlessly view product sales for categories that have
changed names without needing to reload your data. The use of a dimension table to
map old and new category names ensures that your data flow remains consistent and
efficient.
s
spra3049@[Link]
When you are creating a job, can you describe shuffle partitions. How much shuffle
partition should be used in a job? What is the difference between re-portion and coalesce
partition? What is the purpose of coalesce partition? When you reduce the # of
partitions, what will happen to the table skewness? What is a broadcasting join? What are
the advantaged of unity catalog?
Assistant

Shuffle Partitions
Shuffle partitions are the partitions created during shuffle operations in Spark. A shuffle
operation redistributes data across partitions, which is necessary for operations
like reduceByKey, groupByKey, join, and repartition. The number of shuffle partitions
can significantly impact the performance of your Spark job.

How Much Shuffle Partition Should Be Used in a Job?

The number of shuffle partitions should be set based on the size of your data and the
resources available in your cluster. A common rule of thumb is to set the number of
shuffle partitions to a multiple of the number of cores in your cluster. You can adjust the
number of shuffle partitions using the [Link] configuration.

Difference Between Repartition and Coalesce Partition

 Repartition: This operation reshuffles the data across a specified number of

partitions. It can increase or decrease the number of partitions and involves a
full shuffle of the data.
  Coalesce: This operation reduces the number of partitions without a full
shuffle. It is more efficient than repartition when decreasing the number of
partitions because it avoids the full shuffle.

Purpose of Coalesce Partition

The purpose of coalesce is to reduce the number of partitions in a more efficient manner
than repartition. It is particularly useful when you want to reduce the number of
partitions to optimize performance for subsequent operations that do not require a full
shuffle.

Impact on Table Skewness When Reducing the Number of

Partitions
When you reduce the number of partitions, the data may become more skewed if the
data is not evenly distributed. This can lead to some partitions being much larger than
others, which can cause performance issues due to uneven workload distribution.

Broadcasting Join
A broadcasting join is a type of join in Spark where a small dataset is broadcasted to all
nodes in the cluster. This allows the join to be performed locally on each node, avoiding
the need for a shuffle. It is efficient for joins where one of the datasets is small enough to
fit in memory.

Advantages of Unity Catalog

Unity Catalog provides several advantages:

 Centralized Governance: It offers a unified governance solution for all data

assets, including tables, files, and machine learning models.
 Fine-Grained Access Control: It allows for fine-grained access control at the
table, row, and column levels.
 Data Lineage: It provides data lineage tracking to understand data flow and
dependencies.
 Audit Logging: It supports audit logging to track data access and changes
for compliance and security purposes.

Example Code for Setting Shuffle Partitions

%python
# Set the number of shuffle partitions
[Link]("[Link].
partitions", 200)
1
2
s
spra3049@[Link]
Data modeling: List and describe dimensional models What is a fact and what is a
dimension? What is more likely to change, fact or dimension? What is SCD type 2?
Assistant

Dimensional Models
Dimensional modeling is a design technique used in data warehousing to structure data
for easy retrieval and analysis. The main components of dimensional models are fact
tables and dimension tables.

1. Star Schema: This is the simplest form of a dimensional model, where a

central fact table is connected to multiple dimension tables. Each dimension
table is directly related to the fact table.
2. Snowflake Schema: This is a more complex form of a star schema where
dimension tables are normalized into multiple related tables. This reduces
redundancy but can make queries more complex.
3. Galaxy Schema: Also known as a fact constellation schema, it contains
multiple fact tables that share dimension tables. This is useful for complex
data warehouses with multiple business processes.

Fact and Dimension

 Fact: A fact table contains the measurable, quantitative data of a business

process. Examples include sales amount, transaction count, and revenue. Fact
tables typically have foreign keys that reference dimension tables.
 Dimension: A dimension table contains descriptive attributes related to the
facts, such as product details, customer information, and time periods.
Dimensions provide context to the facts and are used for filtering, grouping,
and labeling.

More Likely to Change: Fact or Dimension?

Dimensions are more likely to change than facts. This is because dimensions contain
descriptive attributes that can be updated over time, such as customer addresses or
product names. Facts, on the other hand, are historical records of transactions and
typically do not change once recorded.

SCD Type 2
Slowly Changing Dimension (SCD) Type 2 is a method used to track historical changes in
dimension tables. When an attribute value changes, a new record is added to the
dimension table with a new surrogate key. This allows you to maintain a history of
changes and see the state of the dimension at any point in time.
Example of SCD Type 2:

-- Original record
INSERT INTO customer_dimension
(customer_id, name, address,
start_date, end_date)
VALUES (1, 'John Doe', '123 Main St',
'2021-01-01', NULL);
-- Address change
UPDATE customer_dimension
SET end_date = '2022-01-01'
WHERE customer_id = 1 AND end_date IS
NULL;
INSERT INTO customer_dimension
(customer_id, name, address,
start_date, end_date)
VALUES (1, 'John Doe', '456 Elm St',
'2022-01-01', NULL);
This approach ensures that you can track the history of changes to the customer's
address over time.

References
 1

Tutorial: Build an ETL pipeline using change data capture with DLT: Step 5:
Slowly Changing Dimension of type 2 (SCD2): It's often required to create a
table tracking all the changes resulting from APPEND, UPDATE, and DELETE:
History: You want to keep a history of all the changes to your table.
Traceability: You want to see which operation occurred.

2024-09-13 - E2E Testing User Interview -

Sri Ram

Owned by Benjamin Sivoravong

Last updated: Sep 13, 2024
2 min read

11 people viewed Karma Page Builder Request approval

Attendees:
 Ben Sivoravong
  Sri Ram Rachakonda

Notes
What team, projects are you working on right now, or in the past?

 Right now it’s almost all data ingestion with Veera, more Bronze
layer tables
o From postgres, sqlserver, mongo, OTLP databases for app
teams. Some kind of CDC, ETL processes (but all batch)
 Before, worked on Speedway migration, and aggregation-level data
for PowerBI reports

What is the overall workflow, process for developing a pipeline and

pushing it to production?

 First we build a utility for this type of data-source (like postgres, sql
server, etc.)
 We do a giant bulk load, then setup the incremental load on a
schedule
o For SQL, it’s really easy, table to table
o For NoSQL, we usually have to write some custom logic to
map the data to delta tables
 Raise a PR, get peers to review it
o Used to be 2 peers + manager, but now it’s like 1 peer +
manager, because CAB takes a while. They might be switching
back soon
 After the Merge, it goes directly to production
 There is no UAT environment, it would be very helpful to have one
though, run for a couple days or weeks

How do you validate your pipeline before deployment?

 We just do some ad-hoc tests in the dev environment, manually

 For SQL ingest, we just compare tables to tables (e.g. postgres to
delta)

How do you track what scenarios you should look for?

 No specific thing, but we usually just choose a few store ids or

transaction ids and spot check
 Because this is just bronze-level ingest, there's no real logic to
validate
And what did you do for the more complex aggregations for powerbi?

 We would just break it down for 1 record and manually check that it
matches manually.
 Sometimes we could get some other dataset (that’s already being
used as a source of truth) and compare our values to our expected
values.
 scripts, dashboards are deployed to dev, and we check if that
matches
 Problem: the datasets that people bring in are already flawed, so
they’re not good to use as a source-of-truth. Our logic may be
correct, but it doesn’t match because the reference is wrong.

How do you get and manage test data for building the pipelines?

 We just pull prod data, and create a dataset in dev env

o This is just a copy, not a delta-share. It’s a point-in-time thing
o This works well, because it’s only read-only, and we can have
all our prod data in the dev environment.

Tool / library questions:

 How familiar and comfortable would you be writing unit tests for
your code?
o If it’s defined, I can write it, it would be nice to have a general
library to follow. Guidelines and references.
 Any libraries or frameworks that you’ve worked with in the past?
o Nothing specific
 Opinion on who writes e2e tests?
o I want to write my e2e tests. But the process is not defined
right now
o I want to own the full project e2e, so I don’t have to hand it off
to another person, do a KT, etc.

Any other feedback or things to take into consideration?

 I would like for all of this to be well-documented, so it’s easy to

follow when it is implemented. A beginner should be able to follow
it.
2. What are the transformations we use in bronze,silver and gold layer.
3. how to create delta tables and where it will store.
4. how to write code to optimize the performance of notebook.
5. how data warehouse created in databricks.
6. how we use unity catalog and how to create/use those.
7. autoloader,sckewness,vaccume, how we use delta workflows.
8. detailed cicd process after databricks work.
9. delta workflow archestration.
[Link] to load gen2 files.
Assistant

2. Transformations in Bronze, Silver, and Gold Layers

In a typical data lake architecture, data is organized into three layers: Bronze, Silver, and
Gold. Each layer serves a different purpose and involves different types of
transformations.

 Bronze Layer: Raw data ingestion. This layer stores raw, unprocessed data
as it arrives. Transformations are minimal and typically involve data ingestion
and basic cleaning.

%python
bronze_df = [Link]("json").
load("/path/to/bronze/data")
bronze_df.[Link]("delta").save
("/path/to/delta/bronze")

 Silver Layer: Cleansed and enriched data. This layer involves more complex
transformations such as data cleansing, deduplication, and enrichment.

%python
silver_df = bronze_df.filter("status
= 'active'").dropDuplicates(["id"])
silver_df.[Link]("delta").mode
("overwrite").save("/path/to/delta/
silver")

 Gold Layer: Aggregated and business-level data. This layer involves

aggregations and business logic to create final datasets for reporting and
analytics.

%python
gold_df = silver_df.groupBy
("category").agg({"sales": "sum"})
gold_df.[Link]("delta").mode
("overwrite").save("/path/to/delta/
gold")

3. Creating Delta Tables and Storage

Delta tables can be created using Spark DataFrames and stored in Delta Lake, which is
typically backed by cloud storage like Azure Data Lake Storage (ADLS).

%python
# Create a Delta table
df = [Link]("csv").option
("header", "true").load("/path/to/
csv")
[Link]("delta").save("/path/
to/delta/table")
# Create a Delta table with SQL
[Link]("""
CREATE TABLE delta_table
USING delta
LOCATION '/path/to/delta/table'
AS SELECT * FROM source_table
""")
Delta tables are stored in the specified location, which is usually a directory in your cloud
storage.

4. Optimizing Notebook Performance

To optimize the performance of a Databricks notebook, consider the following
techniques:

 Caching: Cache intermediate DataFrames to avoid recomputation.

%python
[Link]()

 Partitioning: Repartition DataFrames to optimize parallelism.

%python
df = [Link](200)

 Broadcast Joins: Use broadcast joins for small tables.

%python
from [Link] import
broadcast
result = [Link](broadcast(df2),
"key")

 Delta Lake Optimizations: Use Delta Lake features

like OPTIMIZE and VACUUM.

OPTIMIZE delta.`/path/to/delta/table`
VACUUM delta.`/path/to/delta/table`
 RETAIN 168 HOURS

5. Creating a Data Warehouse in Databricks

A data warehouse in Databricks can be created using Delta Lake tables and SQL-based
transformations.

-- Create a database
CREATE DATABASE my_data_warehouse;
-- Create tables
CREATE TABLE my_data_warehouse.sales
USING delta
AS SELECT * FROM raw_sales_data;
-- Perform transformations
INSERT INTO my_data_warehouse.sales
SELECT * FROM transformed_sales_data;

6. Using Unity Catalog

Unity Catalog provides centralized governance for data and AI assets. To use Unity
Catalog, you need to create and manage catalogs, schemas, and tables.

-- Create a catalog
CREATE CATALOG my_catalog;
-- Create a schema
CREATE SCHEMA my_catalog.my_schema;
-- Create a table
CREATE TABLE my_catalog.my_schema.
my_table (
id INT,
name STRING
);

7. Autoloader, Skewness, Vacuum, and Delta Workflows

 Autoloader: Automatically loads data from cloud storage into Delta Lake.

%python
df = [Link]
("cloudFiles") \
.option("[Link]",
"json") \
.load("/path/to/source")
[Link]("delta").
option("checkpointLocation", "/path/
to/checkpoint").start("/path/to/
delta")

 Skewness: Address data skewness by repartitioning or using skew join hints.

%python
df = [Link]("key")

 Vacuum: Clean up old data files.

VACUUM delta.`/path/to/delta/table`
RETAIN 168 HOURS

8. CI/CD Process After Databricks Work

A typical CI/CD process involves version control, automated testing, and deployment.

1. Version Control: Use Git to manage code changes.

2. Automated Testing: Use tools like pytest for testing.
3. Deployment: Use Azure DevOps or Jenkins for deployment.

9. Delta Workflow Orchestration

Orchestrate Delta workflows using Databricks Jobs or Apache Airflow.

%python
# Using Databricks Jobs
[Link]("/path/to/
notebook", 60)
# Using Airflow
from airflow import DAG
from [Link].
dagrun_operator import
TriggerDagRunOperator
dag = DAG('delta_workflow',
default_args=default_args,
schedule_interval='@daily')
trigger = TriggerDagRunOperator(
task_id='trigger_delta_workflow',
trigger_dag_id='delta_workflow',
dag=dag
)

10. Loading Gen2 Files

To load files from Azure Data Lake Storage Gen2, use the following code:
Python
%python
# Configure access
[Link]("[Link].
<storage-account-name>.[Link].
[Link]", "<access-key>")
# Load data
df = [Link]("csv").option
("header", "true").load("abfss://
<container-name>@<storage-account-name
>.[Link]/<path>")
display(df)