from google.cloud import storage from airflow.models import DAG from airflow.utils.dates import days_ago from airflow.providers.google.cloud.operators.gcs import GCSDeleteObjectsOperator from airflow.providers.google.cloud.transfers.gcs_to_bigquery import GCSToBigQueryOperator from airflow.providers.google.cloud.transfers.postgres_to_gcs import PostgresToGCSOperator from airflow.operators.python import PythonOperator from airflow.hooks.postgres_hook import PostgresHook import pandas as pd import io BQ_PROJECT = 'exemplary-city-396006' BQ_DS = 'jadi' GCS_BUCKET = 'tes-airflow' GCS_OBJECT_PATH = 'postgres-test' SOURCE_TABLE_NAME = 'dataset' POSTGRES_CONNECTION_ID = 'postgres_default' storage.blob._DEFAULT_CHUNKSIZE = 5 * 1024* 1024 # 5 MB storage.blob._MAX_MULTIPART_SIZE = 5 * 1024* 1024 # 5 MB hook = PostgresHook(postgres_conn_id=POSTGRES_CONNECTION_ID) def postgres_to_gcs(tables, **kwargs): # Create a GCS client client = storage.Client() #tables = ['locations','measurements','individuals'] # SQL query to execute for table_name in tables: df = hook.get_pandas_df(f"SELECT * FROM {table_name};") # Convert dataframe to parquet (or csv if preferred) csv_buffer = io.StringIO() df.to_csv(csv_buffer, index=False) csv_buffer.seek(0) bucket_name = GCS_BUCKET blob_name = f'{GCS_OBJECT_PATH}/{table_name}.csv' # Get the bucket and blob bucket = client.get_bucket(bucket_name) blob = bucket.blob(blob_name) # Upload the CSV file to GCS blob.upload_from_string(csv_buffer.getvalue(), content_type='text/csv') def transform_balita(ds, **kwargs): client = storage.Client() bucket_name = GCS_BUCKET file_name= 'individuals' source_blob_name = f'{GCS_OBJECT_PATH}/{file_name}.csv' destination_blob_name = f'data/clean_{file_name}.csv' # Get the bucket and blob bucket = client.get_bucket(bucket_name) blob = bucket.blob(source_blob_name) # Download the data from GCS data = blob.download_as_string() data = pd.read_csv(io.StringIO(data.decode('utf-8'))) # Select specific columns and remove duplicates # data_selected = ['nik', 'nama', 'jk', 'tgl_lahir', 'bb_lahir', 'tb_lahir'] # data_cleaned = data[data_selected].drop_duplicates() # Convert 'nik' to string to ensure it's treated as varchar # data_cleaned['nik'] = data['nik'].astype(str) # Mengonversi kolom 'nik' ke string dan menyimpannya sebagai DataFrame # data_cleaned = data['nik'].astype(str).to_frame(name='nik') # Menggantikan nilai '-' dengan NaN data_cleaned = data.replace('-', pd.NA) data_cleaned = data_cleaned.applymap(lambda x: str(x).title() if isinstance(x, str) else x) # Convert the 'tgl_lahir' columns to proper datetime formats data_cleaned['tgl_lahir'] = pd.to_datetime(data_cleaned['tgl_lahir'], errors='coerce') # Impute missing values for 'bb_lahir' and 'tb_lahir' with the median data_cleaned['bb_lahir'].fillna(data_cleaned['bb_lahir'].median(), inplace=True) data_cleaned['tb_lahir'].fillna(data_cleaned['tb_lahir'].median(), inplace=True) # Mengisikan nilai "Tidak Diketahui" pada kolom 'Nama Ortu' yang hilang # data_cleaned['nama_ortu'].fillna('Tidak Diketahui', inplace=True) # data_cleaned['nama_ortu'] = data_cleaned['nama_ortu'].str.replace(' /', '/').str.replace('/ ', '/').str.replace('/', ' / ') # Add an auto-incrementing ID column for primary key data_cleaned.insert(0, 'balita_id', range(1, 1 + len(data_cleaned))) blob = bucket.blob(destination_blob_name) blob.upload_from_string(data_cleaned.to_csv(index=False), 'application/csv') def transform_measurements(ds, **kwargs): client = storage.Client() bucket_name = GCS_BUCKET file_name= 'measurements' source_blob_name = f'{GCS_OBJECT_PATH}/{file_name}.csv' destination_blob_name = f'data/clean_{file_name}.csv' bucket = client.get_bucket(bucket_name) blob = bucket.blob(source_blob_name) data = blob.download_as_string() data = pd.read_csv(io.StringIO(data.decode('utf-8'))) # data_selected = ['usia_saat_ukur', 'berat', 'tinggi', 'bb_u', 'zs_bb_u', 'tb_u', 'zs_tb_u', 'bb_tb', 'zs_bb_tb'] # data_cleaned = data[data_selected].drop_duplicates() # Menggantikan nilai '-' dengan NaN data_cleaned = data.replace('-', pd.NA) data_cleaned = data_cleaned.dropna(subset=['berat','tinggi']) # Define the status categories status_gizi = { 'BB/U': [ ('Berat Badan Sangat Kurang', -9999, -3), ('Berat Badan Kurang', -3, -2), ('Berat Badan Normal', -2, 1), ('Risiko Lebih', 1, 9999) ], 'TB/U': [ ('Sangat Pendek', -9999, -3), ('Pendek', -3, -2), ('Normal', -2, 2), ('Tinggi', 2, 9999) ], 'BB/TB': [ ('Gizi Buruk', -9999, -3), ('Gizi Kurang', -3, -2), ('Gizi Baik', -2, 1), ('Berisiko Gizi Lebih', 1, 2), ('Gizi Lebih', 2, 3), ('Obesitas', 3, 9999) ] } # Function to get nutritional status based on Z-Score and category def get_status_gizi(zscore, kategori): try: zscore = float(zscore) # Ensure zscore is a float except ValueError: return None # Return None if zscore is not a number for status, min_val, max_val in status_gizi[kategori]: if min_val <= zscore < max_val: return status return None # Convert Z-scores to numeric, coercing errors to NaN data_cleaned['zs_bb_u'] = pd.to_numeric(data_cleaned['zs_bb_u'], errors='coerce') data_cleaned['zs_tb_u'] = pd.to_numeric(data_cleaned['zs_tb_u'], errors='coerce') data_cleaned['zs_bb_tb'] = pd.to_numeric(data_cleaned['zs_bb_tb'], errors='coerce') # Impute the nutritional status based on Z-Scores where missing data_cleaned['bb_u'] = data_cleaned.apply(lambda row: get_status_gizi(row['zs_bb_u'], 'BB/U') if pd.isnull(row['bb_u']) else row['bb_u'], axis=1) data_cleaned['tb_u'] = data_cleaned.apply(lambda row: get_status_gizi(row['zs_tb_u'], 'TB/U') if pd.isnull(row['tb_u']) else row['tb_u'], axis=1) data_cleaned['bb_tb'] = data_cleaned.apply(lambda row: get_status_gizi(row['zs_bb_tb'], 'BB/TB') if pd.isnull(row['bb_tb']) else row['bb_tb'], axis=1) data_cleaned['tanggal_pengukuran'] = pd.to_datetime(data_cleaned['tanggal_pengukuran'], errors='coerce') data_cleaned.insert(0, 'pengukuran_id', range(1, 1 + len(data_cleaned))) blob = bucket.blob(destination_blob_name) blob.upload_from_string(data_cleaned.to_csv(index=False), 'application/csv') def transform_locations(ds, **kwargs): client = storage.Client() bucket_name = GCS_BUCKET file_name= 'locations' source_blob_name = f'{GCS_OBJECT_PATH}/{file_name}.csv' destination_blob_name = f'data/clean_{file_name}.csv' bucket = client.get_bucket(bucket_name) blob = bucket.blob(source_blob_name) data = blob.download_as_string() data = pd.read_csv(io.StringIO(data.decode('utf-8'))) # data_selected = ['prov', 'kab_kota', 'kec', 'pukesmas', 'desa_kel', 'posyandu'] # data_cleaned = data[data_selected].drop_duplicates() # Menggantikan nilai '-' dengan NaN data_cleaned = data.replace('-', pd.NA) data_cleaned = data_cleaned.applymap(lambda x: str(x).title() if isinstance(x, str) else x) data_cleaned['posyandu'].fillna('Tidak Diketahui', inplace=True) data_cleaned.insert(0, 'lokasi_id', range(1, 1 + len(data_cleaned))) blob = bucket.blob(destination_blob_name) blob.upload_from_string(data_cleaned.to_csv(index=False), 'application/csv') def transform_statusGizi(ds, **kwargs): client = storage.Client() bucket_name = GCS_BUCKET file_name= 'status_gizi_zscore' source_blob_name = f'{GCS_OBJECT_PATH}/{file_name}.csv' destination_blob_name = f'data/clean_{file_name}.csv' bucket = client.get_bucket(bucket_name) blob = bucket.blob(source_blob_name) data = blob.download_as_string() data = pd.read_csv(io.StringIO(data.decode('utf-8'))) data_selected = ['kategori', 'status_gizi', 'ambang_batas_bawah', 'ambang_batas_atas'] data_cleaned = data[data_selected].drop_duplicates() data_cleaned.insert(0, 'status_gizi_id', range(1, 1 + len(data_cleaned))) blob = bucket.blob(destination_blob_name) blob.upload_from_string(data_cleaned.to_csv(index=False), 'application/csv') def transform_date(ds, **kwargs): client = storage.Client() bucket_name = GCS_BUCKET file_name= 'measurements' source_blob_name = f'{GCS_OBJECT_PATH}/{file_name}.csv' destination_blob_name = f'data/clean_date.csv' bucket = client.get_bucket(bucket_name) blob = bucket.blob(source_blob_name) data = blob.download_as_string() data = pd.read_csv(io.StringIO(data.decode('utf-8'))) data = data.dropna(subset=['tanggal_pengukuran']) # Make sure 'tanggal_pengukuran' is a datetime column data['date'] = pd.to_datetime(data['tanggal_pengukuran']) # Extract year, month, and day into separate columns data['year'] = data['date'].dt.year.astype(int) data['month'] = data['date'].dt.month.astype(int) data['day'] = data['date'].dt.day.astype(int) data_selected = ['date', 'year', 'month', 'day'] data_cleaned = data[data_selected].drop_duplicates() data_cleaned.insert(0, 'date_id', range(1, 1 + len(data_cleaned))) blob = bucket.blob(destination_blob_name) blob.upload_from_string(data_cleaned.to_csv(index=False), 'application/csv') def fact(ds, **kwargs): client = storage.Client() bucket_name = GCS_BUCKET # file_name= 'measurements' # source_blob_name = f'{GCS_OBJECT_PATH}/{file_name}.csv' destination_blob_name = f'data/fact_gizi.csv' bucket = client.get_bucket(bucket_name) # blob = bucket.blob(source_blob_name) # data = blob.download_as_string() # data = pd.read_csv(io.StringIO(data.decode('utf-8'))) # Ubah objek bytes ke dalam bentuk string menggunakan io.BytesIO data_individuals = io.BytesIO(bucket.blob(f'data/clean_individuals.csv').download_as_string()) data_locations = io.BytesIO(bucket.blob(f'data/clean_locations.csv').download_as_string()) data_measurements = io.BytesIO(bucket.blob(f'data/clean_measurements.csv').download_as_string()) # Baca data sebagai DataFrame menggunakan pd.read_csv individuals_df = pd.read_csv(data_individuals) locations_df = pd.read_csv(data_locations) measurements_df = pd.read_csv(data_measurements) # Cleaning and preparing the dimension tables # individuals_df = individuals_df.drop_duplicates(subset=['nik']) # Select specific columns and remove duplicates # Create the Date dimension table measurements_df['tanggal_pengukuran'] = pd.to_datetime(measurements_df['tanggal_pengukuran']) dim_date = measurements_df['tanggal_pengukuran'].drop_duplicates().sort_values().reset_index(drop=True) dim_date = pd.DataFrame({ 'date_key': dim_date.dt.date, 'year': dim_date.dt.year, 'month': dim_date.dt.month, 'day': dim_date.dt.day, 'week': dim_date.dt.isocalendar().week, 'quarter': dim_date.dt.quarter }) # Remove any potential NaT values dim_date = dim_date.dropna(subset=['date_key']) # Preparing the fact table by merging individuals and measurements dataframes fact_table = pd.merge(individuals_df[['id', 'location_id', 'measurement_id']], measurements_df, on='measurement_id', how='inner') fact_table = fact_table.rename(columns={ 'id': 'balita_id', 'tanggal_pengukuran': 'date_key', 'berat': 'weight', 'tinggi': 'height' }) fact_table['date_key'] = pd.to_datetime(fact_table['date_key']).dt.date # Selecting and ordering the columns for the fact table fact_table = fact_table[ ['pengukuran_id', 'balita_id', 'location_id', 'date_key', 'weight', 'height', 'bb_u', 'zs_bb_u', 'tb_u', 'zs_tb_u', 'bb_tb', 'zs_bb_tb'] ] selected_pengukuran = ['usia_saat_ukur', 'berat', 'tinggi', 'zs_bb_u', 'zs_tb_u', 'zs_bb_tb'] dim_pengukuran = measurements_df[selected_pengukuran].drop_duplicates() dim_pengukuran.insert(0, 'pengukuran_id', range(1, 1 + len(dim_pengukuran))) # Step 1: Extract unique combinations for Nutritional Status Dimension nutritional_status_combinations = fact_table[['bb_u', 'tb_u', 'bb_tb']].drop_duplicates().reset_index(drop=True) # Step 2: Add a surrogate key for the nutritional status dimension nutritional_status_combinations.insert(0, 'status_gizi_id', range(1, 1 + len(nutritional_status_combinations))) # Step 3: Rename columns for clarity dim_status_gizi = nutritional_status_combinations.rename(columns={ 'bb_u': 'bb_u_status', 'tb_u': 'tb_u_status', 'bb_tb': 'bb_tb_status' }) # Step 4: Merge this dimension back to the fact table to replace descriptive columns with the surrogate key fact_table_with_status = pd.merge( fact_table, dim_status_gizi, how='left', left_on=['bb_u', 'tb_u', 'bb_tb'], right_on=['bb_u_status', 'tb_u_status', 'bb_tb_status'] ) # Step 5: Select the relevant columns for the fact table, replacing descriptive statuses with nutritional_status_id fact_table_final = fact_table_with_status[['pengukuran_id', 'balita_id', 'location_id', 'date_key', 'weight', 'height', 'status_gizi_id']] # Create Z-Scores Dimension # Update the fact table with the nutritional status id fact_table = pd.merge( fact_table, dim_status_gizi[['status_gizi_id', 'bb_u_status', 'tb_u_status', 'bb_tb_status']], how='left', left_on=['bb_u', 'tb_u', 'bb_tb'], right_on=['bb_u_status', 'tb_u_status', 'bb_tb_status'] ) # Finalize the fact table fact_table_final = fact_table[['pengukuran_id', 'balita_id', 'location_id', 'date_key', 'status_gizi_id', ]] selected_balita = ['nik', 'nama', 'jk', 'tgl_lahir', 'bb_lahir', 'tb_lahir'] dim_balita = individuals_df[selected_balita].drop_duplicates() dim_balita.insert(0, 'id', range(1, 1 + len(dim_balita))) # Define the columns you want to select selected_locations = ['prov', 'kab_kota', 'kec', 'pukesmas', 'desa_kel', 'posyandu'] # Select only the defined columns and drop duplicates dim_locations = locations_df[selected_locations].drop_duplicates() dim_locations.insert(0, 'location_id', range(1, 1 + len(dim_locations))) fact_table_final.insert(0, 'fact_id', range(1, 1 + len(fact_table_final))) destination_blob_name=f'data/fact_gizi.csv' blob = bucket.blob(destination_blob_name) blob.upload_from_string(fact_table_final.to_csv(index=False), 'application/csv') bucket.blob('data/dim_status.csv').upload_from_string(dim_status_gizi.to_csv(index=False), 'application/csv') bucket.blob('data/dim_balita.csv').upload_from_string(dim_balita.to_csv(index=False), 'application/csv') bucket.blob('data/dim_date.csv').upload_from_string(dim_date.to_csv(index=False), 'application/csv') bucket.blob('data/dim_measure.csv').upload_from_string(dim_pengukuran.to_csv(index=False), 'application/csv') bucket.blob('data/dim_location.csv').upload_from_string(dim_locations.to_csv(index=False), 'application/csv') print(dim_balita) print(dim_pengukuran) # NOTE: DAG declaration - using a Context Manager (an implicit way) with DAG( dag_id='bisaatuh', start_date=days_ago(1), default_args={ 'owner': 'airflow', 'retries': 2, #'retry_delay': timedelta(minutes=5), }, schedule_interval='0 9 * * *', max_active_runs=1, ) as dag: postgres_to_gcs_task = PythonOperator( task_id='postgres_to_gcs', python_callable=postgres_to_gcs, provide_context=True, # Jika Anda ingin memberikan konteks eksekusi DAG ke fungsi Anda op_kwargs={ 'tables': ['locations','measurements','individuals', 'status_gizi_zscore'] }, dag=dag ) transform_balita_task = PythonOperator( task_id='transform_balita', python_callable=transform_balita, provide_context=True, # Jika Anda ingin memberikan konteks eksekusi DAG ke fungsi Anda dag=dag ) transform_measurement_task = PythonOperator( task_id='transform_measurement', python_callable=transform_measurements, provide_context=True, # Jika Anda ingin memberikan konteks eksekusi DAG ke fungsi Anda dag=dag ) transform_locations_task = PythonOperator( task_id='transform_locations', python_callable=transform_locations, provide_context=True, # Jika Anda ingin memberikan konteks eksekusi DAG ke fungsi Anda dag=dag ) transform_statusGizi_task = PythonOperator( task_id='transform_statusGizi', python_callable=transform_statusGizi, provide_context=True, # Jika Anda ingin memberikan konteks eksekusi DAG ke fungsi Anda dag=dag ) transform_date_task = PythonOperator( task_id='transform_date', python_callable=transform_date, provide_context=True, # Jika Anda ingin memberikan konteks eksekusi DAG ke fungsi Anda dag=dag ) # Menambahkan PythonOperator ke DAG create_fact_table_task = PythonOperator( task_id='create_fact_table', python_callable=fact, provide_context=True, dag=dag ) tables= ['dim_balita', 'dim_location','dim_measure', 'dim_status', 'dim_date', 'fact_gizi'] gcs_to_bq_tasks = {} for table_name in tables: gcs_to_bq_tasks[table_name] = GCSToBigQueryOperator( task_id=f'gcs_to_bq_{table_name}', bucket=GCS_BUCKET, source_objects=f'data/{table_name}.csv', destination_project_dataset_table=f'{BQ_PROJECT}.{BQ_DS}.{table_name}', #schema_fields=schema, #field_delimiter=',', create_disposition='CREATE_IF_NEEDED', write_disposition='WRITE_TRUNCATE', skip_leading_rows=1, allow_quoted_newlines=True, autodetect=True, source_format='csv', ) transform_tasks = [transform_balita_task,transform_locations_task,transform_measurement_task,transform_statusGizi_task, transform_date_task, create_fact_table_task] postgres_to_gcs_task >> transform_tasks # Then, each transformation should be followed by its corresponding BigQuery load for table_name, transform_task in zip(tables, transform_tasks): transform_task >> gcs_to_bq_tasks[table_name]