Docker Compose
version: '3'
services:
source_postgres:
image: postgres:13
ports:
- "5433:5432"
networks:
- elt_network
environment:
POSTGRES_DB: source_db
POSTGRES_USER: postgres
POSTGRES_PASSWORD: secret
volumes:
- ./data_source/init.sql:/docker-entrypoint-initdb.d/init.sql
destination_postgres:
image: postgres:13
ports:
- "5434:5432"
networks:
- elt_network
depends_on:
- source_postgres
environment:
POSTGRES_DB: destination_db
POSTGRES_USER: postgres
POSTGRES_PASSWORD: secret
postgres:
image: postgres:13
networks:
- elt_network
environment:
- POSTGRES_USER=airflow
- POSTGRES_PASSWORD=airflow
- POSTGRES_DB=airflow
init-airflow:
image: apache/airflow:2.9.2
depends_on:
- postgres
networks:
- elt_network
environment:
- AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@postgres/airflow
command: >
bash -c "airflow db init &&
airflow users create --username airflow --password password --firstname John --lastname Doe --role Admin --email [email protected]"
webserver:
image: apache/airflow:2.9.2
user: root
depends_on:
- postgres
networks:
- elt_network
extra_hosts:
- "host.docker.internal:host-gateway"
volumes:
- ./airflow/dags:/opt/airflow/dags
- ./elt_script:/opt/airflow/elt_script
- ./postgres_transformations:/opt/dbt
- ~/.dbt:/root/.dbt
- /var/run/docker.sock:/var/run/docker.sock
environment:
- LOAD_EX=n
- AIRFLOW__CORE__EXECUTOR=LocalExecutor
- AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@postgres/airflow
- AIRFLOW_CONN_DESTINATION_POSTGRES=postgres://postgres:secret@destination_postgres:5434/destination_db
- AIRFLOW__WEBSERVER__DEFAULT_USER_USERNAME=airflow
- AIRFLOW__WEBSERVER__DEFAULT_USER_PASSWORD=password
- AIRFLOW_WWW_USER_USERNAME=airflow
- AIRFLOW_WWW_USER_PASSWORD=password
- AIRFLOW__WEBSERVER__SECRET_KEY=secret
ports:
- "8080:8080"
command: webserver
scheduler:
image: apache/airflow:2.9.2
user: root
depends_on:
- postgres
networks:
- elt_network
extra_hosts:
- "host.docker.internal:host-gateway"
volumes:
- ./airflow/dags:/opt/airflow/dags
- ./elt_script:/opt/airflow/elt_script
- ./postgres_transformations:/dbt
- ~/.dbt:/root/.dbt
- /var/run/docker.sock:/var/run/docker.sock
environment:
- LOAD_EX=n
- AIRFLOW__CORE__EXECUTOR=LocalExecutor
- AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@postgres/airflow
- AIRFLOW_CONN_DESTINATION_POSTGRES=postgres://postgres:secret@destination_postgres:5434/destination_db
- AIRFLOW__CORE__FERNET_KEY=plIipb9RU3-3wJ1UNaAtqVNJrqFEks1-dGbJM34EW7U=
- AIRFLOW__WEBSERVER__SECRET_KEY=secret
- AIRFLOW_WWW_USER_USERNAME=airflow
- AIRFLOW_WWW_USER_PASSWORD=password
command: scheduler
networks:
elt_network:
driver: bridge
Init SQL
-- Tabel pelanggan
CREATE TABLE customers (
customer_id SERIAL PRIMARY KEY,
name VARCHAR(100),
email VARCHAR(100),
city VARCHAR(50)
);
-- Tabel produk
CREATE TABLE products (
product_id SERIAL PRIMARY KEY,
product_name VARCHAR(100),
price NUMERIC(10, 2),
stock INT
);
-- Tabel transaksi langsung
CREATE TABLE transaction_items (
item_id SERIAL PRIMARY KEY,
customer_id INT REFERENCES customers(customer_id),
product_id INT REFERENCES products(product_id),
quantity INT,
transaction_date DATE
);
-- Data customers
INSERT INTO customers (name, email, city) VALUES
('Rina', '[email protected]', 'Jakarta'),
('Andi', '[email protected]', 'Bandung'),
('Sari', '[email protected]', 'Surabaya'),
('Budi', '[email protected]', 'Medan'),
('Lina', '[email protected]', 'Yogyakarta'),
('Dika', '[email protected]', 'Bekasi'),
('Eka', '[email protected]', 'Depok'),
('Yudi', '[email protected]', 'Semarang'),
('Nina', '[email protected]', 'Bogor'),
('Tono', '[email protected]', 'Makassar');
-- Data products
INSERT INTO products (product_name, price, stock) VALUES
('Sabun Mandi', 5000, 100),
('Shampoo', 12000, 80),
('Pasta Gigi', 8000, 90),
('Sikat Gigi', 7000, 85),
('Minyak Goreng 1L', 15000, 70),
('Beras 5kg', 60000, 60),
('Gula Pasir 1kg', 13000, 75),
('Kopi Instan', 10000, 50),
('Teh Celup', 9000, 65),
('Mie Instan', 3000, 200);
-- Data transaction_items
INSERT INTO transaction_items (customer_id, product_id, quantity, transaction_date) VALUES
(1, 1, 2, '2025-05-01'),
(1, 2, 1, '2025-05-01'),
(2, 3, 1, '2025-05-01'),
(2, 5, 2, '2025-05-01'),
(3, 6, 1, '2025-05-02'),
(3, 4, 1, '2025-05-02'),
(4, 2, 1, '2025-05-02'),
(4, 7, 1, '2025-05-02'),
(5, 5, 1, '2025-05-03'),
(5, 1, 2, '2025-05-03'),
(6, 3, 3, '2025-05-03'),
(6, 10, 5, '2025-05-03'),
(7, 2, 2, '2025-05-03'),
(7, 4, 1, '2025-05-03'),
(8, 6, 1, '2025-05-04'),
(8, 9, 2, '2025-05-04'),
(9, 3, 1, '2025-05-04'),
(9, 2, 1, '2025-05-04'),
(10, 8, 1, '2025-05-04'),
(10, 10, 3, '2025-05-04'),
(1, 7, 1, '2025-05-05'),
(1, 1, 1, '2025-05-05'),
(2, 5, 2, '2025-05-05'),
(2, 3, 1, '2025-05-05'),
(3, 4, 1, '2025-05-06'),
(3, 9, 2, '2025-05-06'),
(4, 6, 1, '2025-05-06'),
(4, 8, 1, '2025-05-06'),
(5, 10, 4, '2025-05-06'),
(5, 2, 2, '2025-05-06'),
(3, 1, 1, '2025-05-02'),
(5, 4, 1, '2025-05-03'),
(6, 9, 2, '2025-05-03'),
(9, 1, 1, '2025-05-04'),
(10, 3, 1, '2025-05-04'),
(3, 10, 2, '2025-05-06'),
(4, 7, 1, '2025-05-06'),
(5, 6, 1, '2025-05-06'),
(8, 3, 2, '2025-05-04'),
(8, 5, 1, '2025-05-04');
Dump & Load Data Function
import subprocess
print("Transfering data....")
# Use pg_dump to dump the source database to a SQL file
dump_command = [
'pg_dump',
'-h', 'source_postgres',
'-U', 'postgres',
'-d', 'source_db',
'-f', 'data_dump.sql',
'-w' # Do not prompt for password
]
# Set the PGPASSWORD environment variable to avoid password prompt
subprocess.run(dump_command, env={'PGPASSWORD': 'secret'}, check=True)
# Use psql to load the dumped SQL file into the destination database
load_command = [
'psql',
'-h', 'destination_postgres',
'-U', 'postgres',
'-d', 'destination_db',
'-a', '-f', 'data_dump.sql'
]
# Set the PGPASSWORD environment variable for the destination database
subprocess.run(load_command, env={'PGPASSWORD': 'secret'}, check=True)
print("Ending transfer...")
DAG Load dan Transfer Data
from airflow import DAG
from datetime import datetime
from airflow.operators.python_operator import PythonOperator
import subprocess
def dump_load_to_destinationdb():
script_path = '/opt/airflow/elt_script/dump_and_load_data.py'
result = subprocess.run(['python', script_path], capture_output = True, text=True)
if result.returncode != 0:
raise Exception(f"Script failed with error: {result.stderr}")
else:
print(result.stdout)
default_args= {
'owner': 'risdan',
'depends_on_past': False
}
dag = DAG(
"dag_load_and_transfer",
default_args = default_args,
start_date = datetime(2025,5,8),
catchup = False,
schedule_interval = '0 1 * * 0'
)
task_load_and_tranfer_data = PythonOperator(
task_id = "load_data_to_destinationdb",
python_callable=dump_load_to_destinationdb,
dag = dag
)
DAG dbt
from airflow import DAG
from datetime import datetime
from airflow.providers.docker.operators.docker import DockerOperator
from docker.types import Mount
default_args= {
'owner': 'risdan',
'depends_on_past': False
}
dag = DAG(
"dag_dbt",
default_args = default_args,
start_date = datetime(2025,5,8),
catchup = False,
schedule_interval = None
)
task_transforms_dbt = DockerOperator(
task_id='transformation_using_dbt',
image='ghcr.io/dbt-labs/dbt-postgres:1.4.7',
command=[
"run",
"--profiles-dir",
"/root",
"--project-dir",
"/dbt",
"--full-refresh"
],
auto_remove=True,
# tty=True,
docker_url="unix://var/run/docker.sock",
network_mode="bridge",
mounts=[
Mount(source='D:/belajar_data_engineering/project_de_2_elt/postgres_transformations',
target='/dbt', type='bind'),
Mount(source='C:/Users/Acer/.dbt', target='/root', type='bind'),
],
dag=dag
)
dbt Profile
postgres_transformations:
outputs:
dev:
dbname: destination_db
host: host.docker.internal
pass: secret
port: 5434
schema: public
threads: 1
type: postgres
user: postgres
target: dev
dbt Schema
version: 2
models:
- name: films
description: "This table contains details about films."
columns:
- name: film_id
description: "Unique identifier for the film."
tests:
- unique
- not_null
- name: title
description: "Title of the film."
tests:
- not_null
- name: release_date
description: "Date the film was released."
tests:
- not_null
- name: price
description: "Price to buy a ticket."
tests:
- not_null
- name: rating
description: "The MPAA rating of the film."
tests:
- not_null
- name: user_rating
description: "Rating that users gave the film."
tests:
- not_null
- name: actors
description: "This table contains details about actors."
columns:
- name: actor_id
description: "Unique identifier for the actor."
tests:
- unique
- not_null
- name: name
description: "First name of the actor."
tests:
- name: film_actors
description: "This table maps films to the actors that starred in them."
columns:
- name: film_id
description: "Identifier for the film."
tests:
- not_null
- name: actor_id
description: "Identifier for the actor."
tests:
- not_null