indonesian version column

dags/etl_food_security.py

@@ -22,6 +22,8 @@ Kimball ETL Flow:
     │                                                agg_pillar_by_country     │
     │                                                agg_framework_by_country  │
     │                                                agg_framework_asean       │
+    │                                                      ↓                   │
+    │                                                agg_indicator_norm        │
     │                                                                          │
     │  AUDIT : etl_logs, etl_metadata (setiap layer)                           │
     └──────────────────────────────────────────────────────────────────────────┘
@@ -36,13 +38,15 @@ Task Order:
         → dimensional_model_to_gold
         → analytical_layer_to_gold
         → aggregation_to_gold
+        → indicator_norm_aggregation_to_gold
 
 Scripts folder harus berisi:
-    - bigquery_raw_layer.py             (run_verify_connection, run_load_fao, ...)
+    - bigquery_raw_layer.py                         (run_verify_connection, run_load_fao, ...)
-    - bigquery_cleaned_layer.py         (run_cleaned_integration)
+    - bigquery_cleaned_layer.py                     (run_cleaned_integration)
-    - bigquery_dimensional_model.py     (run_dimensional_model)
+    - bigquery_dimensional_model.py                 (run_dimensional_model)
-    - bigquery_analytical_layer.py      (run_analytical_layer)
+    - bigquery_analytical_layer.py                  (run_analytical_layer)
-    - bigquery_analysis_aggregation.py  (run_aggregation)
+    - bigquery_analysis_aggregation.py              (run_aggregation)
+    - bigquery_aggraget_fact_selected_layer.py      (run_indicator_norm_aggregation)
     - bigquery_config.py
     - bigquery_helpers.py
     - bigquery_datasource.py
@@ -71,11 +75,14 @@ from scripts.bigquery_analytical_layer import (
 from scripts.bigquery_aggregate_layer import (
     run_aggregation,
 )
+from scripts.bigquery_aggraget_fact_selected_layer import (
+    run_indicator_norm_aggregation,
+)
 
 # DEFAULT ARGS
 
 default_args = {
-    'owner': 'data-engineering',
+    'owner': 'Debby Seftia',
     'email': ['d1041221004@student.untan.ac.id'],
 }
 
@@ -86,7 +93,7 @@ with DAG(
     description       = "Kimball ETL: FAO, World Bank, UNICEF → BigQuery (Bronze → Silver → Gold)",
     default_args      = default_args,
     start_date        = datetime(2026, 3, 1),
-    schedule_interval = "0 0 */3 * *",
+    schedule_interval = "0 0 1 */3 *",
     catchup           = False,
     tags              = ["food-security", "bigquery", "kimball"],
 ) as dag:
@@ -136,5 +143,21 @@ with DAG(
         python_callable = run_aggregation
     )
 
-    
+    task_indicator_norm = PythonOperator(
-    task_verify >> task_fao >> task_worldbank >> task_unicef >> task_staging >> task_cleaned >> task_dimensional >> task_analytical >> task_aggregation
+        task_id         = "indicator_norm_aggregation_to_gold",
+        python_callable = run_indicator_norm_aggregation
+    )
+
+    # Task Dependencies
+    (
+        task_verify
+        >> task_fao
+        >> task_worldbank
+        >> task_unicef
+        >> task_staging
+        >> task_cleaned
+        >> task_dimensional
+        >> task_analytical
+        >> task_aggregation
+        >> task_indicator_norm
+    )

scripts/bigquery_analytical_layer.py

@@ -8,15 +8,9 @@ Filtering Order:
 3. Filter complete indicators PER COUNTRY (auto-detect start year, no gaps)
 4. Filter countries with ALL pillars (FIXED SET)
 5. Filter indicators with consistent presence across FIXED countries
-6. Calculate YoY per indicator per country
+6. Save analytical table (dengan nama/label lengkap untuk Looker Studio)
-7. Save analytical table (dengan nama/label lengkap + kolom framework + YoY untuk Looker Studio)
 
-UPDATED:
+ADDED: Kolom indicator_name_id dan pillar_name_id (terjemahan Bahasa Indonesia)
-- Kolom 'framework' (MDGs/SDGs) dipropagasi dari dim_indicator ke tabel output.
-  Hal ini memungkinkan Looker Studio melakukan filter/slice berdasarkan framework
-  tanpa perlu join ulang ke dim_indicator.
-- Kolom 'yoy_change' dan 'yoy_pct' ditambahkan untuk analisis Year-over-Year
-  per indikator per negara langsung di Looker Studio.
 """
 
 import pandas as pd
@@ -42,6 +36,176 @@ from scripts.bigquery_helpers import (
 from google.cloud import bigquery
 
 
+# =============================================================================
+# TRANSLATION DICTIONARIES
+# =============================================================================
+
+PILLAR_TRANSLATION_ID: dict = {
+    # 4 pilar utama Food Security
+    "Availability"  : "Ketersediaan",
+    "Access"        : "Keterjangkauan",
+    "Utilization"   : "Pemanfaatan",
+    "Stability"     : "Stabilitas",
+    # Variasi penulisan yang mungkin muncul
+    "availability"  : "Ketersediaan",
+    "access"        : "Keterjangkauan",
+    "utilization"   : "Pemanfaatan",
+    "stability"     : "Stabilitas",
+    "Food Availability" : "Ketersediaan Pangan",
+    "Food Access"       : "Keterjangkauan Pangan",
+    "Food Utilization"  : "Pemanfaatan Pangan",
+    "Food Stability"    : "Stabilitas Pangan",
+}
+
+INDICATOR_TRANSLATION_ID: dict = {
+    # -------------------------------------------------------------------------
+    # AVAILABILITY
+    # -------------------------------------------------------------------------
+    "Average dietary energy supply adequacy (percent) (3-year average)":
+        "Kecukupan rata-rata pasokan energi makanan (persen) (rata-rata 3 tahun)",
+    "Average value of food production (constant 2014-2016 thousand US$) (3-year average)":
+        "Nilai rata-rata produksi pangan (ribu US$ konstan 2014-2016) (rata-rata 3 tahun)",
+    "Share of dietary energy supply derived from cereals, roots and tubers (percent) (3-year average)":
+        "Proporsi pasokan energi makanan dari serealia, akar, dan umbi-umbian (persen) (rata-rata 3 tahun)",
+    "Average protein supply (g/cap/day) (3-year average)":
+        "Rata-rata pasokan protein (g/kapita/hari) (rata-rata 3 tahun)",
+    "Average supply of protein of animal origin (g/cap/day) (3-year average)":
+        "Rata-rata pasokan protein hewani (g/kapita/hari) (rata-rata 3 tahun)",
+    "Cereal import dependency ratio (percent) (3-year average)":
+        "Rasio ketergantungan impor sereal (persen) (rata-rata 3 tahun)",
+    "Percent of arable land equipped for irrigation (percent) (3-year average)":
+        "Persentase lahan pertanian yang dilengkapi irigasi (persen) (rata-rata 3 tahun)",
+    "Crop production index (2014-2016 = 100)":
+        "Indeks produksi tanaman pangan (2014-2016 = 100)",
+    "Livestock production index (2014-2016 = 100)":
+        "Indeks produksi peternakan (2014-2016 = 100)",
+    "Value of food imports over total merchandise exports (percent) (3-year average)":
+        "Nilai impor pangan terhadap total ekspor barang (persen) (rata-rata 3 tahun)",
+    "Food production variability (constant 2014-2016 thousand US$ per capita)":
+        "Variabilitas produksi pangan (ribu US$ konstan 2014-2016 per kapita)",
+    "Food supply variability (kcal/cap/day)":
+        "Variabilitas pasokan pangan (kkal/kapita/hari)",
+
+    # -------------------------------------------------------------------------
+    # ACCESS
+    # -------------------------------------------------------------------------
+    "Gross domestic product per capita, PPP (constant 2017 international $)":
+        "Produk domestik bruto per kapita, PPP (internasional konstan 2017 US$)",
+    "Domestic food price level index (2015 = 1.00)":
+        "Indeks tingkat harga pangan domestik (2015 = 1,00)",
+    "Domestic food price volatility index":
+        "Indeks volatilitas harga pangan domestik",
+    "Prevalence of undernourishment (percent) (3-year average)":
+        "Prevalensi kekurangan gizi (persen) (rata-rata 3 tahun)",
+    "Number of people undernourished (million) (3-year average)":
+        "Jumlah penduduk kekurangan gizi (juta jiwa) (rata-rata 3 tahun)",
+    "Depth of the food deficit (kcal/capita/day) (3-year average)":
+        "Kedalaman defisit pangan (kkal/kapita/hari) (rata-rata 3 tahun)",
+    "Percentage of population using at least basic drinking water services (percent)":
+        "Persentase penduduk yang menggunakan layanan air minum dasar (persen)",
+    "Percentage of population using safely managed drinking water services (percent)":
+        "Persentase penduduk yang menggunakan layanan air minum yang dikelola dengan aman (persen)",
+    "Percentage of population using at least basic sanitation services (percent)":
+        "Persentase penduduk yang menggunakan layanan sanitasi dasar (persen)",
+    "Percentage of population using safely managed sanitation services (percent)":
+        "Persentase penduduk yang menggunakan layanan sanitasi yang dikelola dengan aman (persen)",
+    "Access to electricity (percent of rural population)":
+        "Akses listrik (persen penduduk pedesaan)",
+    "Proportion of population with access to electricity (percent)":
+        "Proporsi penduduk dengan akses listrik (persen)",
+    "Road infrastructure index":
+        "Indeks infrastruktur jalan",
+    "Rail lines density (total route-km per 100 square km of land area)":
+        "Kepadatan jalur kereta api (total rute-km per 100 km2 lahan)",
+    "Gross national income per capita (Atlas method, current US$)":
+        "Pendapatan nasional bruto per kapita (metode Atlas, US$ terkini)",
+    "Food Insecurity Experience Scale (FIES)":
+        "Skala Pengalaman Ketidakamanan Pangan (FIES)",
+
+    # -------------------------------------------------------------------------
+    # UTILIZATION
+    # -------------------------------------------------------------------------
+    "Prevalence of severe food insecurity in the total population (percent) (3-year average)":
+        "Prevalensi kerawanan pangan berat pada total penduduk (persen) (rata-rata 3 tahun)",
+    "Prevalence of severe food insecurity in the male adult population (percent) (3-year average)":
+        "Prevalensi kerawanan pangan berat pada penduduk laki-laki dewasa (persen) (rata-rata 3 tahun)",
+    "Prevalence of severe food insecurity in the female adult population (percent) (3-year average)":
+        "Prevalensi kerawanan pangan berat pada penduduk perempuan dewasa (persen) (rata-rata 3 tahun)",
+    "Prevalence of moderate or severe food insecurity in the total population (percent) (3-year average)":
+        "Prevalensi kerawanan pangan sedang atau berat pada total penduduk (persen) (rata-rata 3 tahun)",
+    "Prevalence of moderate or severe food insecurity in the male adult population (percent) (3-year average)":
+        "Prevalensi kerawanan pangan sedang atau berat pada penduduk laki-laki dewasa (persen) (rata-rata 3 tahun)",
+    "Prevalence of moderate or severe food insecurity in the female adult population (percent) (3-year average)":
+        "Prevalensi kerawanan pangan sedang atau berat pada penduduk perempuan dewasa (persen) (rata-rata 3 tahun)",
+    "Number of severely food insecure people (million) (3-year average)":
+        "Jumlah penduduk yang mengalami kerawanan pangan berat (juta jiwa) (rata-rata 3 tahun)",
+    "Number of severely food insecure male adults (million) (3-year average)":
+        "Jumlah laki-laki dewasa yang mengalami kerawanan pangan berat (juta jiwa) (rata-rata 3 tahun)",
+    "Number of severely food insecure female adults (million) (3-year average)":
+        "Jumlah perempuan dewasa yang mengalami kerawanan pangan berat (juta jiwa) (rata-rata 3 tahun)",
+    "Number of moderately or severely food insecure people (million) (3-year average)":
+        "Jumlah penduduk yang mengalami kerawanan pangan sedang atau berat (juta jiwa) (rata-rata 3 tahun)",
+    "Number of moderately or severely food insecure male adults (million) (3-year average)":
+        "Jumlah laki-laki dewasa yang mengalami kerawanan pangan sedang atau berat (juta jiwa) (rata-rata 3 tahun)",
+    "Number of moderately or severely food insecure female adults (million) (3-year average)":
+        "Jumlah perempuan dewasa yang mengalami kerawanan pangan sedang atau berat (juta jiwa) (rata-rata 3 tahun)",
+    "Percentage of children under 5 years of age who are stunted (modelled estimates) (percent)":
+        "Persentase anak di bawah 5 tahun yang mengalami stunting (estimasi model) (persen)",
+    "Number of children under 5 years of age who are stunted (modeled estimates) (million)":
+        "Jumlah anak di bawah 5 tahun yang mengalami stunting (estimasi model) (juta jiwa)",
+    "Percentage of children under 5 years affected by wasting (percent)":
+        "Persentase anak di bawah 5 tahun yang mengalami wasting (persen)",
+    "Number of children under 5 years affected by wasting (million)":
+        "Jumlah anak di bawah 5 tahun yang mengalami wasting (juta jiwa)",
+    "Percentage of children under 5 years of age who are overweight (modelled estimates) (percent)":
+        "Persentase anak di bawah 5 tahun yang mengalami kelebihan berat badan (estimasi model) (persen)",
+    "Number of children under 5 years of age who are overweight (modeled estimates) (million)":
+        "Jumlah anak di bawah 5 tahun yang mengalami kelebihan berat badan (estimasi model) (juta jiwa)",
+    "Prevalence of anemia among women of reproductive age (15-49 years) (percent)":
+        "Prevalensi anemia pada perempuan usia reproduksi (15-49 tahun) (persen)",
+    "Number of women of reproductive age (15-49 years) affected by anemia (million)":
+        "Jumlah perempuan usia reproduksi (15-49 tahun) yang menderita anemia (juta jiwa)",
+    "Prevalence of obesity in the adult population (18 years and older) (percent)":
+        "Prevalensi obesitas pada penduduk dewasa (18 tahun ke atas) (persen)",
+    "Prevalence of exclusive breastfeeding among infants 0-5 months of age (percent)":
+        "Prevalensi pemberian ASI eksklusif pada bayi usia 0-5 bulan (persen)",
+    "Minimum dietary diversity for women (MDD-W) (percent)":
+        "Keragaman pola makan minimum untuk perempuan (MDD-W) (persen)",
+
+    # -------------------------------------------------------------------------
+    # STABILITY
+    # -------------------------------------------------------------------------
+    "Cereal import dependency ratio (percent)":
+        "Rasio ketergantungan impor sereal (persen)",
+    "Political stability and absence of violence/terrorism (index)":
+        "Stabilitas politik dan tidak adanya kekerasan/terorisme (indeks)",
+    "Domestic food price volatility":
+        "Volatilitas harga pangan domestik",
+    "Per capita food supply variability (kcal/cap/day)":
+        "Variabilitas pasokan pangan per kapita (kkal/kapita/hari)",
+    "Percentage of arable land equipped for irrigation (percent)":
+        "Persentase lahan pertanian yang dilengkapi irigasi (persen)",
+    "GDP per capita growth (annual %)":
+        "Pertumbuhan PDB per kapita (% tahunan)",
+    "GDP growth (annual %)":
+        "Pertumbuhan PDB (% tahunan)",
+}
+
+
+def translate_indicator(name: str) -> str:
+    """Terjemahkan nama indikator ke Bahasa Indonesia. Fallback ke nama asli."""
+    if not name:
+        return name
+    return INDICATOR_TRANSLATION_ID.get(name, name)
+
+
+def translate_pillar(name: str) -> str:
+    """Terjemahkan nama pillar ke Bahasa Indonesia. Fallback ke nama asli."""
+    if not name:
+        return name
+    return PILLAR_TRANSLATION_ID.get(name, name)
+
+
 # =============================================================================
 # ANALYTICAL LAYER CLASS
 # =============================================================================
@@ -54,17 +218,13 @@ class AnalyticalLayerLoader:
     1. Complete per country (no gaps from start_year to end_year)
     2. Filter countries with all pillars
     3. Ensure indicators have consistent country count across all years
-    4. Calculate YoY (year-over-year) change per indicator per country
+    4. Save dengan kolom lengkap (nama + ID + nama Indonesia) untuk Looker Studio
-    5. Save dengan kolom lengkap (nama + ID + framework + YoY) untuk Looker Studio
 
     Output: fact_asean_food_security_selected -> DW layer (Gold) -> fs_asean_gold
 
-    Kolom output:
+    Kolom tambahan:
-        country_id, country_name,
+    - indicator_name_id : terjemahan Bahasa Indonesia dari indicator_name
-        indicator_id, indicator_name, direction, framework,
+    - pillar_name_id    : terjemahan Bahasa Indonesia dari pillar_name
-        pillar_id, pillar_name,
-        time_id, year, value,
-        yoy_change, yoy_pct
     """
 
     def __init__(self, client: bigquery.Client):
@@ -103,7 +263,6 @@ class AnalyticalLayerLoader:
         self.logger.info("=" * 80)
 
         try:
-            # Sertakan kolom framework dari dim_indicator dalam query
             query = f"""
             SELECT
                 f.country_id,
@@ -111,7 +270,6 @@ class AnalyticalLayerLoader:
                 f.indicator_id,
                 i.indicator_name,
                 i.direction,
-                i.framework,
                 f.pillar_id,
                 p.pillar_name,
                 f.time_id,
@@ -128,34 +286,15 @@ class AnalyticalLayerLoader:
             JOIN `{get_table_id('dim_time',           layer='gold')}` t ON f.time_id      = t.time_id
             """
 
-            self.logger.info("Loading fact table with dimensions (incl. framework)...")
+            self.logger.info("Loading fact table with dimensions...")
-            self.df_clean = self.client.query(query).result().to_dataframe(
+            self.df_clean = self.client.query(query).result().to_dataframe(create_bqstorage_client=False)
-                create_bqstorage_client=False
-            )
             self.logger.info(f"  Loaded: {len(self.df_clean):,} rows")
 
             if 'is_year_range' in self.df_clean.columns:
                 yr = self.df_clean['is_year_range'].value_counts()
                 self.logger.info(f"  Breakdown:")
-                self.logger.info(
+                self.logger.info(f"    Single years (is_year_range=False): {yr.get(False, 0):,}")
-                    f"    Single years (is_year_range=False): {yr.get(False, 0):,}"
+                self.logger.info(f"    Year ranges  (is_year_range=True):  {yr.get(True,  0):,}")
-                )
-                self.logger.info(
-                    f"    Year ranges  (is_year_range=True):  {yr.get(True,  0):,}"
-                )
-
-            # Validasi kolom framework tersedia
-            if 'framework' not in self.df_clean.columns:
-                raise ValueError(
-                    "Kolom 'framework' tidak ditemukan di dim_indicator. "
-                    "Pastikan bigquery_cleaned_layer.py dan bigquery_dimensional_model.py "
-                    "sudah dijalankan dengan versi terbaru."
-                )
-
-            fw_dist = self.df_clean.drop_duplicates('indicator_id')['framework'].value_counts()
-            self.logger.info(f"  Framework distribution (per indikator unik):")
-            for fw, cnt in fw_dist.items():
-                self.logger.info(f"    {fw}: {cnt} indicators")
 
             self.df_indicator = read_from_bigquery(self.client, 'dim_indicator', layer='gold')
             self.df_country   = read_from_bigquery(self.client, 'dim_country',   layer='gold')
@@ -265,15 +404,10 @@ class AnalyticalLayerLoader:
         self.logger.info(f"\n  [+] Valid:   {len(valid_combinations):,}")
         self.logger.info(f"  [-] Removed: {len(removed_combinations):,}")
 
-        df_valid        = pd.DataFrame(valid_combinations)
+        df_valid       = pd.DataFrame(valid_combinations)
-        df_valid['key'] = (
+        df_valid['key'] = df_valid['country_id'].astype(str) + '_' + df_valid['indicator_id'].astype(str)
-            df_valid['country_id'].astype(str) + '_' +
+        self.df_clean['key'] = (self.df_clean['country_id'].astype(str) + '_' +
-            df_valid['indicator_id'].astype(str)
+                                self.df_clean['indicator_id'].astype(str))
-        )
-        self.df_clean['key'] = (
-            self.df_clean['country_id'].astype(str) + '_' +
-            self.df_clean['indicator_id'].astype(str)
-        )
 
         original_count = len(self.df_clean)
         self.df_clean  = self.df_clean[self.df_clean['key'].isin(df_valid['key'])].copy()
@@ -307,17 +441,13 @@ class AnalyticalLayerLoader:
                 f"{row['pillar_count']}/{total_pillars} pillars"
             )
 
