fix: per-file CSV scan for heterogeneous schemas, add upload file count logging
This commit is contained in:
+16
-11
@@ -311,6 +311,7 @@ def upload_csv(request):
|
||||
files = request.FILES.getlist('files')
|
||||
if not files:
|
||||
return JsonResponse({'error': 'No files uploaded'})
|
||||
logger.info('[UPLOAD] Received %d files', len(files))
|
||||
for f in files:
|
||||
if f.size > 500 * 1024 * 1024:
|
||||
return JsonResponse({'warning': f'文件 {f.name} 超过 500MB,建议分割后上传'})
|
||||
@@ -453,6 +454,7 @@ def _background_process(run_id, upload_dir):
|
||||
logger.warning('[BACKGROUND] Failed to detect lat/lon columns', exc_info=True)
|
||||
|
||||
total_files = len(csv_paths)
|
||||
logger.info('[BACKGROUND] Found %d CSV files in %s', total_files, upload_dir)
|
||||
# Adaptive batch size: aim for ~20 batches total, cap at 500 files per batch
|
||||
batch_size = max(1, min(500, total_files // 20 or 1))
|
||||
total_batches = (total_files + batch_size - 1) // batch_size
|
||||
@@ -496,17 +498,20 @@ def _background_process(run_id, upload_dir):
|
||||
total_rows = 0
|
||||
for i in range(0, total_files, batch_size):
|
||||
batch = csv_paths[i:i+batch_size]
|
||||
batch_lf = pl.scan_csv(batch, infer_schema_length=0,
|
||||
schema_overrides=schema_overrides)
|
||||
df_batch = batch_lf.collect(streaming=True)
|
||||
|
||||
# Schema alignment: fill NULL for columns missing in this batch
|
||||
for col in reference_columns:
|
||||
if col not in df_batch.columns:
|
||||
df_batch = df_batch.with_columns(pl.lit(None).alias(col))
|
||||
|
||||
# Select all reference_columns in canonical order
|
||||
df_batch = df_batch.select(reference_columns)
|
||||
# Scan each file INDIVIDUALLY — batch scan fails when files have different schemas
|
||||
dfs = []
|
||||
for fp in batch:
|
||||
lf = pl.scan_csv(fp, infer_schema_length=0,
|
||||
schema_overrides=schema_overrides)
|
||||
df = lf.collect(streaming=True)
|
||||
# Fill missing columns with NULL
|
||||
for col in reference_columns:
|
||||
if col not in df.columns:
|
||||
df = df.with_columns(pl.lit(None).alias(col))
|
||||
# Select all reference_columns in canonical order
|
||||
df = df.select(reference_columns)
|
||||
dfs.append(df)
|
||||
df_batch = pl.concat(dfs, how='diagonal_relaxed')
|
||||
|
||||
# Append to pre-created SQLite table
|
||||
save_to_db(run.id, df_batch.lazy(), mode='append')
|
||||
|
||||
Reference in New Issue
Block a user