- ETL Module

The ETL module is responsible for inserting preprocessed and embedded movie .json files into Elasticsearch.

- Database (NoSQL - Elasticsearch)

In load.py, we use the elasticsearch Python client to:

• Connect to the Elasticsearch server (with retry logic)
• Create an index (if it doesn't exist)
• Insert the .json documents from the data folder
• Each document includes:
• movie_id
• genres
• type (short, summary, or long)
• chunk_id
• text
• vector (768-dim dense vector)

Indexing uses cosine similarity to support semantic search.

---

- Docker Setup

The ETL service is containerized and configured to run a run_etl.sh script which:

1. Waits for a .zip file of JSON chunks to appear in the data/jsons/ folder.

app/
├── etl/
│ └── data/
│ │ └── jsons/
│ │  ├── move_data.zip
│ └── ...
└──

1. Unzips and flattens the directory structure.
2. Executes load.py to push the content into Elasticsearch.
3. Cleans and archives the .zip file after successful upload.

- Dockerfile

FROM python:3.10-slim-bullseye

# Install system dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
    unzip \
    build-essential libpq-dev libfreetype6-dev libpng-dev libjpeg-dev \
    libblas-dev liblapack-dev gfortran \
    && apt-get clean && rm -rf /var/lib/apt/lists/*

# Set working directory
WORKDIR /etl

# Copy requirements file and install dependencies
COPY requirements.txt .
RUN pip3 install --upgrade pip && pip3 install --no-cache-dir -r requirements.txt

# Copy the rest of the code
COPY . .

# Make the run script executable
RUN chmod +x run_etl.sh

# Run the ETL process
CMD ["sh", "run_etl.sh"]

---

get_elasticsearch()

Creates and returns an Elasticsearch client instance.

Returns:

Name	Type	Description
Elasticsearch		An instance of the Elasticsearch client connected to the URL specified in the environment variable ELASTICSEARCH_URL.

Source code in app/etl/config.py

def get_elasticsearch():
    """
    Creates and returns an Elasticsearch client instance.

    Returns:
        Elasticsearch: An instance of the Elasticsearch client connected to the URL specified in the environment variable `ELASTICSEARCH_URL`.
    """
    return Elasticsearch(ES_URL)

connect_to_elasticsearch(retries=10, delay=2)

Attempts to connect to Elasticsearch with retry logic.

Parameters:

Name	Type	Description	Default
retries	int	Number of retry attempts.	10
delay	int	Delay in seconds between attempts.	2

Returns:

Name	Type	Description
Elasticsearch		An instance of the Elasticsearch client.

Raises:

Type	Description
ConnectionError	If connection fails after all attempts.

Source code in app/etl/load.py

def connect_to_elasticsearch(retries: int = 10, delay: int = 2):
    """
    Attempts to connect to Elasticsearch with retry logic.

    Args:
        retries (int): Number of retry attempts.
        delay (int): Delay in seconds between attempts.

    Returns:
        Elasticsearch: An instance of the Elasticsearch client.

    Raises:
        ConnectionError: If connection fails after all attempts.
    """
    for i in range(retries):
        try:
            es = get_elasticsearch()
            version = es.info()["version"]["number"]
            logger.info(f"- Connected to Elasticsearch {version}")
            return es
        except Exception as e:
            logger.info(f"- Attempt {i+1}/{retries}: {e}")
            time.sleep(delay)
    raise ConnectionError("- Elasticsearch did not become ready in time")

create_index(es)

Creates an Elasticsearch index with the required mappings if it doesn't exist.

Parameters:

Name	Type	Description	Default
es	Elasticsearch	Elasticsearch client.	required

Source code in app/etl/load.py

def create_index(es):
    """
    Creates an Elasticsearch index with the required mappings if it doesn't exist.

    Args:
        es (Elasticsearch): Elasticsearch client.
    """
    if not es.indices.exists(index=INDEX_NAME):
        es.indices.create(
            index=INDEX_NAME,
            mappings={
                "properties": {
                    "movie_id": {"type": "keyword"},  # Unique identifier for the movie
                    "genres": {"type": "keyword"},  # List of genres associated with the movie
                    "type": {"type": "keyword"},  # Type of text (e.g., "short", "summary", "long")   
                    "chunk_id": {"type": "keyword"},  # Identifier for text chunks
                    "text": {"type": "text"},  # Text content of the document
                    "vector": {
                        "type": "dense_vector",  # Dense vector for similarity search
                        "dims": VECTOR_DIM,  # Dimensionality of the vector
                        "index": True,  # Enable indexing for the vector
                        "similarity": "cosine"  # Use cosine similarity for vector comparison
                    }
                }
            }
        )
        logger.info(f"- Created index: {INDEX_NAME}")
    else:
        logger.info(f"- Index '{INDEX_NAME}' already exists")

upload_documents(es)

Uploads JSON documents from the TO_INSERT_DIR to Elasticsearch.

Parameters:

Name	Type	Description	Default
es	Elasticsearch	Elasticsearch client.	required

Source code in app/etl/load.py

def upload_documents(es):
    """
    Uploads JSON documents from the `TO_INSERT_DIR` to Elasticsearch.

    Args:
        es (Elasticsearch): Elasticsearch client.
    """
    uploaded_count = 0
    for filename in os.listdir(TO_INSERT_DIR):
        if filename.endswith(".json"):
            filepath = os.path.join(TO_INSERT_DIR, filename)
            try:
                with open(filepath, "r", encoding="utf-8") as f:
                    data = json.load(f)
                    es.index(index=INDEX_NAME, id=filename, document=data)
                    uploaded_count += 1
            except Exception as e:
                logger.info(f"- Failed to upload {filename}: {e}")
    logger.info(f"- Uploaded {uploaded_count} JSON file(s) to index '{INDEX_NAME}'")