Objective

• Visualization of current available stock
• Projection of future inventory
• Efficient replenishment management
• Identification of shortage/excess risks
• Optimization of supply planning

2.  Architecture

BRONZE:
Result of data extraction from SAP to Kafka using the OneConnect tool.
This layer stores the RAW tables sourced from the relevant SAP modules (SD, MM, PP).

SILVER:
Represents the normalized entities that group and structure the data extracted from SAP.
Each entity corresponds to a specific SAP module, consolidating the information to build the Data Product.

GOLD:

These are the final consolidated entities, where transformations, business rules, and the calculation of key metrics required for the Inventory Status dashboard are applied.

2.1 Data Flow

The data flow of the "Inventory Status" Data Product follows this path through the architecture layers:

1. Extraction (BRONZE):
Data is extracted from the following SAP modules:

• Sales and Distribution (SD): Tables VBAK, VBAP, VBEP, VBKD, VBPA, VBUK
• Materials Management (MM): Tables MARD, MARC, and EKKO, EKPO, EINE, EKET, EKBE for purchase orders
• Production Planning (PP): Tables AFKO, AUFK, AFPO, RESB, AFVV, AFRU
• Catalogs: Tables MAKT, T001W, TSPAT, TVKBT, TVKOT, TVSBT, TVTWT

2. Normalization (SILVER):
The RAW tables are integrated into the following entities:

• SALES_ORDER: Combines sales-related information
• PRODUCTION_ORDER: Represents production orders
• PURCHASE_ORDER: Groups purchase order data
• INVENTORY: Organizes inventory-related data

3. Consolidation and Transformation (GOLD):
Based on the Silver entities:

• OPEN_ORDER_TRACKER and OPEN_RECEIVINGS_TRACKER consolidate the status of open orders
• Finally, the INVENTORY_SITUATION entity is built as the main output, containing integrated data on current inventory, open orders, and pending receipts

4. Flink SQL 

4.1 SILVER_SD_SALES_ORDER 

CREATE TABLE SILVER_SD_SALES_ORDER (

   PRIMARY KEY (mandt_vbak, vbeln_vbak, posnr_vbap) NOT ENFORCED

) WITH (

   'changelog.mode' = 'upsert',

   'kafka.cleanup-policy' = 'compact'

) AS (

 SELECT

   --mandt - client

   coalesce(vbak.mandt, '') as mandt_vbak,

   --vbeln - sales_doc

   coalesce(vbak.vbeln, '') as vbeln_vbak, 

   --posnr -  item

   coalesce(vbap.posnr, 0) as posnr_vbap,

   --kunnr - sold_to_pt

   vbak.kunnr as kunnr_vbak,

   --vdatu - reqdlvdt

   vbak.vdatu as vdatu_vbak,

   --audat - doc_date

   vbak.audat as audat_vbak,

   --vsbed - shp_cond

   vbak.vsbed as vsbed_vbak,

   --vkorg - sales_org

   vbak.vkorg as vkorg_vbak,

   --vtweg - distr_chl

   vbak.vtweg as vtweg_vbak,

   --vkbur - sales_off

   vbak.vkbur as vkbur_vbak,

   --gbstk - ovrll_sts

   --vbak.gbstk as gbstk_vbak,

   --ECC

   vbuk.gbstk as gbstk_vbak,

   --matnr - material

   vbap.matnr as matnr_vbap, 

   --werks - plant

   vbap.werks as werks_vbap,

   --kwmeng - order_qty

   vbap.kwmeng as kwmeng_vbap,

   --matkl - matl_group

   vbap.matkl as matkl_vbap,

   --prodh - prodhier

   vbap.prodh as prodh_vbap,

   --spart - division

   vbap.spart as spart_vbap,

   --lprio - dlv_prior

   vbap.lprio as lprio_vbap,

   --route - route

   vbap.`route` as route_vbap,

   --land1 - ctryreg_vbpa

   vbpa.land1 as land1_vbpa,

   --adrnr - addr_no

   vbpa.adrnr as adrnr_vbpa,

   --lzone - zone

   vbpa.lzone as lzone_vbpa

 FROM `VBAK` AS vbak 

 /*ECC*/

 INNER JOIN VBUK AS vbuk 

   ON --vbak.`eventid` = vbap.`eventid` AND 

   vbak.`mandt` = vbuk.`mandt` 

   and vbak.`vbeln` = vbuk.`vbeln`

 INNER JOIN VBAP AS vbap 

   ON --vbak.`eventid` = vbap.`eventid` AND 

   vbak.`mandt` = vbap.`mandt` 

   and vbak.`vbeln` = vbap.`vbeln`

 LEFT JOIN VBPA AS vbpa 

   ON --vbak.`eventid` = vbpa.`eventid` and 

   vbak.`mandt` = vbpa.`mandt` 

   and vbak.`vbeln` = vbpa.`vbeln`

   and vbpa.`posnr` = 0 and vbpa.`parvw` = 'WE'

);

