Harvard Artifacts Collection Data Engineering

Skill by ara.so — Data Skills collection.

This project provides an end-to-end data engineering and analytics application built on the Harvard Art Museums API. It demonstrates real-world ETL pipelines, SQL database design, analytical queries, and interactive visualization using Streamlit. The architecture follows: API → ETL → SQL → Analytics → Visualization.

What This Project Does

• API Integration: Fetches artifact data from Harvard Art Museums API with pagination and rate limiting

• ETL Pipeline: Extracts, transforms, and loads nested JSON into relational database tables

• SQL Database: Stores structured data across artifactmetadata, artifactmedia, and artifactcolors tables

• Analytics: Executes 20+ predefined SQL queries for insights

• Visualization: Interactive dashboards using Plotly and Streamlit

Installation

# Clone the repository

git clone https://github.com/Manali0711/Harvard-Artifacts-Collection-Data-Engineering-Analytics-App.git

cd Harvard-Artifacts-Collection-Data-Engineering-Analytics-App

# Install dependencies

pip install -r requirements.txt

Required packages:

streamlit

pandas

requests

mysql-connector-python

plotly

Configuration

Environment Variables

Set up your configuration before running:

# Harvard Art Museums API Key (get from https://www.harvardartmuseums.org/collections/api)

export HARVARD_API_KEY="your_api_key_here"

# Database credentials

export DB_HOST="your_database_host"

export DB_PORT="3306"

export DB_USER="your_database_user"

export DB_PASSWORD="your_database_password"

export DB_NAME="harvard_artifacts"

Database Setup

Create the required tables:

CREATE DATABASE harvard_artifacts;

USE harvard_artifacts;

CREATE TABLE artifactmetadata (

    id INT PRIMARY KEY,

    title VARCHAR(500),

    culture VARCHAR(255),

    century VARCHAR(100),

    classification VARCHAR(255),

    division VARCHAR(255),

    department VARCHAR(255),

    technique VARCHAR(500),

    period VARCHAR(255),

    dated VARCHAR(255),

    url TEXT,

    lastupdate DATETIME

);

CREATE TABLE artifactmedia (

    media_id INT AUTO_INCREMENT PRIMARY KEY,

    artifact_id INT,

    baseimageurl TEXT,

    iiifbaseuri TEXT,

    FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)

);

CREATE TABLE artifactcolors (

    color_id INT AUTO_INCREMENT PRIMARY KEY,

    artifact_id INT,

    color VARCHAR(50),

    spectrum VARCHAR(50),

    percentage FLOAT,

    FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)

);

Key Components and Usage

1. API Data Extraction

import requests

import os

class HarvardAPIClient:

    def __init__(self):

        self.api_key = os.getenv('HARVARD_API_KEY')

        self.base_url = "https://api.harvardartmuseums.org/object"

    def fetch_artifacts(self, page=1, size=100):

        """Fetch artifacts with pagination"""

        params = {

            'apikey': self.api_key,

            'page': page,

            'size': size,

            'hasimage': 1  # Only artifacts with images

        }

        response = requests.get(self.base_url, params=params)

        response.raise_for_status()

        return response.json()

    def fetch_multiple_pages(self, num_pages=10):

        """Fetch multiple pages of artifacts"""

        all_artifacts = []

        for page in range(1, num_pages + 1):

            data = self.fetch_artifacts(page=page)

            all_artifacts.extend(data.get('records', []))

        return all_artifacts

2. ETL Pipeline Implementation

import pandas as pd

import mysql.connector

from typing import List, Dict