-        selected_countries        = country_pillar_count[
+        selected_countries        = country_pillar_count[country_pillar_count['pillar_count'] == total_pillars]
-            country_pillar_count['pillar_count'] == total_pillars
-        ]
         self.selected_country_ids = selected_countries['country_id'].tolist()
 
         self.logger.info(f"\n  FIXED SET: {len(self.selected_country_ids)} countries")
 
         original_count = len(self.df_clean)
-        self.df_clean  = self.df_clean[
+        self.df_clean  = self.df_clean[self.df_clean['country_id'].isin(self.selected_country_ids)].copy()
-            self.df_clean['country_id'].isin(self.selected_country_ids)
-        ].copy()
 
         self.logger.info(f"  Rows before: {original_count:,}")
         self.logger.info(f"  Rows after:  {len(self.df_clean):,}")
@@ -331,9 +461,7 @@ class AnalyticalLayerLoader:
         indicator_country_start = self.df_clean.groupby([
             'indicator_id', 'indicator_name', 'country_id'
         ])['year'].min().reset_index()
-        indicator_country_start.columns = [
+        indicator_country_start.columns = ['indicator_id', 'indicator_name', 'country_id', 'start_year']
-            'indicator_id', 'indicator_name', 'country_id', 'start_year'
-        ]
 
         indicator_max_start = indicator_country_start.groupby([
             'indicator_id', 'indicator_name'
@@ -372,9 +500,7 @@ class AnalyticalLayerLoader:
             else:
                 removed_indicators.append({
                     'indicator_name': indicator_name,
-                    'reason'        : (
+                    'reason'        : f"missing countries in years: {', '.join(problematic_years[:5])}"
-                        f"missing countries in years: {', '.join(problematic_years[:5])}"
-                    )
                 })
 
         self.logger.info(f"\n  [+] Valid:   {len(valid_indicators)}")
@@ -384,17 +510,12 @@ class AnalyticalLayerLoader:
             raise ValueError("No valid indicators found after filtering!")
 
         original_count = len(self.df_clean)
-        self.df_clean  = self.df_clean[
+        self.df_clean  = self.df_clean[self.df_clean['indicator_id'].isin(valid_indicators)].copy()
-            self.df_clean['indicator_id'].isin(valid_indicators)
-        ].copy()
 
         self.df_clean = self.df_clean.merge(
-            indicator_max_start[['indicator_id', 'max_start_year']],
+            indicator_max_start[['indicator_id', 'max_start_year']], on='indicator_id', how='left'
-            on='indicator_id', how='left'
         )
-        self.df_clean = self.df_clean[
+        self.df_clean = self.df_clean[self.df_clean['year'] >= self.df_clean['max_start_year']].copy()
-            self.df_clean['year'] >= self.df_clean['max_start_year']
-        ].copy()
         self.df_clean = self.df_clean.drop('max_start_year', axis=1)
 
         self.logger.info(f"\n  Rows before: {original_count:,}")
@@ -410,16 +531,12 @@ class AnalyticalLayerLoader:
         self.logger.info("=" * 80)
 
         expected_countries = len(self.selected_country_ids)
-        verification       = self.df_clean.groupby(
+        verification       = self.df_clean.groupby(['indicator_id', 'year'])['country_id'].nunique().reset_index()
-            ['indicator_id', 'year']
-        )['country_id'].nunique().reset_index()
         verification.columns = ['indicator_id', 'year', 'country_count']
         all_good = (verification['country_count'] == expected_countries).all()
 
         if all_good:
-            self.logger.info(
+            self.logger.info(f"  VERIFICATION PASSED — all combinations have {expected_countries} countries")
-                f"  VERIFICATION PASSED — all combinations have {expected_countries} countries"
-            )
         else:
             bad = verification[verification['country_count'] != expected_countries]
             for _, row in bad.head(10).iterrows():
@@ -431,101 +548,6 @@ class AnalyticalLayerLoader:
 
         return True
 
-    def calculate_yoy(self):
-        """
-        Hitung Year-over-Year (YoY) per indikator per negara.
-
-        Kolom yang ditambahkan ke df_clean:
-            yoy_change : selisih absolut  -> value - value_tahun_sebelumnya
-            yoy_pct    : perubahan relatif -> (yoy_change / abs(value_prev)) * 100
-
-        Catatan:
-        - Baris tahun pertama per kombinasi country-indicator akan bernilai NULL
-          (tidak ada tahun sebelumnya sebagai pembanding) — ini intentional.
-        - value_prev di-drop setelah kalkulasi, tidak ikut disimpan ke BigQuery.
-        - Dilakukan SETELAH verify_no_gaps() agar data sudah clean dan sorted benar.
-        """
-        self.logger.info("\n" + "=" * 80)
-        self.logger.info("STEP 6b: CALCULATE YEAR-OVER-YEAR (YoY) PER INDICATOR PER COUNTRY")
-        self.logger.info("=" * 80)
-
-        df = self.df_clean.sort_values(['country_id', 'indicator_id', 'year']).copy()
-
-        # Nilai tahun sebelumnya (shifted within each country-indicator group)
-        df['value_prev'] = df.groupby(['country_id', 'indicator_id'])['value'].shift(1)
-
-        # YoY absolute change: value(t) - value(t-1)
-        df['yoy_change'] = df['value'] - df['value_prev']
-
-        # YoY percentage change: (yoy_change / |value_prev|) * 100
-        # Hindari division by zero — jika value_prev == 0 atau NaN, hasilnya NaN
-        df['yoy_pct'] = np.where(
-            df['value_prev'].notna() & (df['value_prev'] != 0),
-            (df['yoy_change'] / df['value_prev'].abs()) * 100,
-            np.nan
-        )
-
-        # Drop kolom bantu value_prev, tidak ikut disimpan ke BigQuery
-        df = df.drop(columns=['value_prev'])
-
-        # Log ringkasan
-        total_rows   = len(df)
-        valid_yoy    = df['yoy_pct'].notna().sum()
-        null_yoy     = df['yoy_pct'].isna().sum()
-
-        self.logger.info(f"  Total rows        : {total_rows:,}")
-        self.logger.info(f"  YoY calculated    : {valid_yoy:,}")
-        self.logger.info(f"  YoY NULL (base yr): {null_yoy:,}  <- tahun pertama per country-indicator")
-
-        # Log distribusi YoY per indikator (sample)
-        per_ind = (
-            df[df['yoy_pct'].notna()]
-            .groupby(['indicator_id', 'indicator_name'])['yoy_pct']
-            .agg(['mean', 'std', 'min', 'max'])
-            .reset_index()
-        )
-        per_ind.columns = ['indicator_id', 'indicator_name', 'mean', 'std', 'min', 'max']
-
-        self.logger.info(f"\n  YoY summary per indicator (top 10 by abs mean change):")
-        self.logger.info(f"  {'-'*100}")
-        self.logger.info(
-            f"  {'ID':<5} {'Indicator Name':<52} {'Mean%':>8} {'Std%':>8} {'Min%':>8} {'Max%':>8}"
-        )
-        self.logger.info(f"  {'-'*100}")
-
-        top_ind = per_ind.reindex(
-            per_ind['mean'].abs().sort_values(ascending=False).index
-        ).head(10)
-
-        for _, row in top_ind.iterrows():
-            self.logger.info(
-                f"  {int(row['indicator_id']):<5} {row['indicator_name'][:50]:<52} "
-                f"{row['mean']:>+8.2f} {row['std']:>8.2f} "
-                f"{row['min']:>+8.2f} {row['max']:>+8.2f}"
-            )
-
-        # Log distribusi YoY per negara (ringkasan)
-        per_country = (
-            df[df['yoy_pct'].notna()]
-            .groupby(['country_id', 'country_name'])['yoy_pct']
-            .agg(['mean', 'std'])
-            .reset_index()
-        )
-        per_country.columns = ['country_id', 'country_name', 'mean_yoy', 'std_yoy']
-
-        self.logger.info(f"\n  YoY summary per country:")
-        self.logger.info(f"  {'-'*60}")
-        self.logger.info(f"  {'Country':<30} {'Mean YoY%':>10} {'Std YoY%':>10}")
-        self.logger.info(f"  {'-'*60}")
-        for _, row in per_country.sort_values('mean_yoy', ascending=False).iterrows():
-            self.logger.info(
-                f"  {row['country_name']:<30} {row['mean_yoy']:>+10.2f} {row['std_yoy']:>10.2f}"
-            )
-
-        self.df_clean = df
-        self.logger.info(f"\n  [OK] YoY columns added: yoy_change, yoy_pct")
-        return self.df_clean
-
     def analyze_indicator_availability_by_year(self):
         self.logger.info("\n" + "=" * 80)
         self.logger.info("STEP 7: ANALYZE INDICATOR AVAILABILITY BY YEAR")
@@ -548,62 +570,36 @@ class AnalyticalLayerLoader:
             )
 
         indicator_details = self.df_clean.groupby([
-            'indicator_id', 'indicator_name', 'pillar_name', 'direction', 'framework'
+            'indicator_id', 'indicator_name', 'pillar_name', 'direction'
         ]).agg({'year': ['min', 'max'], 'country_id': 'nunique'}).reset_index()
         indicator_details.columns = [
-            'indicator_id', 'indicator_name', 'pillar_name', 'direction', 'framework',
+            'indicator_id', 'indicator_name', 'pillar_name', 'direction',
             'start_year', 'end_year', 'country_count'
         ]
         indicator_details['year_range'] = (
             indicator_details['start_year'].astype(int).astype(str) + '-' +
             indicator_details['end_year'].astype(int).astype(str)
         )
-        indicator_details = indicator_details.sort_values(
+        indicator_details = indicator_details.sort_values(['pillar_name', 'start_year', 'indicator_name'])
-            ['framework', 'pillar_name', 'start_year', 'indicator_name']
-        )
 
         self.logger.info(f"\nTotal Indicators: {len(indicator_details)}")
         for pillar, count in indicator_details.groupby('pillar_name').size().items():
             self.logger.info(f"  {pillar}: {count} indicators")
 
-        self.logger.info(f"\nFramework breakdown:")
+        self.logger.info(f"\n{'-'*100}")
-        for fw, count in indicator_details.groupby('framework').size().items():
+        self.logger.info(f"{'ID':<5} {'Indicator Name':<55} {'Pillar':<15} {'Years':<12} {'Dir':<8} {'Countries'}")
-            self.logger.info(f"  {fw}: {count} indicators")
+        self.logger.info(f"{'-'*100}")
-
-        self.logger.info(f"\n{'-'*110}")
-        self.logger.info(
-            f"{'ID':<5} {'Indicator Name':<55} {'Pillar':<15} "
-            f"{'Framework':<10} {'Years':<12} {'Dir':<8} {'Countries'}"
-        )
-        self.logger.info(f"{'-'*110}")
         for _, row in indicator_details.iterrows():
             direction = 'higher+' if row['direction'] == 'higher_better' else 'lower-'
             self.logger.info(
                 f"{int(row['indicator_id']):<5} {row['indicator_name'][:52]:<55} "
-                f"{row['pillar_name'][:13]:<15} {row['framework']:<10} "
+                f"{row['pillar_name'][:13]:<15} {row['year_range']:<12} "
-                f"{row['year_range']:<12} {direction:<8} {int(row['country_count'])}"
+                f"{direction:<8} {int(row['country_count'])}"
             )
 
         return year_stats
 
     def save_analytical_table(self):
-        """
-        Simpan fact_asean_food_security_selected ke Gold layer.
-
-        Kolom yang disimpan:
-            country_id, country_name     — dimensi negara
-            indicator_id, indicator_name — dimensi indikator
-            direction                    — arah penilaian (higher/lower_better)
-            framework                    — MDGs / SDGs  (untuk filter Looker Studio)
-            pillar_id, pillar_name       — dimensi pilar
-            time_id, year                — dimensi waktu
-            value                        — nilai indikator
-            yoy_change                   — perubahan absolut YoY (NULLABLE: NULL di tahun pertama)
-            yoy_pct                      — perubahan relatif YoY dalam % (NULLABLE: NULL di tahun pertama)
-
-        Kolom framework memungkinkan filter langsung di Looker Studio tanpa join ke dim_indicator.
-        Kolom yoy_change dan yoy_pct memungkinkan analisis tren tahunan langsung di Looker Studio.
-        """
         table_name = 'fact_asean_food_security_selected'
 
         self.logger.info("\n" + "=" * 80)
@@ -611,100 +607,81 @@ class AnalyticalLayerLoader:
         self.logger.info("=" * 80)
 
         try:
-            # Pastikan kolom framework tersedia di df_clean
-            if 'framework' not in self.df_clean.columns:
-                self.logger.warning(
-                    "  [WARN] Kolom 'framework' tidak ada di df_clean. "
-                    "Melakukan join ke dim_indicator sebagai fallback..."
-                )
-                dim_ind = read_from_bigquery(self.client, 'dim_indicator', layer='gold')
-                if 'framework' in dim_ind.columns:
-                    self.df_clean = self.df_clean.merge(
-                        dim_ind[['indicator_id', 'framework']],
-                        on='indicator_id', how='left'
-                    )
-                    self.df_clean['framework'] = self.df_clean['framework'].fillna('MDGs')
-                    self.logger.info("  [OK] framework di-join dari dim_indicator")
-                else:
-                    self.df_clean['framework'] = 'MDGs'
-                    self.logger.warning(
-                        "  [WARN] dim_indicator juga tidak punya kolom framework. "
-                        "Default: MDGs. Jalankan ulang pipeline dari cleaned_layer."
-                    )
-
-            # Pastikan kolom YoY tersedia — fallback jika calculate_yoy() tidak dipanggil
-            if 'yoy_change' not in self.df_clean.columns or 'yoy_pct' not in self.df_clean.columns:
-                self.logger.warning(
-                    "  [WARN] Kolom YoY tidak ditemukan. Menjalankan calculate_yoy() sebagai fallback..."
-                )
-                self.calculate_yoy()
-
             analytical_df = self.df_clean[[
                 'country_id',
                 'country_name',
                 'indicator_id',
                 'indicator_name',
                 'direction',
-                'framework',
                 'pillar_id',
                 'pillar_name',
                 'time_id',
                 'year',
                 'value',
-                'yoy_change',
-                'yoy_pct',
             ]].copy()
 
+            # ------------------------------------------------------------------
+            # TAMBAHAN: kolom terjemahan Bahasa Indonesia
+            # indicator_name_id : terjemahan Bahasa Indonesia dari indicator_name
+            # pillar_name_id    : terjemahan Bahasa Indonesia dari pillar_name
+            # ------------------------------------------------------------------
+            analytical_df['indicator_name_id'] = analytical_df['indicator_name'].apply(translate_indicator)
+            analytical_df['pillar_name_id']    = analytical_df['pillar_name'].apply(translate_pillar)
+
+            # Log indikator yang belum punya terjemahan (fallback ke nama asli)
+            no_trans_ind = analytical_df[
+                analytical_df['indicator_name_id'] == analytical_df['indicator_name']
+            ]['indicator_name'].unique()
+            if len(no_trans_ind) > 0:
+                self.logger.warning(
+                    f"  [TRANSLATION] {len(no_trans_ind)} indicator(s) tidak ada di kamus "
+                    f"(menggunakan nama asli): {list(no_trans_ind)[:5]}"
+                )
+
+            no_trans_pil = analytical_df[
+                analytical_df['pillar_name_id'] == analytical_df['pillar_name']
+            ]['pillar_name'].unique()
+            if len(no_trans_pil) > 0:
+                self.logger.warning(
+                    f"  [TRANSLATION] {len(no_trans_pil)} pillar(s) tidak ada di kamus "
+                    f"(menggunakan nama asli): {list(no_trans_pil)}"
+                )
+
             analytical_df = analytical_df.sort_values(
                 ['year', 'country_name', 'pillar_name', 'indicator_name']
             ).reset_index(drop=True)
 
             # Pastikan tipe data konsisten
-            analytical_df['country_id']    = analytical_df['country_id'].astype(int)
+            analytical_df['country_id']        = analytical_df['country_id'].astype(int)
-            analytical_df['country_name']  = analytical_df['country_name'].astype(str)
+            analytical_df['country_name']      = analytical_df['country_name'].astype(str)
-            analytical_df['indicator_id']  = analytical_df['indicator_id'].astype(int)
+            analytical_df['indicator_id']      = analytical_df['indicator_id'].astype(int)
-            analytical_df['indicator_name']= analytical_df['indicator_name'].astype(str)
+            analytical_df['indicator_name']    = analytical_df['indicator_name'].astype(str)
-            analytical_df['direction']     = analytical_df['direction'].astype(str)
+            analytical_df['indicator_name_id'] = analytical_df['indicator_name_id'].astype(str)
-            analytical_df['framework']     = analytical_df['framework'].astype(str)
+            analytical_df['direction']         = analytical_df['direction'].astype(str)
-            analytical_df['pillar_id']     = analytical_df['pillar_id'].astype(int)
+            analytical_df['pillar_id']         = analytical_df['pillar_id'].astype(int)
-            analytical_df['pillar_name']   = analytical_df['pillar_name'].astype(str)
+            analytical_df['pillar_name']       = analytical_df['pillar_name'].astype(str)
-            analytical_df['time_id']       = analytical_df['time_id'].astype(int)
+            analytical_df['pillar_name_id']    = analytical_df['pillar_name_id'].astype(str)
-            analytical_df['year']          = analytical_df['year'].astype(int)
+            analytical_df['time_id']           = analytical_df['time_id'].astype(int)
-            analytical_df['value']         = analytical_df['value'].astype(float)
+            analytical_df['year']              = analytical_df['year'].astype(int)
-            # yoy_change dan yoy_pct tetap float — NULL (NaN) di tahun pertama adalah intentional
+            analytical_df['value']             = analytical_df['value'].astype(float)
-            analytical_df['yoy_change']    = analytical_df['yoy_change'].astype(float)
-            analytical_df['yoy_pct']       = analytical_df['yoy_pct'].astype(float)
 
             self.logger.info(f"  Kolom yang disimpan: {list(analytical_df.columns)}")
             self.logger.info(f"  Total rows: {len(analytical_df):,}")
 
-            # Log distribusi framework
+            # Schema BigQuery
-            fw_dist = analytical_df.drop_duplicates('indicator_id')['framework'].value_counts()
-            self.logger.info(f"  Framework distribution (per indikator unik):")
-            for fw, cnt in fw_dist.items():
-                self.logger.info(f"    {fw}: {cnt} indicators")
-
-            # Log statistik YoY
-            yoy_valid = analytical_df['yoy_pct'].notna().sum()
-            yoy_null  = analytical_df['yoy_pct'].isna().sum()
-            self.logger.info(f"  YoY rows (calculated): {yoy_valid:,}")
-            self.logger.info(f"  YoY rows (NULL/base) : {yoy_null:,}")
-
             schema = [
-                bigquery.SchemaField("country_id",     "INTEGER", mode="REQUIRED"),
+                bigquery.SchemaField("country_id",        "INTEGER", mode="REQUIRED"),
-                bigquery.SchemaField("country_name",   "STRING",  mode="REQUIRED"),
+                bigquery.SchemaField("country_name",      "STRING",  mode="REQUIRED"),
-                bigquery.SchemaField("indicator_id",   "INTEGER", mode="REQUIRED"),
+                bigquery.SchemaField("indicator_id",      "INTEGER", mode="REQUIRED"),
-                bigquery.SchemaField("indicator_name", "STRING",  mode="REQUIRED"),
+                bigquery.SchemaField("indicator_name",    "STRING",  mode="REQUIRED"),
-                bigquery.SchemaField("direction",      "STRING",  mode="REQUIRED"),
+                bigquery.SchemaField("indicator_name_id", "STRING",  mode="REQUIRED"),
-                bigquery.SchemaField("framework",      "STRING",  mode="REQUIRED"),
+                bigquery.SchemaField("direction",         "STRING",  mode="REQUIRED"),
-                bigquery.SchemaField("pillar_id",      "INTEGER", mode="REQUIRED"),
+                bigquery.SchemaField("pillar_id",         "INTEGER", mode="REQUIRED"),
-                bigquery.SchemaField("pillar_name",    "STRING",  mode="REQUIRED"),
+                bigquery.SchemaField("pillar_name",       "STRING",  mode="REQUIRED"),
-                bigquery.SchemaField("time_id",        "INTEGER", mode="REQUIRED"),
+                bigquery.SchemaField("pillar_name_id",    "STRING",  mode="REQUIRED"),
-                bigquery.SchemaField("year",           "INTEGER", mode="REQUIRED"),
+                bigquery.SchemaField("time_id",           "INTEGER", mode="REQUIRED"),
-                bigquery.SchemaField("value",          "FLOAT",   mode="REQUIRED"),
+                bigquery.SchemaField("year",              "INTEGER", mode="REQUIRED"),
-                # NULLABLE karena tahun pertama per country-indicator tidak memiliki nilai sebelumnya
+                bigquery.SchemaField("value",             "FLOAT",   mode="REQUIRED"),
-                bigquery.SchemaField("yoy_change",     "FLOAT",   mode="NULLABLE"),
-                bigquery.SchemaField("yoy_pct",        "FLOAT",   mode="NULLABLE"),
             ]
 
             rows_loaded = load_to_bigquery(
@@ -730,24 +707,16 @@ class AnalyticalLayerLoader:
                     'fixed_countries': len(self.selected_country_ids),
                     'no_gaps'        : True,
                     'layer'          : 'gold',
-                    'columns'        : (
+                    'columns'        : 'id + name + name_id (Looker Studio ready)'
-                        'id + name + direction + framework + value + '
-                        'yoy_change + yoy_pct (Looker Studio ready)'
-                    )
                 }),
                 'validation_metrics' : json.dumps({
                     'fixed_countries' : len(self.selected_country_ids),
-                    'total_indicators': int(self.df_clean['indicator_id'].nunique()),
+                    'total_indicators': int(self.df_clean['indicator_id'].nunique())
-                    'framework_dist'  : fw_dist.to_dict(),
-                    'yoy_rows_valid'  : int(yoy_valid),
-                    'yoy_rows_null'   : int(yoy_null),
                 })
             }
             save_etl_metadata(self.client, metadata)
 
