당신의 삶을 정량화하세요: InfluxDB, Grafana, Python으로 고성능 건강 데이터 레이크 구축 🚀

Source: Dev.to

우리는 “퀀티파이드 셀프(Quantified Self)” 시대에 살고 있습니다. Apple Watch가 심박 변동성을 추적하고, Strava가 매번 라이딩을 기록하며, MyFitnessPal이 단백질 1그램씩까지 기록하는 등 우리는 개인 건강 데이터 수 기가바이트를 생성합니다. 문제는 이 데이터가 각각의 사일로에 존재한다는 점입니다. 수면 품질(Apple Health)과 훈련 부하(Strava), 그리고 칼로리 섭취량(MyFitnessPal)을 연관시키려면 보통 세 개의 앱을 오가야 합니다.

이 가이드에서는 데이터 엔지니어링 모범 사례를 적용해 이를 해결합니다: 시계열 저장을 위한 InfluxDB 기반 개인 데이터 레이크, ETL을 위한 Python, 그리고 미션 컨트롤 스타일 대시보드를 위한 Grafana. 최종적으로 Docker만으로 실행되는 건강 지표에 대한 단일 진실의 원천을 갖게 될 것입니다.

아키텍처 개요

graph TD
    A[Apple Health Export] -->|XML/CSV| B(Python ETL Script)
    C[Strava API] -->|JSON| B
    D[MyFitnessPal] -->|Web Scraping/Export| B
    B -->|Clean & Transform| E{InfluxDB}
    E -->|Query via Flux| F[Grafana Dashboard]
    F -->|Visualization| G[Your Big Screen]

    style E fill:#f96,stroke:#333,stroke-width:2px
    style F fill:#3262a8,stroke:#333,stroke-width:2px

필수 조건

• Docker 및 Docker Compose 설치
• Python 3.9+ (ETL 로직용)
• Strava API 자격 증명에 대한 접근 권한
• “공개 학습” 마인드셋

Docker‑Compose 스택

# docker-compose.yml
version: '3.8'
services:
  influxdb:
    image: influxdb:2.7
    ports:
      - "8086:8086"
    volumes:
      - influxdb_data:/var/lib/influxdb2
    environment:
      - DOCKER_INFLUXDB_INIT_MODE=setup
      - DOCKER_INFLUXDB_INIT_USERNAME=admin
      - DOCKER_INFLUXDB_INIT_PASSWORD=password123
      - DOCKER_INFLUXDB_INIT_ORG=my_health_org
      - DOCKER_INFLUXDB_INIT_BUCKET=health_metrics

  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    depends_on:
      - influxdb
    volumes:
      - grafana_data:/var/lib/grafana

volumes:
  influxdb_data:
  grafana_data:

스택 실행:

docker compose up -d

파이썬 ETL 예제 (Strava)

import pandas as pd
from influxdb_client import InfluxDBClient, Point, WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS

# Configuration
token = "YOUR_INFLUXDB_TOKEN"
org = "my_health_org"
bucket = "health_metrics"

client = InfluxDBClient(url="http://localhost:8086", token=token, org=org)
write_api = client.write_api(write_options=SYNCHRONOUS)

def process_strava_data(json_data):
    """
    Transforms raw Strava JSON into InfluxDB Points.
    """
    for activity in json_data:
        point = (
            Point("fitness")
            .tag("source", "strava")
            .tag("type", activity["type"])
            .field("distance", float(activity["distance"]))
            .field("heart_rate_avg", float(activity.get("average_heartrate", 0)))
            .time(activity["start_date"], WritePrecision.NS)
        )
        write_api.write(bucket, org, point)

    print("✅ Strava data ingested successfully!")

# Example usage with mock data
mock_strava = [
    {
        "type": "Run",
        "distance": 5000.5,
        "average_heartrate": 155,
        "start_date": "2023-10-27T08:00:00Z",
    }
]
process_strava_data(mock_strava)

Apple Health XML 레코드 매핑

Tip: Apple Health 내보내기 파일은 수 기가바이트 규모의 XML 파일일 수 있습니다. 전체 문서를 메모리에 로드하지 않도록 점진적으로 파싱하세요(예: lxml.iterparse 사용).

Grafana 대시보드 설정

1. Grafana에 접속합니다 .
2. InfluxDB를 데이터 소스로 추가하고 쿼리 언어로 Flux를 선택합니다.
3. 새 대시보드를 만들고 → Time Series 패널을 추가합니다.

샘플 Flux 쿼리 (시간에 따른 심박수)

from(bucket: "health_metrics")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "fitness")
  |> filter(fn: (r) => r["_field"] == "heart_rate_avg")
  |> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
  |> yield(name: "mean_hr")

다음은?

• 알림: 휴식 심박수가 상승한 상태를 유지할 때 알림을 보내는 텔레그램 봇(또는 기타 웹훅)을 추가합니다.
• AI 인사이트: 대형 언어 모델(예: GPT‑4o)을 통합하여 월별 추세를 분석하고 개인 맞춤형 건강 조언을 제공합니다.
• 확장성: 프로덕션 수준 배포를 위해 고급 데이터 수집 패턴, 시계열 최적화 및 데이터 프라이버시 보호 조치를 탐색하세요(자세한 내용은 WellAlly Tech 블로그를 참조).

자신의 건강 데이터 레이크를 구축하면 메트릭을 완전히 제어하고, 맞춤형 상관관계를 실행하며, 장기 추세를 감지하고, 디지털 트윈을 진정으로 소유할 수 있습니다.