class ArtifactETL:

    def __init__(self, db_config):

        self.db_config = db_config

        self.conn = None

    def connect_db(self):

        """Establish database connection"""

        self.conn = mysql.connector.connect(

            host=self.db_config['host'],

            port=self.db_config['port'],

            user=self.db_config['user'],

            password=self.db_config['password'],

            database=self.db_config['database']

        )

        return self.conn

    def transform_artifacts(self, raw_data: List[Dict]) -> pd.DataFrame:

        """Transform raw JSON to structured DataFrame"""

        artifacts = []

        for item in raw_data:

            artifact = {

                'id': item.get('id'),

                'title': item.get('title', '')[:500],

                'culture': item.get('culture', ''),

                'century': item.get('century', ''),

                'classification': item.get('classification', ''),

                'division': item.get('division', ''),

                'department': item.get('department', ''),

                'technique': item.get('technique', ''),

                'period': item.get('period', ''),

                'dated': item.get('dated', ''),

                'url': item.get('url', ''),

                'lastupdate': item.get('lastupdate', '')

            }

            artifacts.append(artifact)

        return pd.DataFrame(artifacts)

    def transform_media(self, raw_data: List[Dict]) -> pd.DataFrame:

        """Extract media information"""

        media_records = []

        for item in raw_data:

            artifact_id = item.get('id')

            if 'primaryimageurl' in item:

                media_records.append({

                    'artifact_id': artifact_id,

                    'baseimageurl': item.get('primaryimageurl', ''),

                    'iiifbaseuri': item.get('iiifbaseuri', '')

                })

        return pd.DataFrame(media_records)

    def transform_colors(self, raw_data: List[Dict]) -> pd.DataFrame:

        """Extract color information"""

        color_records = []

        for item in raw_data:

            artifact_id = item.get('id')

            colors = item.get('colors', [])

            for color in colors:

                color_records.append({

                    'artifact_id': artifact_id,

                    'color': color.get('color', ''),

                    'spectrum': color.get('spectrum', ''),

                    'percentage': color.get('percent', 0)

                })

        return pd.DataFrame(color_records)

    def load_data(self, df: pd.DataFrame, table_name: str):

        """Batch insert data into SQL table"""

        cursor = self.conn.cursor()

        # Prepare column names and placeholders

        cols = ','.join(df.columns)

        placeholders = ','.join(['%s'] * len(df.columns))

        sql = f"INSERT INTO {table_name} ({cols}) VALUES ({placeholders})"

        # Convert DataFrame to list of tuples

        data = [tuple(row) for row in df.values]

        # Execute batch insert

        cursor.executemany(sql, data)

        self.conn.commit()

        cursor.close()

        return len(data)

3. Complete ETL Workflow

def run_etl_pipeline():

    """Execute complete ETL pipeline"""

    # Configuration

    db_config = {

        'host': os.getenv('DB_HOST'),

        'port': int(os.getenv('DB_PORT', 3306)),

        'user': os.getenv('DB_USER'),

        'password': os.getenv('DB_PASSWORD'),

        'database': os.getenv('DB_NAME')

    }

    # Initialize clients

    api_client = HarvardAPIClient()

    etl = ArtifactETL(db_config)

    etl.connect_db()

    # Extract

    print("Extracting data from API...")

    raw_artifacts = api_client.fetch_multiple_pages(num_pages=5)

    print(f"Extracted {len(raw_artifacts)} artifacts")

    # Transform

    print("Transforming data...")

    df_artifacts = etl.transform_artifacts(raw_artifacts)

    df_media = etl.transform_media(raw_artifacts)

    df_colors = etl.transform_colors(raw_artifacts)

    # Load

    print("Loading data to database...")

    etl.load_data(df_artifacts, 'artifactmetadata')

    etl.load_data(df_media, 'artifactmedia')

    etl.load_data(df_colors, 'artifactcolors')

    print("ETL pipeline completed successfully!")

    etl.conn.close()

4. SQL Analytics Queries

