[CRDC_EducationSchool_HarassmentOrBullying] Partitioned import inputs by years

statvar_imports/crdc_import/crdc_harassment_or_bullying/manifest.json

@@ -8,24 +8,54 @@
             "provenance_url": "https://civilrightsdata.ed.gov/data",
             "provenance_description": "The Civil Rights Data Collection (CRDC) provides data on key education and civil rights issues in our nation's public schools. This import focuses on harassment or bullying data.",
             "scripts": [
-                "download.py","preprocess.py","run.sh"
-                        ],
+                "download.py",
+                "preprocess.py",
+                "run.sh"
+            ],
             "import_inputs": [
+                {
+                    "template_mcf": "processed_output/crdc_harassment_or_bullying_2012_part_01.tmcf",
+                    "cleaned_csv": "processed_output/crdc_harassment_or_bullying_2012*.csv",
+                    "node_mcf": "processed_output/crdc_harassment_or_bullying_2012*.mcf"
+                },
+                {
+                    "template_mcf": "processed_output/crdc_harassment_or_bullying_2014.tmcf",
+                    "cleaned_csv": "processed_output/crdc_harassment_or_bullying_2014.csv",
+                    "node_mcf": "processed_output/crdc_harassment_or_bullying_2014*.mcf"
+                },
+                {
+                    "template_mcf": "processed_output/crdc_harassment_or_bullying_2018.tmcf",
+                    "cleaned_csv": "processed_output/crdc_harassment_or_bullying_2018.csv",
+                    "node_mcf": "processed_output/crdc_harassment_or_bullying_2018*.mcf"
+                },
+                {
+                    "template_mcf": "processed_output/crdc_harassment_or_bullying_2021.tmcf",
+                    "cleaned_csv": "processed_output/crdc_harassment_or_bullying_2021.csv",
+                    "node_mcf": "processed_output/crdc_harassment_or_bullying_2021*.mcf"
+                },
                 {
                     "template_mcf": "processed_output/crdc_harassment_or_bullying_2022.tmcf",
-                    "cleaned_csv": "processed_output/*.csv"
+                    "cleaned_csv": "processed_output/crdc_harassment_or_bullying_2022.csv",
+                    "node_mcf": "processed_output/crdc_harassment_or_bullying_2022*.mcf"
+                },
+                {
+                    "template_mcf": "processed_output/crdc_harassment_or_bullying_2024Onwards.tmcf",
+                    "cleaned_csv": "processed_output/crdc_harassment_or_bullying_2024Onwards.csv",
+                    "node_mcf": "processed_output/crdc_harassment_or_bullying_2024Onwards*.mcf"
                 }
-                ],
+            ],
             "source_files": [
-                "input_files/*.csv"
+                "input_files/*.csv",
+                "processed_output/*_counters.txt",
+                "validation_config.json"
             ],
             "cron_schedule": "30 6 * * 5",
             "_comment": "Schedule: Weekly on Friday at 06:30 UTC",
             "resource_limits": {
-                "cpu": 8,
-                "memory": 64,
-                "disk": 100
+                "cpu": 16,
+                "memory": 128,
+                "disk": 200
             }
         }
     ]
-}
+}

statvar_imports/crdc_import/crdc_harassment_or_bullying/run.sh

@@ -22,66 +22,93 @@
 INPUT_DIR="./input_files"
 PV_MAP_FILE="harassment_or_bullying_pvmap.csv"
 CONFIG_FILE="harassment_or_bullying_metadata.csv"
-OUTPUT_DIR="processed_output" 
+OUTPUT_DIR="processed_output"
 EXISTING_MCF="gs://unresolved_mcf/scripts/statvar/stat_vars.mcf"
 PYTHON_SCRIPT="../../../tools/statvar_importer/stat_var_processor.py"
 
-# TODO: Limit concurrency to avoid overloading system resources
-MAX_PARALLEL_JOBS=4
+# Use all available vCPUs (defaults to 16 matching manifest.json resource_limits)
+MAX_PARALLEL_JOBS=$(nproc 2>/dev/null || echo 16)
 
 # =================================================================
 # EXECUTION LOGIC
 # =================================================================
 
+rm -rf "$OUTPUT_DIR"
 mkdir -p "$OUTPUT_DIR"
 
 if [ ! -d "$INPUT_DIR" ]; then
     echo "Error: Input directory '$INPUT_DIR' not found."
     exit 1
 fi
 
-echo "Starting parallel processing of files in $INPUT_DIR..."
+echo "Starting parallel processing of files in $INPUT_DIR (MAX_PARALLEL_JOBS=$MAX_PARALLEL_JOBS)..."
 JOB_COUNT=0
 FAILED_JOBS=0