-            self.logger.info(
+            self.logger.info(f"  [OK] {table_name}: {rows_loaded:,} rows -> [DW/Gold] fs_asean_gold")
-                f"  {table_name}: {rows_loaded:,} rows -> [DW/Gold] fs_asean_gold"
-            )
             self.logger.info(f"    Metadata -> [AUDIT] etl_metadata")
             return rows_loaded
 
@@ -761,8 +730,6 @@ class AnalyticalLayerLoader:
 
         self.logger.info("\n" + "=" * 80)
         self.logger.info("Output: fact_asean_food_security_selected -> fs_asean_gold")
-        self.logger.info("Kolom: country_id/name, indicator_id/name, direction, framework,")
-        self.logger.info("       pillar_id/name, time_id, year, value, yoy_change, yoy_pct")
         self.logger.info("=" * 80)
 
         self.load_source_data()
@@ -771,7 +738,6 @@ class AnalyticalLayerLoader:
         self.select_countries_with_all_pillars()
         self.filter_indicators_consistent_across_fixed_countries()
         self.verify_no_gaps()
-        self.calculate_yoy()                          # <-- Step 6b: hitung YoY
         self.analyze_indicator_availability_by_year()
         self.save_analytical_table()
 

scripts/bigquery_cleaned_layer.py

