Multi Source Data Merger Overview This skill guides the process of merging data from multiple sources with different formats into a unified dataset. It covers reading heterogeneous file formats, applying field name mappings, resolving conflicts using priority ordering, and generating comprehensive output files including conflict reports. Workflow Step 1: Analyze Requirements and Source Files Before writing any code, thoroughly understand the task: Identify all source files and their formats (JSON, CSV, Parquet, XML, etc.) Determine the merge key (e.g., user_id , record_id ) that links records across sources Review field mapping requirements - source fields may have different names that map to common output fields Understand conflict resolution rules - typically based on source priority ordering Identify expected output formats and structure Important: Do not attempt to read binary formats (Parquet, Excel, etc.) as text files - use appropriate libraries. Step 2: Set Up Environment Create a Python virtual environment using uv or venv Install required dependencies based on source formats: pandas - Core data manipulation pyarrow - Parquet file support openpyxl - Excel file support lxml - XML parsing (if needed) Verify installations before proceeding Example environment setup: uv venv .venv source .venv/bin/activate uv pip install pandas pyarrow Step 3: Write the Merge Script Structure the script with clear separation of concerns: Data reading functions - One per format type Field mapping function - Apply column renames Data normalization - Handle date formats, type conversions Merge logic - Combine records using the merge key Conflict resolution - Apply priority rules Output generation - Write merged data and conflict reports Script quality practices: Validate syntax before execution: python -m py_compile script.py Use try-except blocks with informative error messages Document assumptions about data formats Step 4: Execute and Verify Run a comprehensive verification process: Check output file existence at expected locations Validate merged data contains expected values Verify conflict report structure and content Run any provided test suites Common Pitfalls Binary File Handling Mistake: Attempting to read Parquet/Excel files as text Solution: Always use pandas with appropriate engine ( pyarrow for Parquet, openpyxl for Excel) Syntax Errors in Scripts Mistake: Writing long scripts without validation, leading to indentation or syntax errors Solution: Run python -m py_compile script.py before execution Date Format Normalization Mistake: Assuming consistent date formats across sources Solution: Implement flexible date parsing that handles multiple formats: ISO format: 2024-01-15 US format: 01/15/2024 European format: 15-01-2024 Datetime: 2024-01-15T10:30:00 Incomplete Script Output Mistake: Writing very long scripts that may get truncated Solution: Break into modular functions, verify complete code visibility Environment Path Issues Mistake: Repeating PATH exports in every command Solution: Set PATH once in a setup step or use absolute paths to executables Verification Strategies Output Validation Checklist File existence check: import os assert os . path . exists ( "output/merged_data.json" ) assert os . path . exists ( "output/conflict_report.json" ) Data completeness check: import json with open ( "output/merged_data.json" ) as f : data = json . load ( f )

Verify expected record count

assert len ( data ) == expected_count Conflict report validation: with open ( "output/conflict_report.json" ) as f : conflicts = json . load ( f )

Verify conflict structure has required fields

for conflict in conflicts : assert "field" in conflict assert "selected" in conflict assert "sources" in conflict Sample value verification:

Spot-check specific merged records

record

next ( r for r in data if r [ "user_id" ] == "expected_id" ) assert record [ "field_name" ] == "expected_value" Consolidate Verification Instead of running multiple separate verification commands, create a single comprehensive test script that validates all aspects of the output. Edge Cases to Consider Empty source files - Handle gracefully with appropriate warnings Missing merge keys - Decide whether to skip or error Type mismatches - Convert consistently (e.g., user_id as string vs integer) Null/None values - Determine handling in conflict resolution Unicode/encoding - Specify encoding when reading text-based formats Records in some sources but not others - Include partial records or require complete matches Field Mapping Example When sources have different field names for the same concept: FIELD_MAPPINGS = { "source_a" : { "firstName" : "first_name" , "lastName" : "last_name" , "emailAddress" : "email" } , "source_b" : { "fname" : "first_name" , "lname" : "last_name" , "mail" : "email" } } def apply_mapping ( df , source_name ) : mapping = FIELD_MAPPINGS . get ( source_name , { } ) return df . rename ( columns = mapping ) Conflict Resolution Pattern When the same field has different values across sources: def resolve_conflict ( values_by_source , priority_order ) : """ Select value based on source priority. Args: values_by_source: dict mapping source name to value priority_order: list of source names from highest to lowest priority Returns: tuple: (selected_value, conflict_info) """ conflict_info = None unique_values = set ( v for v in values_by_source . values ( ) if v is not None ) if len ( unique_values )

1 : conflict_info = { "sources" : values_by_source , "resolved_by" : "priority" } for source in priority_order : if source in values_by_source and values_by_source [ source ] is not None : return values_by_source [ source ] , conflict_info return None , conflict_info