harvard-art-museum-data-pipeline
Build ETL pipelines and analytics dashboards using the Harvard Art Museums API with Streamlit, MySQL, and Python
ソース情報
- リポジトリ
- reason-machines/data-skills
- ソースの最終更新活動
- 2026年6月27日 09:22
- 検出された SKILL.md の言語
- 英語
- スター
- 5
- フォーク
- 1
インストール方法
デフォルトでは、最初にソースを確認する Prompt が選択されています。直接コマンドに切り替えるか、ローカルコピーをダウンロードすることもできます。
ソースファイルを確認
インストールを決める前に、SKILL.md と SkillsMP に表示されている付属ファイルをお読みください。
SKILL.md を表示中
SKILL.md
ソースの指示 · 読み取り専用プレビュー- name
- harvard-art-museum-data-pipeline
- description
- Build ETL pipelines and analytics dashboards using the Harvard Art Museums API with Streamlit, MySQL, and Python
- triggers
- ["how do I set up a data pipeline for Harvard Art Museums API","build an ETL workflow for museum artifact data","create analytics dashboard with Harvard Art Museums data","extract and transform Harvard Art Museums API data","set up SQL database for museum artifacts collection","visualize Harvard Art Museums data with Streamlit","query Harvard Art Museums API with pagination","design relational schema for museum artifact data"]
# Harvard Art Museum Data Pipeline
> Skill by [ara.so](https://ara.so) — Data Skills collection.
This project provides an end-to-end data engineering solution for collecting, transforming, storing, and analyzing artifact data from the Harvard Art Museums API. It demonstrates production-ready ETL pipelines, SQL analytics, and interactive visualization using Streamlit.
## What It Does
The Harvard Art Museum Data Pipeline:
- **Extracts** artifact data from the Harvard Art Museums API with pagination and rate limiting
- **Transforms** nested JSON into normalized relational tables (metadata, media, colors)
- **Loads** data into MySQL/TiDB Cloud with batch inserts for performance
- **Analyzes** data using predefined SQL queries for business insights
- **Visualizes** results through interactive Streamlit dashboards with Plotly charts
## Architecture
```
Harvard Art Museums API → Python ETL → MySQL/TiDB → SQL Analytics → Streamlit Dashboard
```
**Key Components:**
- API integration with secure key management
- Three-table relational schema: `artifactmetadata`, `artifactmedia`, `artifactcolors`
- 20+ analytical SQL queries
- Real-time interactive visualizations
## Installation
### Prerequisites
- Python 3.8+
- MySQL or TiDB Cloud account
- Harvard Art Museums API key (obtain from https://harvardartmuseums.org/collections/api)
### Setup
```bash
# 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
# Set up environment variables
export HARVARD_API_KEY="your_api_key_here"
export DB_HOST="your_database_host"
export DB_USER="your_database_user"
export DB_PASSWORD="your_database_password"
export DB_NAME="harvard_artifacts"
```
### Required Dependencies
```txt
streamlit
pandas
requests
mysql-connector-python
plotly
python-dotenv
```
## Configuration
### Database Connection
Configure your MySQL/TiDB connection in your application:
```python
import mysql.connector
import os
def get_db_connection():
"""Create database connection using environment variables"""
return mysql.connector.connect(
host=os.getenv('DB_HOST'),
user=os.getenv('DB_USER'),
password=os.getenv('DB_PASSWORD'),
database=os.getenv('DB_NAME')
)
```
### API Configuration
Set up the Harvard Art Museums API client:
```python
import os
import requests
class HarvardMuseumAPI:
def __init__(self):
self.api_key = os.getenv('HARVARD_API_KEY')
self.base_url = "https://api.harvardartmuseums.org"
def get_objects(self, page=1, size=100):
"""Fetch objects with pagination"""
url = f"{self.base_url}/object"
params = {
'apikey': self.api_key,
'page': page,
'size': size
}
response = requests.get(url, params=params)
response.raise_for_status()
return response.json()
```
## Database Schema
### Create Tables
```sql
-- Artifact Metadata
CREATE TABLE artifactmetadata (
id INT PRIMARY KEY,
title VARCHAR(500),
culture VARCHAR(255),
century VARCHAR(100),
classification VARCHAR(255),
department VARCHAR(255),
technique VARCHAR(500),
medium VARCHAR(500),
dated VARCHAR(255),
accession_number VARCHAR(100),
url VARCHAR(500)
);
-- Artifact Media
CREATE TABLE artifactmedia (
media_id INT AUTO_INCREMENT PRIMARY KEY,
artifact_id INT,
image_url VARCHAR(1000),
alt_text TEXT,
FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)
);
-- Artifact Colors
CREATE TABLE artifactcolors (
color_id INT AUTO_INCREMENT PRIMARY KEY,
artifact_id INT,
color_name VARCHAR(100),
hex_code VARCHAR(10),
percentage DECIMAL(5,2),
FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)
);
```
## ETL Pipeline
### Extract Phase
```python
import requests
import time
def extract_artifacts(api_key, num_pages=10, page_size=100):
"""
Extract artifacts from Harvard Art Museums API with pagination
"""
base_url = "https://api.harvardartmuseums.org/object"
all_records = []
for page in range(1, num_pages + 1):
params = {
'apikey': api_key,
'page': page,
'size': page_size
}
try:
response = requests.get(base_url, params=params)
response.raise_for_status()
data = response.json()
if 'records' in data:
all_records.extend(data['records'])
# Rate limiting
time.sleep(0.5)
except requests.exceptions.RequestException as e:
print(f"Error on page {page}: {e}")
continue
return all_records
```
### Transform Phase
```python
import pandas as pd
def transform_metadata(records):
"""Transform artifact records into metadata DataFrame"""
metadata = []
for record in records:
metadata.append({
'id': record.get('id'),
'title': record.get('title', '')[:500],
'culture': record.get('culture', '')[:255],
'century': record.get('century', '')[:100],
'classification': record.get('classification', '')[:255],
'department': record.get('department', '')[:255],
'technique': record.get('technique', '')[:500],
'medium': record.get('medium', '')[:500],
'dated': record.get('dated', '')[:255],
'accession_number': record.get('accessionyear', '')[:100],
'url': record.get('url', '')[:500]
})
return pd.DataFrame(metadata)
def transform_media(records):
"""Extract media information from artifacts"""
media = []
for record in records:
artifact_id = record.get('id')
images = record.get('images', [])
for image in images:
media.append({
'artifact_id': artifact_id,
'image_url': image.get('baseimageurl', '')[:1000],
'alt_text': image.get('alttext', '')
})
return pd.DataFrame(media)
def transform_colors(records):
"""Extract color data from artifacts"""
colors = []
for record in records:
artifact_id = record.get('id')
color_list = record.get('colors', [])
for color in color_list:
colors.append({
'artifact_id': artifact_id,
'color_name': color.get('color', '')[:100],
'hex_code': color.get('hex', '')[:10],
'percentage': color.get('percent', 0.0)
})
return pd.DataFrame(colors)
```
### Load Phase
```python
def load_to_database(df, table_name, connection):
"""
Batch insert DataFrame into MySQL table
"""
cursor = connection.cursor()
# Generate INSERT statement
columns = ', '.join(df.columns)
placeholders = ', '.join(['%s'] * len(df.columns))
insert_query = f"INSERT IGNORE INTO {table_name} ({columns}) VALUES ({placeholders})"
# Batch insert
data_tuples = [tuple(row) for row in df.values]
cursor.executemany(insert_query, data_tuples)
connection.commit()
print(f"Inserted {cursor.rowcount} rows into {table_name}")
cursor.close()
```
### Complete ETL Workflow
```python
def run_etl_pipeline(api_key, db_connection, num_pages=10):
"""
Execute complete ETL pipeline
"""
# Extract
print("Extracting data from API...")
records = extract_artifacts(api_key, num_pages=num_pages)
# Transform
print("Transforming data...")
metadata_df = transform_metadata(records)
media_df = transform_media(records)
colors_df = transform_colors(records)
# Load
print("Loading data to database...")
load_to_database(metadata_df, 'artifactmetadata', db_connection)
load_to_database(media_df, 'artifactmedia', db_connection)
load_to_database(colors_df, 'artifactcolors', db_connection)
print("ETL pipeline completed successfully!")
```
## SQL Analytics Queries
### Common Analytics Patterns
```python
# Top cultures by artifact count
query_cultures = """
SELECT culture, COUNT(*) as artifact_count
FROM artifactmetadata
WHERE culture IS NOT NULL AND culture != ''
GROUP BY culture
ORDER BY artifact_count DESC
LIMIT 10;
"""
# Artifacts by century
query_centuries = """
SELECT century, COUNT(*) as count
FROM artifactmetadata
WHERE century IS NOT NULL
GROUP BY century
ORDER BY count DESC;
"""
# Media availability analysis
query_media = """
SELECT
CASE
WHEN EXISTS (SELECT 1 FROM artifactmedia WHERE artifactmedia.artifact_id = artifactmetadata.id)
THEN 'Has Media'
ELSE 'No Media'
END as media_status,
COUNT(*) as count
FROM artifactmetadata
GROUP BY media_status;
"""
# Color distribution
query_colors = """
SELECT color_name, COUNT(*) as usage_count, AVG(percentage) as avg_percentage
FROM artifactcolors
GROUP BY color_name
ORDER BY usage_count DESC
LIMIT 15;
"""
# Department breakdown
query_departments = """
SELECT department, COUNT(*) as artifact_count
FROM artifactmetadata
WHERE department IS NOT NULL
GROUP BY department
ORDER BY artifact_count DESC;
"""
```
## Streamlit Dashboard
### Main Application Structure
```python
import streamlit as st
import pandas as pd
import plotly.express as px
def main():
st.set_page_config(page_title="Harvard Art Museums Analytics", layout="wide")
st.title("🏛️ Harvard Art Museums Data Analytics")
st.markdown("---")
# Sidebar configuration
with st.sidebar:
st.header("Configuration")
if st.button("Run ETL Pipeline"):
run_etl_with_progress()
# Main content tabs
tab1, tab2, tab3 = st.tabs(["📊 Analytics", "🗄️ Data Explorer", "📈 Visualizations"])
with tab1:
show_analytics_dashboard()
with tab2:
show_data_explorer()
with tab3:
show_visualizations()
def run_etl_with_progress():
"""Run ETL with progress bar"""
with st.spinner("Running ETL pipeline..."):
connection = get_db_connection()
api_key = os.getenv('HARVARD_API_KEY')
run_etl_pipeline(api_key, connection, num_pages=5)
st.success("ETL completed successfully!")
def show_analytics_dashboard():
"""Display predefined analytics queries"""
st.subheader("Analytical Insights")
queries = {
"Top Cultures": query_cultures,
"Artifacts by Century": query_centuries,
"Media Availability": query_media,
"Color Distribution": query_colors,
"Department Breakdown": query_departments
}
query_choice = st.selectbox("Select Analysis", list(queries.keys()))
if st.button("Run Query"):
connection = get_db_connection()
df = pd.read_sql(queries[query_choice], connection)
st.dataframe(df)
# Auto-generate visualization
if len(df.columns) == 2:
fig = px.bar(df, x=df.columns[0], y=df.columns[1],
title=query_choice)
st.plotly_chart(fig, use_container_width=True)
def show_data_explorer():
"""Interactive data exploration"""
st.subheader("Data Explorer")
table = st.selectbox("Select Table",
["artifactmetadata", "artifactmedia", "artifactcolors"])
connection = get_db_connection()
df = pd.read_sql(f"SELECT * FROM {table} LIMIT 100", connection)
st.dataframe(df)
st.caption(f"Showing first 100 rows from {table}")
if __name__ == "__main__":
main()
```
### Running the Dashboard
```bash
streamlit run app.py
```
## Common Patterns
### Pagination Handler
```python
GitHubで見るこの SKILL.md は非常に大きいため、SkillsMP では最初のセクションだけを表示しています。 GitHubで見る