@@ -40,12 +40,12 @@ def load_staging_data(client: bigquery.Client) -> pd.DataFrame:
     """Load data dari staging_integrated (STAGING/Silver layer)."""
     print("\nLoading data from staging_integrated (fs_asean_silver)...")
     df_staging = read_from_bigquery(client, 'staging_integrated', layer='silver')
-    print(f"  Loaded        : {len(df_staging):,} rows")
+    print(f"  ✓ Loaded        : {len(df_staging):,} rows")
-    print(f"    Columns     : {len(df_staging.columns)}")
+    print(f"    Columns       : {len(df_staging.columns)}")
-    print(f"    Sources     : {df_staging['source'].nunique()}")
+    print(f"    Sources       : {df_staging['source'].nunique()}")
-    print(f"    Indicators  : {df_staging['indicator_standardized'].nunique()}")
+    print(f"    Indicators    : {df_staging['indicator_standardized'].nunique()}")
-    print(f"    Countries   : {df_staging['country'].nunique()}")
+    print(f"    Countries     : {df_staging['country'].nunique()}")
-    print(f"    Year range  : {int(df_staging['year'].min())}-{int(df_staging['year'].max())}")
+    print(f"    Year range    : {int(df_staging['year'].min())}-{int(df_staging['year'].max())}")
     return df_staging
 
 
@@ -53,6 +53,7 @@ def load_staging_data(client: bigquery.Client) -> pd.DataFrame:
 # COLUMN CONSTRAINT HELPERS
 # =============================================================================
 
+# Schema constraints — semua varchar max lengths
 COLUMN_CONSTRAINTS = {
     'source'                : 20,
     'indicator_original'    : 255,
@@ -61,8 +62,7 @@ COLUMN_CONSTRAINTS = {
     'year_range'            : 20,
     'unit'                  : 20,
     'pillar'                : 20,
-    'direction'             : 15,
+    'direction'             : 15,   # 'higher_better'=13, 'lower_better'=12
-    'framework'             : 5,   # 'MDGs'=4, 'SDGs'=4
 }
 
 
@@ -101,11 +101,11 @@ def apply_column_constraints(df: pd.DataFrame) -> pd.DataFrame:
             )
 
     if truncation_report:
