feat: implement chunked Parquet uploads to Hugging Face for memory efficiency and improved data handling

README.md

@@ -85,16 +85,15 @@ UI 연결:
      - 마감 확정 전(today) 데이터는 제외
    - 실패 시 최대 3회 재시도
    - 10티커마다 0.5초 대기(호출 차단 완화)
-   - 설정한 간격(`checkpoint_batch_size`)마다 기존 데이터셋과 병합 후 중간 업로드(체크포인트) 수행
+   - 설정한 간격(`checkpoint_batch_size`)마다 수집 버퍼를 Parquet 청크로 업로드
+   - 청크 업로드 후 메모리 버퍼를 즉시 비워 메모리 사용량을 제한
 4. **최종 병합 반영**
-   - 마지막 미반영 버퍼를 기존 데이터셋에 병합
-   - `Ticker + Date` 기준 중복 제거(최신값 유지)
+   - 마지막 미반영 버퍼를 청크 업로드로 반영
 5. **최근 30일 데이터셋 갱신**
-   - 병합된 데이터 기준으로 티커별 30일 윈도우 유지
+   - 티커별 최근 30일 데이터 청크를 함께 업로드
 6. **Hugging Face 업로드**
-   - `upload_dataset_to_hf(all_df, all_dataset_name, hf_token)`
-   - `upload_dataset_to_hf(recent_30d_df, recent_dataset_name, hf_token)`
-   - 내부적으로 `Dataset.from_pandas(...).push_to_hub(...)` 사용
+   - 내부적으로 `HfApi.upload_file(..., path_in_repo="data/chunks/...parquet")` 방식 사용
+   - 대용량 전체 데이터프레임 재업로드를 피해서 메모리 피크를 낮춤
 7. **완료 로그 반환**
    - 시작/종료 시간, 성공/실패 개수, 업로드 결과를 텍스트로 반환
 
@@ -170,6 +169,7 @@ UI 연결:
 
 - 이미 데이터셋에 존재하는 티커는 자동으로 스킵합니다.
 - 이전 실행에서 중간에 실패해도, 다음 실행 시 남은 티커 위주로 이어서 수집합니다.
+- 체크포인트 간격(예: 100) 기준으로 저장되므로, 실패 시 최근 체크포인트 이후 구간만 재수집됩니다.
 
 ---
 

app.py

@@ -17,6 +17,8 @@ import logging
 import json
 import traceback
 import gc
+import tempfile
+import uuid
 from urllib.request import Request, urlopen
 
 # 로깅 설정
@@ -500,6 +502,83 @@ def upload_dataset_to_hf(df, repo_name, hf_token, max_retries=3, retry_wait_sec=
     }
 
 