4.2 SILVER_MM_INVENTORY

CREATE TABLE SILVER_MM_INVENTORY (

   PRIMARY KEY (mandt, matnr, werks, lgort) NOT ENFORCED

) WITH (

   'changelog.mode' = 'upsert',

   'kafka.cleanup-policy' = 'compact'

) AS (

SELECT 

 coalesce( mard.mandt, '') as mandt,

 coalesce(mard.matnr, '') as matnr ,

 coalesce( mard.werks, '') werks,

 coalesce( mard.lgort, '') lgort,

 --MARD-UMLME-stockintfr1

 mard.umlme AS umlme_mard,

 --MARD-EINME-restricted1

 mard.einme as einme_mard,

 --MARD-INSME-qual_insp1

 mard.insme insme_mard,

 --mard-SPEME-blocked3

 mard.speme as speme_mard,

 --MARD-RETME-returns1

 mard.retme retme_mard,

 --MARD-LABST-unrestr2

 mard.labst labst_mard,

 --MABST

 marc.mabst mabst_marc,

 --EISBE

 marc.eisbe eisbe_marc

FROM MARD AS mard

INNER JOIN MARC as marc

 ON

   mard.mandt = marc.mandt 

 AND mard.werks = marc.werks 

 and mard.matnr = marc.matnr

);

4.3 SILVER_PP_PRODUCTION_ORDER

CREATE TABLE SILVER_PP_PRODUCTION_ORDER (

   PRIMARY KEY (mandt_afpo, aufnr_afpo, posnr_afpo, mandt_afru, rueck_afru, rmzhl_afru) NOT ENFORCED

) WITH (

   'changelog.mode' = 'upsert',

   'kafka.cleanup-policy' = 'compact'

) AS (

SELECT

 COALESCE(afpo.`mandt`,'') as mandt_afpo,

 COALESCE(afpo.`aufnr`, '') as aufnr_afpo,

 COALESCE(afpo.`posnr`, 0) as posnr_afpo,

 COALESCE(afru.`mandt`, '') as mandt_afru,

 --RUECK

 COALESCE(afru.`rueck`, 0) as rueck_afru,

 --RMZHL

 COALESCE(afru.`rmzhl`, 0) as rmzhl_afru,

-- afpo

 afpo.`dwerk` as dwerk_afpo,

 afpo.`matnr` as matnr_afpo,

 afpo.`dglts` as dglts_afpo,

 CAST(afpo.`psmng` AS DECIMAL(20, 4)) as psmng_afpo,

 --afru

  COALESCE(afru.`aueru`, '') as aueru_afru

FROM AFPO AS afpo  

LEFT JOIN AFRU AS afru 

 ON --afko.`eventid` = afru.`eventid` AND 

 afpo.`mandt` = afru.`mandt` 

 AND afpo.`aufnr` = afru.`aufnr`);

4.4 SILVER_MM_PURCHASE_ORDER