+PIDS=()
+
+# 1. Process 2024 onwards files dynamically into a single 2024Onwards output
+CSV_FILES_2024_ONWARD=$(find "$INPUT_DIR" -maxdepth 1 -type f -name "*.csv" | grep -Ev "_(2012|2014|2018|2021|2022)(_|\.)" | sort | paste -sd, -)
+
+if [ -z "$CSV_FILES_2024_ONWARD" ]; then
+    echo "Error: No 2024+ CSV files found in '$INPUT_DIR'."
+    exit 1
+fi
+
+output_file="$OUTPUT_DIR/crdc_harassment_or_bullying_2024Onwards"
+echo "Launching job for 2024Onwards: $CSV_FILES_2024_ONWARD"
+python3 "$PYTHON_SCRIPT" \
+    --input_data="$CSV_FILES_2024_ONWARD" \
+    --pv_map="$PV_MAP_FILE" \
+    --config_file="$CONFIG_FILE" \
+    --existing_statvar_mcf="$EXISTING_MCF" \
+    --output_path="$output_file" \
+    --output_counters="${output_file}_counters.txt" \
+    --log_level=-2 \
+    --log_every_n=10000 &
+PIDS+=("$!")
+JOB_COUNT=$((JOB_COUNT + 1))
+
+# 2. Sort remaining (2012-2022) CSV files by size descending so the largest files start first
+while IFS= read -r base_name; do
+    [ -z "$base_name" ] && continue
+    input_file="$INPUT_DIR/$base_name"
+    if [ -f "$input_file" ] && echo "$base_name" | grep -Eq "_(2012|2014|2018|2021|2022)(_.*)?\.csv$"; then
+        # If we reached the max number of parallel jobs, wait for a slot to open
+        while [[ $(jobs -r -p | wc -l) -ge $MAX_PARALLEL_JOBS ]]; do
+            wait -n 2>/dev/null || sleep 1
+        done
 
-for input_file in "$INPUT_DIR"/*; do
-    if [ -f "$input_file" ]; then
-
-        base_name=$(basename "$input_file")
         filename_only="${base_name%.*}"
         output_file="$OUTPUT_DIR/${filename_only}"
-        
+
         echo "Launching job for: $input_file"
-        
+
         python3 "$PYTHON_SCRIPT" \
             --input_data="$input_file" \
             --pv_map="$PV_MAP_FILE" \
             --config_file="$CONFIG_FILE" \
             --existing_statvar_mcf="$EXISTING_MCF" \
-            --output_path="$output_file" &
-
-        JOB_COUNT=$((JOB_COUNT + 1))
+            --output_path="$output_file" \
+            --output_counters="${output_file}_counters.txt" \
+            --log_level=-2 \
+            --log_every_n=10000 &
 
-        # TODO: Check job status and limit concurrency
-        # If we reached the max number of parallel jobs, wait for at least one to finish
-        if [[ $(jobs -r -p | wc -l) -ge $MAX_PARALLEL_JOBS ]]; then
-            wait -n
-        fi
+        PIDS+=("$!")
+        JOB_COUNT=$((JOB_COUNT + 1))
     fi
-done
+done < <(find "$INPUT_DIR" -maxdepth 1 -type f -name "*.csv" -printf '%s\t%f\n' | sort -rn -k1,1 -k2,2 | cut -f2-)
 
 echo "---"
-echo "Waiting for remaining $(jobs -r -p | wc -l) jobs to complete..."
+echo "Waiting for remaining $(jobs -r -p | wc -l) of $JOB_COUNT jobs to complete..."
 
-# TODO: Check job status of all background processes
-# Wait for all processes and capture exit codes to identify failures
-for job in $(jobs -p); do
-    wait "$job" || FAILED_JOBS=$((FAILED_JOBS + 1))
+# Wait for all spawned processes and capture exit codes to identify failures
+for pid in "${PIDS[@]}"; do
+    wait "$pid" || FAILED_JOBS=$((FAILED_JOBS + 1))
 done
 
 if [ "$FAILED_JOBS" -gt 0 ]; then
     echo "Processing complete with $FAILED_JOBS errors."
     exit 1
 else
     echo "All parallel processing jobs completed successfully. Output files are in $OUTPUT_DIR"
-fi
+fi

statvar_imports/crdc_import/crdc_harassment_or_bullying/validation_config.json

@@ -0,0 +1,28 @@
+{
+  "schema_version": "1.0",
+  "rules": [
+    {
+      "rule_id": "check_deleted_records_percent",
+      "description": "Checks that the percentage of deleted records across the import does not exceed 0.1% (accounts for school closures/mergers across biennial CRDC releases).",
+      "validator": "DELETED_RECORDS_PERCENT",
+      "params": {
+        "threshold": 0.1
+      }
+    },
+    {
+      "rule_id": "check_max_date_consistent",
+      "description": "Ensures all StatVars within each survey-year input share the same latest observation date.",
+      "validator": "MAX_DATE_CONSISTENT",
+      "params": {}
+    },
+    {
+      "rule_id": "check_latest_date_freshness",
+      "description": "Ensures the 2024Onwards partition contains data for year 2024 or later.",
+      "validator": "SQL_VALIDATOR",
+      "params": {
+        "query": "SELECT StatVar, CAST(MaxDate AS INT) AS latest_year FROM stats WHERE CAST(MinDate AS INT) >= 2024",
+        "condition": "latest_year >= 2024"
+      }
+    }
+  ]
+}