+def append_parquet_chunk_to_hf(df, repo_name, hf_token, subdir="data/chunks", max_retries=3, retry_wait_sec=2):
+    """데이터프레임을 Parquet 청크 파일로 허깅페이스 데이터셋 저장소에 추가 업로드"""
+    if df is None or df.empty:
+        return {
+            "ok": False,
+            "repo": repo_name,
+            "rows": 0,
+            "attempts": 0,
+            "elapsed_sec": 0.0,
+            "error": "업로드할 데이터가 없습니다.",
+            "traceback": "",
+        }
+
+    api = HfApi()
+    last_error = ""
+    last_traceback = ""
+    start_ts = time.time()
+
+    for attempt in range(1, max_retries + 1):
+        temp_path = None
+        try:
+            api.create_repo(
+                repo_id=repo_name,
+                repo_type="dataset",
+                token=hf_token,
+                private=False,
+                exist_ok=True,
+            )
+
+            chunk_name = f"chunk-{datetime.now().strftime('%Y%m%d-%H%M%S')}-{uuid.uuid4().hex[:8]}.parquet"
+            with tempfile.NamedTemporaryFile(suffix=".parquet", delete=False) as tmp:
+                temp_path = tmp.name
+
+            df.to_parquet(temp_path, index=False)
+
+            path_in_repo = f"{subdir}/{chunk_name}"
+            api.upload_file(
+                path_or_fileobj=temp_path,
+                path_in_repo=path_in_repo,
+                repo_id=repo_name,
+                repo_type="dataset",
+                token=hf_token,
+            )
+
+            return {
+                "ok": True,
+                "repo": repo_name,
+                "rows": len(df),
+                "attempts": attempt,
+                "elapsed_sec": time.time() - start_ts,
+                "error": "",
+                "traceback": "",
+            }
+        except Exception as e:
+            last_error = str(e)
+            last_traceback = traceback.format_exc()
+            logger.warning(f"[{repo_name}] 청크 업로드 실패 (시도 {attempt}/{max_retries}): {last_error}")
+            if attempt < max_retries:
+                time.sleep(retry_wait_sec * attempt)
+        finally:
+            if temp_path and os.path.exists(temp_path):
+                try:
+                    os.remove(temp_path)
+                except Exception:
+                    pass
+
+    return {
+        "ok": False,
+        "repo": repo_name,
+        "rows": len(df),
+        "attempts": max_retries,
+        "elapsed_sec": time.time() - start_ts,
+        "error": last_error,
+        "traceback": last_traceback,
+    }
+
+
 def run_pipeline(
     hf_token,
     all_dataset_name,
@@ -521,16 +600,6 @@ def run_pipeline(
 
     logs = []
     try:
-        def _merge_rows(base_df, incoming_df):
-            if incoming_df is None or incoming_df.empty:
-                return base_df
-            if base_df is None or base_df.empty:
-                return incoming_df.reset_index(drop=True)
-
-            merged = pd.concat([base_df, incoming_df], ignore_index=True)
-            merged = merged.drop_duplicates(subset=["Ticker", "Date"], keep="last")
-            return merged.reset_index(drop=True)
-
         def _df_stats(df, label):
             if df is None or df.empty:
                 return f"{label}: 0행"
@@ -553,6 +622,7 @@ def run_pipeline(
         logs.append("📊 주식 데이터 수집 파이프라인 시작")
         logs.append(f"⏰ 시작 시간: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
         logs.append("=" * 60)
+        logs.append("ℹ️ 메모리 절약 모드: 100개 단위 등 청크 업로드 후 버퍼를 즉시 비웁니다.")
 
         # ========== 1단계: 티커 목록 수집 ==========
         progress(0, desc="나스닥 & 뉴욕 티커 목록 수집 중...")
@@ -611,24 +681,16 @@ def run_pipeline(
         progress(0.08, desc="기존 데이터셋 로드 중...")
         logs.append("\n📂 [2단계] 기존 데이터셋 로드 및 재개 대상 계산...")
 
-        base_all_df = pd.DataFrame(columns=["Ticker", "Date", "Open", "High", "Low", "Close", "Volume"])
-        base_recent_df = pd.DataFrame(columns=["Ticker", "Date", "Open", "High", "Low", "Close", "Volume"])
-
+        recent_for_resume = pd.DataFrame(columns=["Ticker"])
         try:
-            base_all_df = load_hf_dataset_as_df(all_dataset_name, hf_token)
-            logs.append(f"  - 기존 all 데이터: {len(base_all_df)}행")
+            recent_for_resume = load_hf_dataset_as_df(recent_dataset_name, hf_token)
+            logs.append(f"  - 기존 30d 데이터: {len(recent_for_resume)}행")
         except Exception as e:
-            logs.append(f"  - 기존 all 데이터 로드 실패(신규 생성으로 진행): {e}")
+            logs.append(f"  - 기존 30d 데이터 로드 실패(신규 수집 기준으로 진행): {e}")
 
-        try:
-            base_recent_df = load_hf_dataset_as_df(recent_dataset_name, hf_token)
-            logs.append(f"  - 기존 30d 데이터: {len(base_recent_df)}행")
-        except Exception as e:
-            logs.append(f"  - 기존 30d 데이터 로드 실패(신규 생성으로 진행): {e}")
-
-        existing_tickers = set(base_all_df["Ticker"].dropna().astype(str).str.upper().tolist())
+        existing_tickers = set(recent_for_resume["Ticker"].dropna().astype(str).str.upper().tolist())
         if not existing_tickers:
-            existing_tickers = set(base_recent_df["Ticker"].dropna().astype(str).str.upper().tolist())
+            logs.append("  - 기존 티커 정보가 비어 있어 전체 대상 기준으로 진행합니다.")
 
         pending_tickers = [ticker for ticker in all_tickers if ticker not in existing_tickers]
 
@@ -655,7 +717,7 @@ def run_pipeline(
         total = len(pending_tickers)
 
         def _upload_checkpoint(end_index):
-            nonlocal last_checkpoint_success_index, base_all_df, base_recent_df
+            nonlocal last_checkpoint_success_index
 
             if success_count <= last_checkpoint_success_index:
                 return
@@ -671,15 +733,21 @@ def run_pipeline(
             checkpoint_all_df = pd.concat(all_data_frames, ignore_index=True)
             checkpoint_recent_df = pd.concat(recent_30d_frames, ignore_index=True)
 
-            base_all_df = _merge_rows(base_all_df, checkpoint_all_df)
-            base_recent_df = _merge_rows(base_recent_df, checkpoint_recent_df)
-            base_recent_df = filter_last_30_days(base_recent_df)
+            logs.append(f"  - {_df_stats(checkpoint_all_df, 'all 청크')}")
+            logs.append(f"  - {_df_stats(checkpoint_recent_df, '30d 청크')}")
 
-            logs.append(f"  - {_df_stats(base_all_df, 'all 병합 결과')}")
-            logs.append(f"  - {_df_stats(base_recent_df, '30d 병합 결과')}")
-
-            result_all_ckpt = upload_dataset_to_hf(base_all_df, all_dataset_name, hf_token)
-            result_30d_ckpt = upload_dataset_to_hf(base_recent_df, recent_dataset_name, hf_token)
+            result_all_ckpt = append_parquet_chunk_to_hf(
+                checkpoint_all_df,
+                all_dataset_name,
+                hf_token,
+                subdir="data/chunks/all"
+            )
+            result_30d_ckpt = append_parquet_chunk_to_hf(
+                checkpoint_recent_df,
+                recent_dataset_name,
+                hf_token,
+                subdir="data/chunks/30d"
+            )
 
             _append_upload_result("all 체크포인트", result_all_ckpt)
             _append_upload_result("30d 체크포인트", result_30d_ckpt)
@@ -729,37 +797,14 @@ def run_pipeline(
         progress(0.9, desc="마지막 체크포인트 반영 중...")
         logs.append("\n🔧 [4단계] 마지막 미반영 데이터 반영 중...")
 
-        if checkpoint_batch_size > 0 and success_count > last_checkpoint_success_index:
-            logs.append("\n💾 [체크포인트] 마지막 미반영 구간 업로드")
-            _upload_checkpoint(total)
-
-        if checkpoint_batch_size == 0 and all_data_frames:
+        if success_count > last_checkpoint_success_index:
             logs.append("\n💾 [최종반영] 중간 업로드 없이 누적된 데이터 반영")
             _upload_checkpoint(total)
 
-        logs.append(f"  - 전체 데이터: {len(base_all_df)}행 x {len(base_all_df.columns)}열")
-        logs.append(f"  - 고유 티커 수: {base_all_df['Ticker'].nunique()}")
-        logs.append(f"  - 30일 데이터: {len(base_recent_df)}행 x {len(base_recent_df.columns)}열")
-        logs.append(f"  - 30일 고유 티커 수: {base_recent_df['Ticker'].nunique()}")
-
-        # ========== 5단계: 허깅페이스 데이터셋 업로드 ==========
-        progress(0.97, desc="all 데이터셋 업로드 중...")
-        logs.append("\n🚀 [5단계] 허깅페이스 데이터셋 업로드 중...")
-
-        # all 데이터셋 업로드
-        logs.append(f"  - {_df_stats(base_all_df, 'all 최종 업로드 대상')}")
-        result_all = upload_dataset_to_hf(base_all_df, all_dataset_name, hf_token)
-        _append_upload_result("all 최종", result_all)
-        if not result_all["ok"]:
-            raise RuntimeError("all 최종 업로드 실패")
-
-        # 30일 데이터셋 업로드
-        progress(0.99, desc="30d 데이터셋 업로드 중...")
-        logs.append(f"  - {_df_stats(base_recent_df, '30d 최종 업로드 대상')}")
-        result_30d = upload_dataset_to_hf(base_recent_df, recent_dataset_name, hf_token)
-        _append_upload_result("30d 최종", result_30d)
-        if not result_30d["ok"]:
-            raise RuntimeError("30d 최종 업로드 실패")
+        progress(0.97, desc="청크 업로드 상태 마무리 중...")
+        logs.append("\n🚀 [5단계] 청크 업로드 모드 완료")
+        logs.append("  - all/30d 모두 청크 파일 기준으로 저장되었습니다.")
+        logs.append("  - 다음 실행 시 30d 티커 목록 기준으로 자동 스킵/재개됩니다.")
 
         # ========== 완료 ==========
         progress(1.0, desc="완료!")