-        print("\n  Column Truncations Applied:")
+        print("\n  ⚠ Column Truncations Applied:")
         for column, info in truncation_report.items():
             print(f"    - {column}: {info['count']} values truncated to {info['max_length']} chars")
     else:
-        print("\n  No truncations needed — all values within constraints")
+        print("\n  ✓ No truncations needed — all values within constraints")
 
     return df_constrained
 
@@ -156,11 +156,11 @@ def standardize_country_names_asean(df: pd.DataFrame, country_column: str = 'cou
     def map_country(country):
         if pd.isna(country):
             return country
-        s      = str(country).strip()
+        s = str(country).strip()
         mapped = ASEAN_MAPPING.get(s.upper(), s)
         return mapped[:100] if len(mapped) > 100 else mapped
 
-    original                = df_clean[country_column].copy()
+    original               = df_clean[country_column].copy()
     df_clean[country_column] = df_clean[country_column].apply(map_country)
     changes = {orig: new for orig, new in zip(original, df_clean[country_column]) if orig != new}
 
@@ -177,16 +177,16 @@ def standardize_country_names_asean(df: pd.DataFrame, country_column: str = 'cou
 def assign_pillar(indicator_name: str) -> str:
     """
     Assign pillar berdasarkan keyword indikator.
-    Return values: 'Availability', 'Access', 'Utilization', 'Stability', 'Other'
+    Return values: 'Availability', 'Access', 'Utilization', 'Stability', 'Sustainability'
-    All <= 20 chars (varchar(20) constraint).
+    All ≤ 20 chars (varchar(20) constraint).
     """
     if pd.isna(indicator_name):
-        return 'Other'
+        return 'Sustainability'
     ind = str(indicator_name).lower()
 
     for kw in ['requirement', 'coefficient', 'losses', 'fat supply']:
         if kw in ind:
-            return 'Other'
+            return 'Sustainability'
 
     if any(kw in ind for kw in [
         'adequacy', 'protein supply', 'supply of protein',
@@ -210,13 +210,12 @@ def assign_pillar(indicator_name: str) -> str:
 
     if any(kw in ind for kw in [
         'wasting', 'wasted', 'stunted', 'overweight', 'obese', 'obesity',
-        'anemia', 'anaemia', 'birthweight', 'breastfeeding', 'drinking water',
+        'anemia', 'birthweight', 'breastfeeding', 'drinking water', 'sanitation',
-        'sanitation', 'children under 5', 'newborns with low',
+        'children under 5', 'newborns with low', 'women of reproductive'
-        'women of reproductive'
     ]):
         return 'Utilization'
 
-    return 'Other'
+    return 'Sustainability'
 
 
 # =============================================================================
@@ -227,15 +226,17 @@ def assign_direction(indicator_name: str) -> str:
     """
     Assign direction berdasarkan indikator.
     Return values: 'higher_better' (13 chars) atau 'lower_better' (12 chars)
-    Both <= 15 chars (varchar(15) constraint).
+    Both ≤ 15 chars (varchar(15) constraint).
     """
     if pd.isna(indicator_name):
         return 'higher_better'
     ind = str(indicator_name).lower()
 
+    # Spesifik lower_better
     if 'share of dietary energy supply derived from cereals' in ind:
         return 'lower_better'
 
+    # Higher_better exceptions — cek sebelum lower_better keywords
     for kw in [
         'exclusive breastfeeding',
         'dietary energy supply',
@@ -247,6 +248,7 @@ def assign_direction(indicator_name: str) -> str:
         if kw in ind:
             return 'higher_better'
 
+    # Lower_better — masalah yang harus diminimalkan
     for kw in [
         'prevalence of undernourishment',
         'prevalence of severe food insecurity',
@@ -257,7 +259,6 @@ def assign_direction(indicator_name: str) -> str:
         'prevalence of overweight',
         'prevalence of obesity',
         'prevalence of anemia',
-        'prevalence of anaemia',
         'prevalence of low birthweight',
         'number of people undernourished',
         'number of severely food insecure',
@@ -282,9 +283,6 @@ def assign_direction(indicator_name: str) -> str:
         'coefficient of variation',
         'incidence of caloric losses',
         'food losses',
-        'indicator of food price anomalies',
-        'proportion of local breeds classified as being at risk',
-        'agricultural export subsidies',
     ]:
         if kw in ind:
             return 'lower_better'
@@ -292,62 +290,6 @@ def assign_direction(indicator_name: str) -> str:
     return 'higher_better'
 
 
-# =============================================================================
-# FRAMEWORK CLASSIFICATION (MDGs vs SDGs)
-# =============================================================================
-
-# Daftar keyword eksplisit dari SDG Goal 2 Khusus FIES (2030 Agenda for Sustainable Development).
-# Disimpan lowercase agar matching tidak sensitif terhadap kapitalisasi input.
-
-SDG_INDICATOR_KEYWORDS = frozenset([
-    "prevalence of severe food insecurity in the total population (percent) (3-year average)",
-    "prevalence of severe food insecurity in the male adult population (percent) (3-year average)",
-    "prevalence of severe food insecurity in the female adult population (percent) (3-year average)",
-    "prevalence of moderate or severe food insecurity in the total population (percent) (3-year average)",
-    "prevalence of moderate or severe food insecurity in the male adult population (percent) (3-year average)",
-    "prevalence of moderate or severe food insecurity in the female adult population (percent) (3-year average)",
-    "number of severely food insecure people (million) (3-year average)",
-    "number of severely food insecure male adults (million) (3-year average)",
-    "number of severely food insecure female adults (million) (3-year average)",
-    "number of moderately or severely food insecure people (million) (3-year average)",
-    "number of moderately or severely food insecure male adults (million) (3-year average)",
-    "number of moderately or severely food insecure female adults (million) (3-year average)",
-])
-
-
-def assign_framework(indicator_name: str) -> str:
-    """
-    Assign framework berdasarkan daftar eksplisit indikator SDG Goal 2
-    dari 2030 Agenda for Sustainable Development (versi Maret 2020).
-
-    Logika:
-    - Lowercase nama indikator input
-    - Cek apakah ada keyword SDG (lowercase) yang terkandung di dalam nama indikator
-    - Jika ya  -> 'SDGs'
-    - Jika tidak -> 'MDGs' (indikator FAO/lama yang bukan SDG resmi)
-
-    FIX: Bug sebelumnya menggunakan `kw in ind` (cek apakah keyword ada di dalam ind),
-    padahal seharusnya `kw in ind` sudah benar secara logika — tapi keyword di-set
-    dengan kapitalisasi campuran sementara `ind` sudah di-lowercase, sehingga
-    perbandingan selalu gagal. Solusi: simpan keyword dalam lowercase di set,
-    sehingga `kw in ind` bekerja dengan benar.
-
-    Return values: 'MDGs' atau 'SDGs'
-    Panjang max 4 chars (dalam constraint varchar(5)).
-    """
-    if pd.isna(indicator_name):
-        return 'MDGs'
-
-    # Lowercase input agar matching tidak sensitif terhadap kapitalisasi
-    ind = str(indicator_name).lower().strip()
-
-    # Cek apakah salah satu keyword SDG (sudah lowercase) ada di dalam ind
-    if any(kw in ind for kw in SDG_INDICATOR_KEYWORDS):
-        return 'SDGs'
-
-    return 'MDGs'
-
-
 # =============================================================================
 # CLEANED DATA LOADER
 # =============================================================================
@@ -357,18 +299,19 @@ class CleanedDataLoader:
     Loader untuk cleaned integrated data ke STAGING layer (Silver).
 
     Kimball context:
-        Input  : staging_integrated  -> STAGING (Silver) — fs_asean_silver
+        Input  : staging_integrated  → STAGING (Silver) — fs_asean_silver
-        Output : cleaned_integrated  -> STAGING (Silver) — fs_asean_silver
+        Output : cleaned_integrated  → STAGING (Silver) — fs_asean_silver
-        Audit  : etl_logs, etl_metadata -> AUDIT — fs_asean_audit
+        Audit  : etl_logs, etl_metadata → AUDIT — fs_asean_audit
 
     Pipeline steps:
         1. Standardize country names (ASEAN)
         2. Remove missing values
         3. Remove duplicates
-        4. Add pillar, direction & framework classification
+        4. Add pillar classification
-        5. Apply column constraints
+        5. Add direction classification
-        6. Load ke BigQuery
+        6. Apply column constraints
-        7. Log ke Audit layer
+        7. Load ke BigQuery
+        8. Log ke Audit layer
     """
 
     SCHEMA = [
@@ -382,7 +325,6 @@ class CleanedDataLoader:
         bigquery.SchemaField("unit",                   "STRING",  mode="NULLABLE"),
         bigquery.SchemaField("pillar",                 "STRING",  mode="REQUIRED"),
         bigquery.SchemaField("direction",              "STRING",  mode="REQUIRED"),
-        bigquery.SchemaField("framework",              "STRING",  mode="REQUIRED"),
     ]
 
     def __init__(self, client: bigquery.Client, load_mode: str = 'full_refresh'):
@@ -413,7 +355,7 @@ class CleanedDataLoader:
     def _step_standardize_countries(self, df: pd.DataFrame) -> pd.DataFrame:
         print("\n  [Step 1/5] Standardize country names...")
         df, report = standardize_country_names_asean(df, country_column='country')
-        print(f"    ASEAN countries mapped : {report['countries_mapped']}")
+        print(f"    ✓ ASEAN countries mapped : {report['countries_mapped']}")
         unique_countries = sorted(df['country'].unique())
         print(f"    Countries ({len(unique_countries)}) : {', '.join(unique_countries)}")
         log_update(self.client, 'STAGING', 'staging_integrated',
@@ -435,9 +377,7 @@ class CleanedDataLoader:
     def _step_remove_duplicates(self, df: pd.DataFrame) -> pd.DataFrame:
         print("\n  [Step 3/5] Remove duplicates...")
         exact_dups = df.duplicated().sum()
-        data_dups  = df.duplicated(
+        data_dups  = df.duplicated(subset=['indicator_standardized', 'country', 'year', 'value']).sum()
-            subset=['indicator_standardized', 'country', 'year', 'value']
-        ).sum()
         print(f"    Exact duplicates : {exact_dups:,}")
         print(f"    Data duplicates  : {data_dups:,}")
         rows_before = len(df)
@@ -449,39 +389,21 @@ class CleanedDataLoader:
         return df_clean
 
     def _step_add_classifications(self, df: pd.DataFrame) -> pd.DataFrame:
-        print("\n  [Step 4/5] Add pillar, direction & framework classification...")
+        print("\n  [Step 4/5] Add pillar & direction classification...")
         df = df.copy()
-
         df['pillar']    = df['indicator_standardized'].apply(assign_pillar)
         df['direction'] = df['indicator_standardized'].apply(assign_direction)
-        df['framework'] = df['indicator_standardized'].apply(assign_framework)
 
         pillar_counts = df['pillar'].value_counts()
-        print(f"    Pillar distribution:")
+        print(f"    ✓ Pillar distribution:")
         for pillar, count in pillar_counts.items():
             print(f"      - {pillar}: {count:,}")
 
         direction_counts = df['direction'].value_counts()
-        print(f"    Direction distribution:")
+        print(f"    ✓ Direction distribution:")
         for direction, count in direction_counts.items():
             pct = count / len(df) * 100
             print(f"      - {direction}: {count:,} ({pct:.1f}%)")
-
-        framework_counts = df['framework'].value_counts()
-        print(f"    Framework distribution:")
-        for fw, count in framework_counts.items():
-            pct = count / len(df) * 100
-            print(f"      - {fw}: {count:,} ({pct:.1f}%)")
-
-        # Log indikator yang terklasifikasi SDGs untuk verifikasi
-        sdg_inds = (
-            df[df['framework'] == 'SDGs']['indicator_standardized']
-            .drop_duplicates().sort_values().tolist()
-        )
-        print(f"\n    SDG indicators ({len(sdg_inds)}):")
-        for ind in sdg_inds:
-            print(f"      - {ind}")
-
         return df
 
     def _step_apply_constraints(self, df: pd.DataFrame) -> pd.DataFrame:
@@ -506,7 +428,7 @@ class CleanedDataLoader:
                 'max'         : int(df['year'].max()) if not df['year'].isnull().all() else None,
                 'unique_years': int(df['year'].nunique())
             }
-        for col in ('pillar', 'direction', 'framework', 'source'):
+        for col in ('pillar', 'direction', 'source'):
             if col in df.columns:
                 validation[f'{col}_breakdown'] = {
                     str(k): int(v) for k, v in df[col].value_counts().to_dict().items()
@@ -516,6 +438,7 @@ class CleanedDataLoader:
         if 'country' in df.columns:
             validation['unique_countries'] = int(df['country'].nunique())
 
+        # Column length check
         column_length_check = {}
         for col, max_len in COLUMN_CONSTRAINTS.items():
             if col in df.columns:
@@ -534,7 +457,7 @@ class CleanedDataLoader:
 
     def run(self, df: pd.DataFrame) -> int:
         """
-        Execute full cleaning pipeline -> load ke STAGING (Silver).
+        Execute full cleaning pipeline → load ke STAGING (Silver).
 
         Returns:
             int: Rows loaded
@@ -546,6 +469,7 @@ class CleanedDataLoader:
             print("  ERROR: DataFrame is empty, nothing to process.")
             return 0
 
+        # Pipeline steps
         df = self._step_standardize_countries(df)
         df = self._step_remove_missing(df)
         df = self._step_remove_duplicates(df)
@@ -554,6 +478,7 @@ class CleanedDataLoader:
 
         self.metadata['rows_transformed'] = len(df)
 
+        # Validate
         validation = self.validate_data(df)
         self.metadata['validation_metrics'] = validation
 
@@ -562,12 +487,13 @@ class CleanedDataLoader:
             for info in validation.get('column_length_check', {}).values()
         )
         if not all_within_limits:
-            print("\n  WARNING: Some columns still exceed length constraints!")
+            print("\n  ⚠ WARNING: Some columns still exceed length constraints!")
             for col, info in validation['column_length_check'].items():
                 if not info['within_limit']:
                     print(f"    - {col}: {info['max_actual_length']} > {info['max_length_constraint']}")
 
-        print(f"\n  Loading to [STAGING/Silver] {self.table_name} -> fs_asean_silver...")
+        # Load ke Silver
+        print(f"\n  Loading to [STAGING/Silver] {self.table_name} → fs_asean_silver...")
         rows_loaded = load_to_bigquery(
             self.client, df, self.table_name,
             layer='silver',
@@ -576,8 +502,10 @@ class CleanedDataLoader:
         )
         self.metadata['rows_loaded'] = rows_loaded
 
+        # Audit logs
         log_update(self.client, 'STAGING', self.table_name, 'full_refresh', rows_loaded)
 
+        # ETL metadata
         self.metadata['end_time']            = datetime.now()
         self.metadata['duration_seconds']    = (
             self.metadata['end_time'] - self.metadata['start_time']
@@ -588,31 +516,33 @@ class CleanedDataLoader:
         self.metadata['validation_metrics']  = json.dumps(validation)
         save_etl_metadata(self.client, self.metadata)
 
-        print(f"\n  Cleaned Integration completed: {rows_loaded:,} rows")
+        # Summary
+        print(f"\n  ✓ Cleaned Integration completed: {rows_loaded:,} rows")
         print(f"    Duration     : {self.metadata['duration_seconds']:.2f}s")
         print(f"    Completeness : {validation['completeness_pct']:.2f}%")
         if 'year_range' in validation:
             yr = validation['year_range']
             if yr['min'] and yr['max']:
-                print(f"    Year range   : {yr['min']}-{yr['max']}")
+                print(f"    Year range   : {yr['min']}–{yr['max']}")
         print(f"    Indicators   : {validation.get('unique_indicators', '-')}")
         print(f"    Countries    : {validation.get('unique_countries', '-')}")
         print(f"\n    Schema Validation:")
         for col, info in validation.get('column_length_check', {}).items():
-            status = "OK" if info['within_limit'] else "FAIL"
+            status = "✓" if info['within_limit'] else "✗"
-            print(f"    [{status}] {col}: {info['max_actual_length']}/{info['max_length_constraint']}")
+            print(f"    {status} {col}: {info['max_actual_length']}/{info['max_length_constraint']}")
-        print(f"\n    Metadata -> [AUDIT] etl_metadata")
+        print(f"\n    Metadata → [AUDIT] etl_metadata")
 
         return rows_loaded
 
 
 # =============================================================================
-# AIRFLOW TASK FUNCTIONS
+# AIRFLOW TASK FUNCTIONS  ← sama polanya dengan raw layer
 # =============================================================================
 
 def run_cleaned_integration():
     """
     Airflow task: Load cleaned_integrated dari staging_integrated.
+
     Dipanggil oleh DAG setelah task staging_integration_to_silver selesai.
     """
     from scripts.bigquery_config import get_bigquery_client
@@ -631,21 +561,21 @@ if __name__ == "__main__":
     print("=" * 60)
     print("BIGQUERY CLEANED LAYER ETL")
     print("Kimball DW Architecture")
-    print("  Input  : STAGING (Silver) -> staging_integrated")
+    print("  Input  : STAGING (Silver) → staging_integrated")
-    print("  Output : STAGING (Silver) -> cleaned_integrated")
+    print("  Output : STAGING (Silver) → cleaned_integrated")
-    print("  Audit  : AUDIT            -> etl_logs, etl_metadata")
+    print("  Audit  : AUDIT            → etl_logs, etl_metadata")
     print("=" * 60)
 
     logger     = setup_logging()
     client     = get_bigquery_client()
     df_staging = load_staging_data(client)
 
-    print("\n[1/1] Cleaned Integration -> STAGING (Silver)...")
+    print("\n[1/1] Cleaned Integration → STAGING (Silver)...")
     loader      = CleanedDataLoader(client, load_mode='full_refresh')
     final_count = loader.run(df_staging)
 
     print("\n" + "=" * 60)
-    print("[OK] CLEANED LAYER ETL COMPLETED")
+    print("✓ CLEANED LAYER ETL COMPLETED")
-    print(f"  STAGING (Silver) : cleaned_integrated ({final_count:,} rows)")
+    print(f"  🥈 STAGING (Silver) : cleaned_integrated ({final_count:,} rows)")
-    print(f"  AUDIT            : etl_logs, etl_metadata")
+    print(f"  📋 AUDIT            : etl_logs, etl_metadata")
     print("=" * 60)

scripts/bigquery_dimensional_model.py

@@ -46,13 +46,13 @@ class DimensionalModelLoader:
     Loader untuk dimensional model ke DW layer (Gold) — fs_asean_gold.
 
     Kimball context:
-        Input  : cleaned_integrated -> STAGING (Silver) — fs_asean_silver
+        Input  : cleaned_integrated → STAGING (Silver) — fs_asean_silver
-        Output : dim_* + fact_*     -> DW     (Gold)   — fs_asean_gold
+        Output : dim_* + fact_*     → DW     (Gold)   — fs_asean_gold
-        Audit  : etl_logs, etl_metadata -> AUDIT — fs_asean_audit
+        Audit  : etl_logs, etl_metadata → AUDIT — fs_asean_audit
 
     Pipeline steps:
         1. Load dim_country
-        2. Load dim_indicator  (+ kolom framework dari cleaned_integrated)
+        2. Load dim_indicator
         3. Load dim_time
         4. Load dim_source
         5. Load dim_pillar
@@ -117,7 +117,7 @@ class DimensionalModelLoader:
         """
         try:
             self.client.query(query).result()
-            self.logger.info(f"    [OK] FK: {table_name}.{fk_column} -> {ref_table}.{ref_column}")
+            self.logger.info(f"    [OK] FK: {table_name}.{fk_column} → {ref_table}.{ref_column}")
         except Exception as e:
             if "already exists" in str(e).lower():
                 self.logger.info(f"    [INFO] FK already exists: {constraint_name}")
@@ -129,7 +129,7 @@ class DimensionalModelLoader:
     # ------------------------------------------------------------------
 
     def _save_table_metadata(self, table_name: str):
-        meta     = self.load_metadata[table_name]
+        meta = self.load_metadata[table_name]
         metadata = {
             'source_class'       : self.__class__.__name__,
             'table_name'         : table_name,
@@ -145,7 +145,7 @@ class DimensionalModelLoader:
         }
         try:
             save_etl_metadata(self.client, metadata)
-            self.logger.info(f"    Metadata -> [AUDIT] etl_metadata")
+            self.logger.info(f"    Metadata → [AUDIT] etl_metadata")
         except Exception as e:
             self.logger.warning(f"    [WARN] Could not save metadata for {table_name}: {e}")
 
@@ -156,13 +156,13 @@ class DimensionalModelLoader:
     def load_dim_time(self):
         table_name = 'dim_time'
         self.load_metadata[table_name]['start_time'] = datetime.now()
-        self.logger.info("Loading dim_time -> [DW/Gold] fs_asean_gold...")
+        self.logger.info("Loading dim_time → [DW/Gold] fs_asean_gold...")
 
         try:
             if 'year_range' in self.df_clean.columns:
                 dim_time = self.df_clean[['year', 'year_range']].drop_duplicates().copy()
             else:
-                dim_time               = self.df_clean[['year']].drop_duplicates().copy()
+                dim_time              = self.df_clean[['year']].drop_duplicates().copy()
                 dim_time['year_range'] = None
 
             dim_time['year'] = dim_time['year'].astype(int)
@@ -194,10 +194,10 @@ class DimensionalModelLoader:
                                 pass
                 return pd.Series({'year': year, 'start_year': start_year, 'end_year': end_year})
 
-            parsed                    = dim_time.apply(parse_year_range, axis=1)
+            parsed                   = dim_time.apply(parse_year_range, axis=1)
-            dim_time['year']          = parsed['year'].astype(int)
+            dim_time['year']         = parsed['year'].astype(int)
-            dim_time['start_year']    = parsed['start_year'].astype(int)
+            dim_time['start_year']   = parsed['start_year'].astype(int)
-            dim_time['end_year']      = parsed['end_year'].astype(int)
+            dim_time['end_year']     = parsed['end_year'].astype(int)
             dim_time['is_year_range'] = (dim_time['start_year'] != dim_time['end_year'])
             dim_time['decade']        = (dim_time['year'] // 10) * 10
             dim_time['is_range']      = (dim_time['start_year'] != dim_time['end_year']).astype(int)
@@ -229,7 +229,7 @@ class DimensionalModelLoader:
             )
             log_update(self.client, 'DW', table_name, 'full_load', rows_loaded)
             self._save_table_metadata(table_name)
-            self.logger.info(f"  dim_time: {rows_loaded} rows\n")
+            self.logger.info(f"  ✓ dim_time: {rows_loaded} rows\n")
             return rows_loaded
 
         except Exception as e:
@@ -240,11 +240,11 @@ class DimensionalModelLoader:
     def load_dim_country(self):
         table_name = 'dim_country'
         self.load_metadata[table_name]['start_time'] = datetime.now()
-        self.logger.info("Loading dim_country -> [DW/Gold] fs_asean_gold...")
+        self.logger.info("Loading dim_country → [DW/Gold] fs_asean_gold...")
 
         try:
-            dim_country         = self.df_clean[['country']].drop_duplicates().copy()
+            dim_country            = self.df_clean[['country']].drop_duplicates().copy()
-            dim_country.columns = ['country_name']
+            dim_country.columns    = ['country_name']
 
             region_mapping = {
                 'Brunei Darussalam': ('Southeast Asia', 'ASEAN'),
@@ -270,9 +270,7 @@ class DimensionalModelLoader:
                 lambda x: region_mapping.get(x, ('Unknown', 'Unknown'))[1])
             dim_country['iso_code']  = dim_country['country_name'].map(iso_mapping)
 
-            dim_country_final = dim_country[
+            dim_country_final = dim_country[['country_name', 'region', 'subregion', 'iso_code']].copy()
-                ['country_name', 'region', 'subregion', 'iso_code']
-            ].copy()
             dim_country_final = dim_country_final.reset_index(drop=True)
             dim_country_final.insert(0, 'country_id', range(1, len(dim_country_final) + 1))
 
@@ -295,7 +293,7 @@ class DimensionalModelLoader:
             )
             log_update(self.client, 'DW', table_name, 'full_load', rows_loaded)
             self._save_table_metadata(table_name)
-            self.logger.info(f"  dim_country: {rows_loaded} rows\n")
+            self.logger.info(f"  ✓ dim_country: {rows_loaded} rows\n")
             return rows_loaded
 
         except Exception as e:
@@ -304,106 +302,60 @@ class DimensionalModelLoader:
             raise
 
     def load_dim_indicator(self):
-        """
-        Load dim_indicator ke Gold layer.
-
-        Kolom yang dimuat:
-            indicator_id       — surrogate key
-            indicator_name     — nama standar indikator
-            indicator_category — kategori (Health & Nutrition, dll.)
-            unit               — satuan ukuran
-            direction          — higher_better / lower_better
-            framework          — MDGs / SDGs  <-- BARU: dibaca dari cleaned_integrated
-        """
         table_name = 'dim_indicator'
         self.load_metadata[table_name]['start_time'] = datetime.now()
-        self.logger.info("Loading dim_indicator -> [DW/Gold] fs_asean_gold...")
+        self.logger.info("Loading dim_indicator → [DW/Gold] fs_asean_gold...")
 
         try:
             has_direction = 'direction'          in self.df_clean.columns
             has_unit      = 'unit'               in self.df_clean.columns
             has_category  = 'indicator_category' in self.df_clean.columns
-            has_framework = 'framework'          in self.df_clean.columns
 
-            dim_indicator         = self.df_clean[['indicator_standardized']].drop_duplicates().copy()
+            dim_indicator          = self.df_clean[['indicator_standardized']].drop_duplicates().copy()
-            dim_indicator.columns = ['indicator_name']
+            dim_indicator.columns  = ['indicator_name']
 
-            # Unit
             if has_unit:
                 unit_map = self.df_clean[['indicator_standardized', 'unit']].drop_duplicates()
-                unit_map.columns = ['indicator_name', 'unit']
+                unit_map.columns  = ['indicator_name', 'unit']
-                dim_indicator    = dim_indicator.merge(unit_map, on='indicator_name', how='left')
+                dim_indicator     = dim_indicator.merge(unit_map, on='indicator_name', how='left')
             else:
                 dim_indicator['unit'] = None
 
-            # Direction
             if has_direction:
                 dir_map = self.df_clean[['indicator_standardized', 'direction']].drop_duplicates()
-                dir_map.columns  = ['indicator_name', 'direction']
+                dir_map.columns   = ['indicator_name', 'direction']
-                dim_indicator    = dim_indicator.merge(dir_map, on='indicator_name', how='left')
+                dim_indicator     = dim_indicator.merge(dir_map, on='indicator_name', how='left')
                 self.logger.info("  [OK] direction column from cleaned_integrated")
             else:
                 dim_indicator['direction'] = 'higher_better'
                 self.logger.warning("  [WARN] direction not found, default: higher_better")
 
-            # Indicator category
             if has_category:
-                cat_map = self.df_clean[
+                cat_map = self.df_clean[['indicator_standardized', 'indicator_category']].drop_duplicates()
-                    ['indicator_standardized', 'indicator_category']
+                cat_map.columns   = ['indicator_name', 'indicator_category']
-                ].drop_duplicates()
+                dim_indicator     = dim_indicator.merge(cat_map, on='indicator_name', how='left')
-                cat_map.columns = ['indicator_name', 'indicator_category']
-                dim_indicator   = dim_indicator.merge(cat_map, on='indicator_name', how='left')
             else:
                 def categorize_indicator(name):
                     n = str(name).lower()
-                    if any(w in n for w in [
+                    if any(w in n for w in ['undernourishment', 'malnutrition', 'stunting',
-                        'undernourishment', 'malnutrition', 'stunting',
+                                             'wasting', 'anemia', 'food security', 'food insecure', 'hunger']):
-                        'wasting', 'anemia', 'anaemia', 'food security',
-                        'food insecure', 'hunger'
-                    ]):
                         return 'Health & Nutrition'
-                    elif any(w in n for w in [
+                    elif any(w in n for w in ['production', 'yield', 'cereal', 'crop',
-                        'production', 'yield', 'cereal', 'crop',
+                                               'import dependency', 'share of dietary']):
-                        'import dependency', 'share of dietary'
-                    ]):
                         return 'Agricultural Production'
                     elif any(w in n for w in ['import', 'export', 'trade']):
                         return 'Trade'
                     elif any(w in n for w in ['gdp', 'income', 'economic']):
                         return 'Economic'
-                    elif any(w in n for w in [
+                    elif any(w in n for w in ['water', 'sanitation', 'infrastructure', 'rail']):
-                        'water', 'sanitation', 'infrastructure', 'rail'
-                    ]):
                         return 'Infrastructure'
                     else:
-                        return 'Other'
+                        return 'Sustainability'
-                dim_indicator['indicator_category'] = dim_indicator['indicator_name'].apply(
+                dim_indicator['indicator_category'] = dim_indicator['indicator_name'].apply(categorize_indicator)
-                    categorize_indicator
-                )
-
-            # Framework — KOLOM BARU
-            # Dibaca dari cleaned_integrated yang sudah menjalankan assign_framework().
-            # Jika kolom belum ada (misal pipeline lama), fallback ke 'MDGs' dengan warning.
-            if has_framework:
-                fw_map = self.df_clean[
-                    ['indicator_standardized', 'framework']
-                ].drop_duplicates()
-                fw_map.columns = ['indicator_name', 'framework']
-                dim_indicator  = dim_indicator.merge(fw_map, on='indicator_name', how='left')
-                # Pastikan tidak ada NULL setelah merge
-                dim_indicator['framework'] = dim_indicator['framework'].fillna('MDGs')
-                self.logger.info("  [OK] framework column from cleaned_integrated")
-            else:
-                dim_indicator['framework'] = 'MDGs'
-                self.logger.warning(
-                    "  [WARN] framework column not found in cleaned_integrated. "
-                    "Default: MDGs. Jalankan bigquery_cleaned_layer.py terlebih dahulu."
-                )
-
-            dim_indicator = dim_indicator.drop_duplicates(subset=['indicator_name'], keep='first')
 
+            dim_indicator     = dim_indicator.drop_duplicates(subset=['indicator_name'], keep='first')
             dim_indicator_final = dim_indicator[
-                ['indicator_name', 'indicator_category', 'unit', 'direction', 'framework']
+                ['indicator_name', 'indicator_category', 'unit', 'direction']
             ].copy()
             dim_indicator_final = dim_indicator_final.reset_index(drop=True)
             dim_indicator_final.insert(0, 'indicator_id', range(1, len(dim_indicator_final) + 1))
@@ -414,7 +366,6 @@ class DimensionalModelLoader:
                 bigquery.SchemaField("indicator_category", "STRING",  mode="REQUIRED"),
                 bigquery.SchemaField("unit",               "STRING",  mode="NULLABLE"),
                 bigquery.SchemaField("direction",          "STRING",  mode="REQUIRED"),
-                bigquery.SchemaField("framework",          "STRING",  mode="REQUIRED"),
             ]
 
             rows_loaded = load_to_bigquery(
@@ -423,23 +374,17 @@ class DimensionalModelLoader:
             )
             self._add_primary_key(table_name, 'indicator_id')
 
-            # Log distribusi
+            for label, col in [('Categories', 'indicator_category'), ('Direction', 'direction')]:
-            for label, col in [
-                ('Categories', 'indicator_category'),
-                ('Direction',  'direction'),
-                ('Framework',  'framework'),
-            ]:
                 self.logger.info(f"  {label}:")
                 for val, cnt in dim_indicator_final[col].value_counts().items():
-                    pct = cnt / len(dim_indicator_final) * 100
+                    self.logger.info(f"    - {val}: {cnt} ({cnt/len(dim_indicator_final)*100:.1f}%)")
-                    self.logger.info(f"    - {val}: {cnt} ({pct:.1f}%)")
 
             self.load_metadata[table_name].update(
                 {'rows_loaded': rows_loaded, 'status': 'success', 'end_time': datetime.now()}
             )
             log_update(self.client, 'DW', table_name, 'full_load', rows_loaded)
             self._save_table_metadata(table_name)
-            self.logger.info(f"  dim_indicator: {rows_loaded} rows\n")
+            self.logger.info(f"  ✓ dim_indicator: {rows_loaded} rows\n")
             return rows_loaded
 
         except Exception as e:
@@ -450,7 +395,7 @@ class DimensionalModelLoader:
     def load_dim_source(self):
         table_name = 'dim_source'
         self.load_metadata[table_name]['start_time'] = datetime.now()
-        self.logger.info("Loading dim_source -> [DW/Gold] fs_asean_gold...")
+        self.logger.info("Loading dim_source → [DW/Gold] fs_asean_gold...")
 
         try:
             source_details = {
@@ -510,7 +455,7 @@ class DimensionalModelLoader:
             )
             log_update(self.client, 'DW', table_name, 'full_load', rows_loaded)
             self._save_table_metadata(table_name)
-            self.logger.info(f"  dim_source: {rows_loaded} rows\n")
+            self.logger.info(f"  ✓ dim_source: {rows_loaded} rows\n")
             return rows_loaded
 
         except Exception as e:
@@ -521,15 +466,15 @@ class DimensionalModelLoader:
     def load_dim_pillar(self):
         table_name = 'dim_pillar'
         self.load_metadata[table_name]['start_time'] = datetime.now()
-        self.logger.info("Loading dim_pillar -> [DW/Gold] fs_asean_gold...")
+        self.logger.info("Loading dim_pillar → [DW/Gold] fs_asean_gold...")
 
         try:
             pillar_codes = {
                 'Availability': 'AVL', 'Access'  : 'ACC',
-                'Utilization' : 'UTL', 'Stability': 'STB', 'Other': 'OTH',
+                'Utilization' : 'UTL', 'Stability': 'STB', 'Sustainability': 'STN',
             }
             pillars_data = [
-                {'pillar_name': p, 'pillar_code': pillar_codes.get(p, 'OTH')}
+                {'pillar_name': p, 'pillar_code': pillar_codes.get(p, 'STN')}
                 for p in self.df_clean['pillar'].unique()
             ]
 
@@ -556,7 +501,7 @@ class DimensionalModelLoader:
             )
             log_update(self.client, 'DW', table_name, 'full_load', rows_loaded)
             self._save_table_metadata(table_name)
-            self.logger.info(f"  dim_pillar: {rows_loaded} rows\n")
+            self.logger.info(f"  ✓ dim_pillar: {rows_loaded} rows\n")
             return rows_loaded
 
         except Exception as e:
@@ -571,9 +516,10 @@ class DimensionalModelLoader:
     def load_fact_food_security(self):
         table_name = 'fact_food_security'
         self.load_metadata[table_name]['start_time'] = datetime.now()
-        self.logger.info("Loading fact_food_security -> [DW/Gold] fs_asean_gold...")
+        self.logger.info("Loading fact_food_security → [DW/Gold] fs_asean_gold...")
 
         try:
+            # Load dims dari Gold untuk FK resolution
             dim_country   = read_from_bigquery(self.client, 'dim_country',   layer='gold')
             dim_indicator = read_from_bigquery(self.client, 'dim_indicator', layer='gold')
             dim_time      = read_from_bigquery(self.client, 'dim_time',      layer='gold')
@@ -615,9 +561,9 @@ class DimensionalModelLoader:
                 fact_table['start_year'] = fact_table['year'].astype(int)
                 fact_table['end_year']   = fact_table['year'].astype(int)
 
+            # Resolve FKs
             fact_table = fact_table.merge(
-                dim_country[['country_id', 'country_name']].rename(
+                dim_country[['country_id', 'country_name']].rename(columns={'country_name': 'country'}),
-                    columns={'country_name': 'country'}),
                 on='country', how='left'
             )
             fact_table = fact_table.merge(
@@ -630,16 +576,15 @@ class DimensionalModelLoader:
                 on=['start_year', 'end_year'], how='left'
             )
             fact_table = fact_table.merge(
-                dim_source[['source_id', 'source_name']].rename(
+                dim_source[['source_id', 'source_name']].rename(columns={'source_name': 'source'}),
-                    columns={'source_name': 'source'}),
                 on='source', how='left'
             )
             fact_table = fact_table.merge(
-                dim_pillar[['pillar_id', 'pillar_name']].rename(
+                dim_pillar[['pillar_id', 'pillar_name']].rename(columns={'pillar_name': 'pillar'}),
-                    columns={'pillar_name': 'pillar'}),
                 on='pillar', how='left'
             )
 
+            # Filter hanya row dengan FK lengkap
             fact_table = fact_table[
                 fact_table['country_id'].notna()   &
                 fact_table['indicator_id'].notna() &
@@ -676,6 +621,7 @@ class DimensionalModelLoader:
                 layer='gold', write_disposition="WRITE_TRUNCATE", schema=schema
             )
 
+            # Add PK + FKs
             self._add_primary_key(table_name, 'fact_id')
             self._add_foreign_key(table_name, 'country_id',   'dim_country',   'country_id')
             self._add_foreign_key(table_name, 'indicator_id', 'dim_indicator', 'indicator_id')
@@ -688,7 +634,7 @@ class DimensionalModelLoader:
             )
             log_update(self.client, 'DW', table_name, 'full_load', rows_loaded)
             self._save_table_metadata(table_name)
-            self.logger.info(f"  fact_food_security: {rows_loaded:,} rows\n")
+            self.logger.info(f"  ✓ fact_food_security: {rows_loaded:,} rows\n")
             return rows_loaded
 
         except Exception as e:
@@ -766,36 +712,16 @@ class DimensionalModelLoader:
             self.logger.info(f"    Unique Sources    : {int(stats['unique_sources']):>10,}")
             self.logger.info(f"    Unique Pillars    : {int(stats['unique_pillars']):>10,}")
 
-            # Validasi distribusi framework di dim_indicator
-            query_fw = f"""
-            SELECT framework, COUNT(*) AS count
-            FROM `{get_table_id('dim_indicator', layer='gold')}`
-            GROUP BY framework ORDER BY framework
-            """
-            df_fw = self.client.query(query_fw).result().to_dataframe(
-                create_bqstorage_client=False
-            )
-            if len(df_fw) > 0:
-                self.logger.info(f"\n  Framework Distribution (dim_indicator):")
-                for _, row in df_fw.iterrows():
-                    self.logger.info(
-                        f"    {row['framework']:10s}: {int(row['count']):>5,} indicators"
-                    )
-
             query_dir = f"""
             SELECT direction, COUNT(*) AS count
             FROM `{get_table_id('dim_indicator', layer='gold')}`
             GROUP BY direction ORDER BY direction
             """
-            df_dir = self.client.query(query_dir).result().to_dataframe(
+            df_dir = self.client.query(query_dir).result().to_dataframe(create_bqstorage_client=False)
-                create_bqstorage_client=False
-            )
             if len(df_dir) > 0:
                 self.logger.info(f"\n  Direction Distribution:")
                 for _, row in df_dir.iterrows():
-                    self.logger.info(
+                    self.logger.info(f"    {row['direction']:15s}: {int(row['count']):>5,} indicators")
-                        f"    {row['direction']:15s}: {int(row['count']):>5,} indicators"
-                    )
 
             self.logger.info("\n  [OK] Validation completed")
         except Exception as e:
@@ -812,19 +738,22 @@ class DimensionalModelLoader:
         self.pipeline_metadata['rows_fetched'] = len(self.df_clean)
 
         self.logger.info("\n" + "=" * 60)
-        self.logger.info("DIMENSIONAL MODEL LOAD — DW (Gold) -> fs_asean_gold")
+        self.logger.info("DIMENSIONAL MODEL LOAD — DW (Gold) → fs_asean_gold")
         self.logger.info("=" * 60)
 
-        self.logger.info("\nLOADING DIMENSION TABLES -> fs_asean_gold")
+        # Dimensions
+        self.logger.info("\nLOADING DIMENSION TABLES → fs_asean_gold")
         self.load_dim_country()
         self.load_dim_indicator()
         self.load_dim_time()
         self.load_dim_source()
         self.load_dim_pillar()
 
-        self.logger.info("\nLOADING FACT TABLE -> fs_asean_gold")
+        # Fact
+        self.logger.info("\nLOADING FACT TABLE → fs_asean_gold")
         self.load_fact_food_security()
 
+        # Validate
         self.validate_constraints()
         self.validate_data_load()
 
@@ -833,23 +762,22 @@ class DimensionalModelLoader:
         total_loaded = sum(m['rows_loaded'] for m in self.load_metadata.values())
 
         self.pipeline_metadata.update({
-            'end_time'           : pipeline_end,
+            'end_time'          : pipeline_end,
-            'duration_seconds'   : duration,
+            'duration_seconds'  : duration,
-            'rows_transformed'   : total_loaded,
+            'rows_transformed'  : total_loaded,
-            'rows_loaded'        : total_loaded,
+            'rows_loaded'       : total_loaded,
             'execution_timestamp': self.pipeline_metadata['start_time'],
-            'completeness_pct'   : 100.0,
+            'completeness_pct'  : 100.0,
-            'config_snapshot'    : json.dumps({'load_mode': 'full_refresh', 'layer': 'gold'}),
+            'config_snapshot'   : json.dumps({'load_mode': 'full_refresh', 'layer': 'gold'}),
-            'validation_metrics' : json.dumps(
+            'validation_metrics': json.dumps({t: m['status'] for t, m in self.load_metadata.items()}),
-                {t: m['status'] for t, m in self.load_metadata.items()}
+            'table_name'        : 'dimensional_model_pipeline',
-            ),
-            'table_name'         : 'dimensional_model_pipeline',
         })
         try:
             save_etl_metadata(self.client, self.pipeline_metadata)
         except Exception as e:
             self.logger.warning(f"  [WARN] Could not save pipeline metadata: {e}")
 
+        # Summary
         self.logger.info("\n" + "=" * 60)
         self.logger.info("DIMENSIONAL MODEL LOAD COMPLETED")
         self.logger.info("=" * 60)
@@ -857,19 +785,20 @@ class DimensionalModelLoader:
         self.logger.info(f"  Duration : {duration:.2f}s")
         self.logger.info(f"  Tables   :")
         for tbl, meta in self.load_metadata.items():
-            icon = "OK" if meta['status'] == 'success' else "FAIL"
+            icon = "✓" if meta['status'] == 'success' else "✗"
-            self.logger.info(f"    [{icon}] {tbl:25s}: {meta['rows_loaded']:>10,} rows")
+            self.logger.info(f"    {icon} {tbl:25s}: {meta['rows_loaded']:>10,} rows")
-        self.logger.info(f"\n  Metadata -> [AUDIT] etl_metadata")
+        self.logger.info(f"\n  Metadata → [AUDIT] etl_metadata")
         self.logger.info("=" * 60)
 
 
 # =============================================================================
-# AIRFLOW TASK FUNCTIONS
+# AIRFLOW TASK FUNCTIONS  ← sama polanya dengan raw & cleaned layer
 # =============================================================================
 
 def run_dimensional_model():
     """
     Airflow task: Load dimensional model dari cleaned_integrated.
+
     Dipanggil oleh DAG setelah task cleaned_integration_to_silver selesai.
     """
     from scripts.bigquery_config import get_bigquery_client
@@ -888,9 +817,9 @@ if __name__ == "__main__":
     print("=" * 60)
     print("BIGQUERY DIMENSIONAL MODEL LOAD")
     print("Kimball DW Architecture")
-    print("  Input  : STAGING (Silver) -> cleaned_integrated (fs_asean_silver)")
+    print("  Input  : STAGING (Silver) → cleaned_integrated (fs_asean_silver)")
-    print("  Output : DW      (Gold)   -> dim_*, fact_*       (fs_asean_gold)")
+    print("  Output : DW      (Gold)   → dim_*, fact_*       (fs_asean_gold)")
-    print("  Audit  : AUDIT            -> etl_logs, etl_metadata (fs_asean_audit)")
+    print("  Audit  : AUDIT            → etl_logs, etl_metadata (fs_asean_audit)")
     print("=" * 60)
 
     logger   = setup_logging()
@@ -898,26 +827,24 @@ if __name__ == "__main__":
 
     print("\nLoading cleaned_integrated (fs_asean_silver)...")
     df_clean = read_from_bigquery(client, 'cleaned_integrated', layer='silver')
-    print(f"  Loaded         : {len(df_clean):,} rows")
+    print(f"  ✓ Loaded       : {len(df_clean):,} rows")
     print(f"    Columns      : {len(df_clean.columns)}")
     print(f"    Sources      : {df_clean['source'].nunique()}")
     print(f"    Indicators   : {df_clean['indicator_standardized'].nunique()}")
     print(f"    Countries    : {df_clean['country'].nunique()}")
-    print(f"    Year range   : {int(df_clean['year'].min())}-{int(df_clean['year'].max())}")
+    print(f"    Year range   : {int(df_clean['year'].min())}–{int(df_clean['year'].max())}")
     if 'direction' in df_clean.columns:
         print(f"    Direction    : {df_clean['direction'].value_counts().to_dict()}")
-    if 'framework' in df_clean.columns:
-        print(f"    Framework    : {df_clean['framework'].value_counts().to_dict()}")
     else:
-        print("    [WARN] framework column not found — run bigquery_cleaned_layer.py first")
+        print(f"    [WARN] direction column not found — run bigquery_cleaned_layer.py first")
 
-    print("\n[1/1] Dimensional Model Load -> DW (Gold)...")
+    print("\n[1/1] Dimensional Model Load → DW (Gold)...")
     loader = DimensionalModelLoader(client, df_clean)
     loader.run()
 
     print("\n" + "=" * 60)
-    print("[OK] DIMENSIONAL MODEL ETL COMPLETED")
+    print("✓ DIMENSIONAL MODEL ETL COMPLETED")
-    print("  DW (Gold) : dim_country, dim_indicator (+ framework),")
+    print("  🥇 DW (Gold) : dim_country, dim_indicator, dim_time,")
-    print("              dim_time, dim_source, dim_pillar, fact_food_security")
+    print("                dim_source, dim_pillar, fact_food_security")
-    print("  AUDIT     : etl_logs, etl_metadata")
+    print("  📋 AUDIT     : etl_logs, etl_metadata")
     print("=" * 60)