Pastikan dahulu python sudah terinstall di dalam komputer/laptop masing-masing, jika belum silahkan download dahulu python 3.12.10 disini.

Di dalam project ini kita akan mengextract data dari API Open Library, melakukan transformasi (pembersihan data) dan memasukannya ke dalam postgree database. Kita akan menggunakan apache airflow sebagai orchestrator dan seluruh project ini akan berjalan di dalam docker.

Membuat Virtual Environment

Virtual Environment (virtual env) dalam Python adalah sebuah lingkungan terisolasi yang digunakan untuk mengelola dependensi (library atau package) yang dibutuhkan oleh sebuah proyek Python. Dengan menggunakan virtual environment, setiap proyek bisa memiliki versi pustaka yang berbeda-beda tanpa saling mengganggu.

Python: Ekstraksi-Cleaning-Transform

import requests

# Ekstraksi data dari API
query = "Data Engineering"
url = f"https://openlibrary.org/search.json?q={query.replace(' ', '+')}"
response = requests.get(url)
data = response.json()

# Pembersihan dan Transformasi Data
books = []

for book in data['docs'][:10]:
    # print(book.get('title'))
    # print(book.get('author_name')[0])
    # print(book.get('first_publish_year'))
    # print('======')
    books.append({
        'title':book.get('title'),
        'author_name':book.get('author_name')[0],
        'first_publish_year':book.get('first_publish_year')
    })

for book in books:
    print(book)
    print('=====')

Docker Compose:

version: '3'

services:
  airflow:
    image: apache/airflow:2.9.2
    ports:
      - "8080:8080"
    volumes:
      - ./airflow:/opt/airflow
    command: >
      bash -c "rm -f /opt/airflow/airflow-webserver.pid && airflow standalone"

  postgres:
    image: postgres:13
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: airflow
    volumes:
      - postgres-db-volume:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD", "pg_isready", "-U", "airflow"]
      interval: 10s
      retries: 5
      start_period: 5s
    restart: always
    ports:
      - "5432:5432"

volumes:
  postgres-db-volume:

SQL: Create Table

CREATE TABLE IF NOT EXISTS books (
    id SERIAL,
    title TEXT NOT NULL,
    author_name TEXT,
    first_publish_year TEXT
)

Python: DAG ETL

from airflow import DAG
from datetime import datetime, timedelta
import requests
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook

def extract_and_cleaning_data(ti):
    query = "Data Engineering"
    url = f"https://openlibrary.org/search.json?q={query.replace(' ', '+')}"

    response = requests.get(url)
    data = response.json()

    books = []

    for book in data['docs'][:10]:
        books.append({
            'title':book.get('title'),
            'author_name':book.get('author_name')[0],
            'first_publish_year':book.get('first_publish_year')
        })
    
    ti.xcom_push(key='book_data', value=books)

def insert_data(ti):
    books = ti.xcom_pull(key='book_data', task_ids= 'extract_cleaning_data')

    postgres_hook = PostgresHook(postgres_conn_id = 'books_connection')

    # {'title': 'Fundamentals of Data Engineering', 'author_name': 'Joe Reis', 'first_publish_year': 2022}
    
    insert_query = """
        INSERT INTO books (title, author_name, first_publish_year)
        VALUES (%s, %s, %s)
    """

    for book in books:
        parameter_book = (book['title'], book['author_name'], book['first_publish_year'])
        postgres_hook.run(insert_query, parameters = parameter_book)

default_args = {
    'owner':'risdan',
    'start_date':datetime(2025,5,4),
    'retries':0
}

dag = DAG(
    'dag_project_de_etl_v04',
    default_args = default_args,
    schedule_interval = None
)

task_1 = PythonOperator(
    task_id = 'extract_cleaning_data',
    python_callable = extract_and_cleaning_data,
    dag=dag
)

task_2 = PostgresOperator(
    task_id ='create_table',
    postgres_conn_id = 'books_connection',
    sql = "./files/create_table.sql",
    dag = dag
)

task_3 = PythonOperator(
    task_id='insert_data',
    python_callable= insert_data,
    dag=dag
)

task_1 >> task_2 >> task_3

Project 2: ELT Pipeline