CREATE TABLE SILVER_MM_PURCHASE_ORDER (

   PRIMARY KEY (mandt_ekpo, ebeln_ekpo, ebelp_ekpo, etenr_eket) NOT ENFORCED

) WITH (

   'changelog.mode' = 'upsert',

   'kafka.cleanup-policy' = 'compact'

) AS (

SELECT

 --mandt

 COALESCE(ekpo.`mandt`, '') as mandt_ekpo,

 --ebeln - pur_doc

 COALESCE(ekpo.`ebeln`,'') AS ebeln_ekpo,

 --ebelp - item

 COALESCE(ekpo.`ebelp`, 0) as ebelp_ekpo,

 --etenr - schedule_line

 COALESCE(eket.`etenr` ,0) as etenr_eket,

 --werks - plant

 ekpo.`werks` werks_ekpo,

 --matnr - material

 ekpo.`matnr` matnr_ekpo,

 --menge - quantity

 CAST( ekpo.`menge` AS DECIMAL(20, 4)) as menge_ekpo,

 --elikz - del_cmpl

 ekpo.`elikz` as elikz_ekpo,

 --pstyp - item_categ

 ekpo.`pstyp` as pstyp_ekpo,

 --slfdt - stat_del_date

 COALESCE(eket.`slfdt`,'') as slfdt_eket

FROM  EKPO AS ekpo 

LEFT JOIN EKET AS eket  

 ON  --eket.`eventid` = ekpo.`eventid` AND 

 eket.`mandt` = ekpo.`mandt` 

 AND eket.`ebeln` = ekpo.`ebeln` 

 AND eket.`ebelp` = ekpo.`ebelp`

);

4.5 SILVER_PP_PRODUCTION_ORDER 

CREATE TABLE GOLD_SD_OPEN_ORDER_TRACKER (

   PRIMARY KEY (client, sales_order, item) NOT ENFORCED

) WITH (

   'changelog.mode' = 'upsert',

   'kafka.cleanup-policy' = 'compact'

) AS (

with inventory AS (

 SELECT

   mandt,

   matnr,

   werks,

   sum(labst_mard) as labst

 FROM SILVER_MM_INVENTORY AS mard

 GROUP BY mandt,

   matnr,

   werks

)

SELECT 

 sso.mandt_vbak as client,

 sso.vbeln_vbak as sales_order, 

 sso.posnr_vbap as item,

 sso.matnr_vbap as material_id, 

 sso.werks_vbap as plant_id,

 sso.kwmeng_vbap as order_qty,

 sso.kunnr_vbak as customer_id,

 sso.matkl_vbap as matl_group_vbap,

 sso.vdatu_vbak as delivery_date,

 TO_TIMESTAMP(sso.vdatu_vbak, 'yyyy-MM-dd') AS delivery_date_ts,

 SUBSTRING(sso.vdatu_vbak FROM 1 FOR 7)  as delivery_month,

 sso.audat_vbak as order_date,

 sso.adrnr_vbpa as addr_no,

 sso.vsbed_vbak as shipping_conditions_id,

 sso.vkorg_vbak as sales_org_id,

 sso.vtweg_vbak as distr_chl_id,

 sso.vkbur_vbak as sales_office_id,

 sso.prodh_vbap as prodh_id,

 sso.spart_vbap as division_id,

 sso.lprio_vbap as delivery_priority,

 sso.route_vbap as route_id,

 sso.lzone_vbpa as zone_id,

 sso.gbstk_vbak as ovrll_sts,

 TIMESTAMPDIFF(DAY, CAST(sso.`$rowtime` AS TIMESTAMP), TO_TIMESTAMP(sso.vdatu_vbak, 'yyyy-MM-dd')) AS delivery_time,

 LEAST(TIMESTAMPDIFF(DAY, CAST(sso.`$rowtime` AS TIMESTAMP), TO_TIMESTAMP(sso.vdatu_vbak, 'yyyy-MM-dd')), 8) AS delivery_status,

 ssiv.labst  as  avaliable_stock,

 t001w.pstlz as postal_code,

 t001w.name1  as plant, 

 makt.maktg as material,

 tvkot.vtext as sales_org,

 tvtwt.vtext as channel, 

 tspat.vtext as division, 

 tvkbt.bezei as sales_office, 

 tvsbt.vtext as shipping_conditions,

 kna1.name1 as customer,

 CASE

   WHEN sso.gbstk_vbak = 'C' THEN 'CLOSE'

   ELSE 'OPEN'

 END as order_status

 FROM SILVER_SD_SALES_ORDER as sso

 INNER JOIN inventory as ssiv 

   ON ssiv.matnr  = sso.matnr_vbap 

   and ssiv.werks  = sso.werks_vbap

 LEFT JOIN T001W as t001w

   ON t001w.`spras` = 'E' 

   AND t001w.werks = sso.werks_vbap

 LEFT JOIN MAKT AS makt 

   ON t001w.`spras` = makt.`spras` 

   AND makt.matnr = sso.matnr_vbap

 LEFT JOIN TVKOT AS tvkot

   ON t001w.`spras` = tvkot.`spras` 

   AND tvkot.vkorg = sso.vkorg_vbak

 LEFT JOIN TVTWT AS tvtwt 

   ON t001w.`spras` = tvtwt.`spras` 

   AND tvtwt.vtweg = sso.vtweg_vbak

 LEFT JOIN  TSPAT AS tspat 

   ON t001w.`spras` = tspat.`spras` 

   AND tspat.spart = sso.spart_vbap

 LEFT JOIN TVKBT AS tvkbt 

   ON t001w.`spras` = tvkbt.`spras` 

   AND tvkbt.vkbur = sso.vkbur_vbak

 LEFT JOIN TVSBT AS tvsbt 

   ON t001w.`spras` = tvsbt.`spras` 

   AND tvsbt.vsbed = sso.vsbed_vbak

 LEFT JOIN KNA1 as kna1

   ON kna1.kunnr = sso.kunnr_vbak

 );

