diff --git a/statvar_imports/crdc_import/crdc_harassment_or_bullying/manifest.json b/statvar_imports/crdc_import/crdc_harassment_or_bullying/manifest.json index afa5686ec7..6d6a308bcc 100644 --- a/statvar_imports/crdc_import/crdc_harassment_or_bullying/manifest.json +++ b/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 } } ] -} \ No newline at end of file +} diff --git a/statvar_imports/crdc_import/crdc_harassment_or_bullying/run.sh b/statvar_imports/crdc_import/crdc_harassment_or_bullying/run.sh index ade36d0b71..af49d20471 100644 --- a/statvar_imports/crdc_import/crdc_harassment_or_bullying/run.sh +++ b/statvar_imports/crdc_import/crdc_harassment_or_bullying/run.sh @@ -22,17 +22,18 @@ 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 @@ -40,43 +41,69 @@ if [ ! -d "$INPUT_DIR" ]; then 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 @@ -84,4 +111,4 @@ if [ "$FAILED_JOBS" -gt 0 ]; then exit 1 else echo "All parallel processing jobs completed successfully. Output files are in $OUTPUT_DIR" -fi \ No newline at end of file +fi diff --git a/statvar_imports/crdc_import/crdc_harassment_or_bullying/validation_config.json b/statvar_imports/crdc_import/crdc_harassment_or_bullying/validation_config.json new file mode 100644 index 0000000000..1b32b1fcaa --- /dev/null +++ b/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" + } + } + ] +} \ No newline at end of file