class ArtifactAnalytics:

    def __init__(self, db_config):

        self.db_config = db_config

    def execute_query(self, query: str) -> pd.DataFrame:

        """Execute SQL query and return DataFrame"""

        conn = mysql.connector.connect(**self.db_config)

        df = pd.read_sql(query, conn)

        conn.close()

        return df

    def get_artifacts_by_culture(self):

        """Get artifact distribution by culture"""

        query = """

        SELECT culture, COUNT(*) as artifact_count

        FROM artifactmetadata

        WHERE culture IS NOT NULL AND culture != ''

        GROUP BY culture

        ORDER BY artifact_count DESC

        LIMIT 20;

        """

        return self.execute_query(query)

    def get_artifacts_by_century(self):

        """Get artifact distribution by century"""

        query = """

        SELECT century, COUNT(*) as count

        FROM artifactmetadata

        WHERE century IS NOT NULL AND century != ''

        GROUP BY century

        ORDER BY count DESC;

        """

        return self.execute_query(query)

    def get_color_distribution(self):

        """Get color usage across artifacts"""

        query = """

        SELECT color, COUNT(*) as frequency, AVG(percentage) as avg_percentage

        FROM artifactcolors

        GROUP BY color

        ORDER BY frequency DESC

        LIMIT 15;

        """

        return self.execute_query(query)

    def get_artifacts_with_media(self):

        """Get artifacts with and without media"""

        query = """

        SELECT

            CASE WHEN m.artifact_id IS NOT NULL THEN 'With Media' ELSE 'Without Media' END as media_status,

            COUNT(*) as count

        FROM artifactmetadata a

        LEFT JOIN artifactmedia m ON a.id = m.artifact_id

        GROUP BY media_status;

        """

        return self.execute_query(query)

5. Streamlit Dashboard

import streamlit as st

import plotly.express as px

def create_dashboard():

    st.title("Harvard Art Museums Analytics Dashboard")

    # Sidebar for navigation

    st.sidebar.header("Navigation")

    option = st.sidebar.selectbox(

        "Choose Analysis",

        ["Overview", "Culture Analysis", "Century Distribution", "Color Patterns", "Media Analysis"]

    )

    # Initialize analytics

    db_config = {

        'host': os.getenv('DB_HOST'),

        'port': int(os.getenv('DB_PORT', 3306)),

        'user': os.getenv('DB_USER'),

        'password': os.getenv('DB_PASSWORD'),

        'database': os.getenv('DB_NAME')

    }

    analytics = ArtifactAnalytics(db_config)

    if option == "Culture Analysis":

        st.header("Artifacts by Culture")

        df = analytics.get_artifacts_by_culture()

        # Display data table

        st.dataframe(df)

        # Create visualization

        fig = px.bar(df, x='culture', y='artifact_count',

                     title='Top 20 Cultures by Artifact Count')

        st.plotly_chart(fig)

    elif option == "Century Distribution":

        st.header("Artifacts by Century")

        df = analytics.get_artifacts_by_century()

        st.dataframe(df)

        fig = px.bar(df, x='century', y='count',

                     title='Artifact Distribution Across Centuries')

        st.plotly_chart(fig)

    elif option == "Color Patterns":

        st.header("Color Usage Analysis")

        df = analytics.get_color_distribution()

        st.dataframe(df)

        fig = px.bar(df, x='color', y='frequency',

                     title='Most Common Colors in Artifacts')

        st.plotly_chart(fig)

if __name__ == "__main__":

    create_dashboard()

Running the Application

# Start the Streamlit dashboard

streamlit run app.py

# Run ETL pipeline separately (if needed)

python etl_pipeline.py

Common Patterns

Incremental Data Loading

def incremental_load(last_update_date):

    """Load only new or updated artifacts"""

    query = f"""

    SELECT id FROM artifactmetadata

    WHERE lastupdate > '{last_update_date}'

    """

    # Fetch only updated records from API

    # Update existing records in database

Error Handling in ETL

def safe_etl_run():

    try:

        run_etl_pipeline()

    except requests.HTTPError as e:

        print(f"API Error: {e}")

    except mysql.connector.Error as e:

        print(f"Database Error: {e}")

    except Exception as e:

        print(f"Unexpected Error: {e}")

Troubleshooting

API Rate Limiting: Add delays between requests

import time

time.sleep(0.5)  # 500ms delay between API calls

Database Connection Issues: Verify credentials and network access

# Test connection

try:

    conn = mysql.connector.connect(**db_config)

    print("Database connection successful")

    conn.close()

except Exception as e:

    print(f"Connection failed: {e}")

Missing Data Fields: Handle null values during transformation

artifact = {

    'title': item.get('title', 'Unknown')[:500],

    'culture': item.get('culture') or 'Not Specified'

}

Memory Issues with Large Datasets: Use chunking

chunk_size = 1000

for i in range(0, len(data), chunk_size):

    chunk = data[i:i+chunk_size]

    etl.load_data(pd.DataFrame(chunk), 'artifactmetadata')