4.6 GOLD_ORDER_TRACKING_RECEPTION

CREATE TABLE GOLD_ORDER_TRACKING_RECEPTION (

   PRIMARY KEY (client, operation, `order`, item) NOT ENFORCED

) WITH (

   'changelog.mode' = 'upsert',

   'kafka.cleanup-policy' = 'compact'

) AS (

WITH inventory AS (

 SELECT

   mandt,

   matnr,

   werks,

   sum(labst_mard) as labst

 FROM SILVER_MM_INVENTORY AS mard

 GROUP BY mandt,

   matnr,

   werks

),

............. To get complete access to the SQL query, please get in contact with the Onibex Team at contact@onibex.com

4.7 GOLD_INVENTORY_SITUATION 

 CREATE TABLE GOLD_INVENTORY_SITUATION(

   future_date DATE,

   client STRING,

   material_id STRING,

   plant_id STRING,

   daily_qty_production_order DECIMAL(20, 4),

   cumulative_production_order DECIMAL(20, 4),

   daily_qty_purchase_order DECIMAL(20, 4),

   cumulative_purchase_order DECIMAL(20, 4),

   daily_qty_stock_order DECIMAL(20, 4),

   cumulative_stock_order DECIMAL(20, 4),

   daily_qty_sale_order DECIMAL(20, 4),

   cumulative_sales_order DECIMAL(20, 4),

   daily_qty_receivings DECIMAL(20, 4),

   cumulative_receivings DECIMAL(20, 4),

   future_stock DECIMAL(20, 4),

   actual_stock DECIMAL(20, 4),

............. To get complete access to the SQL query, please get in contact with the Onibex Team at contact@onibex.com

5. Example Charts

 Open Order Stock by Material

    

Open Order Tracker

Future Stock vs Actual Stock

Receivings

Plant Stock Map