json_to_parquet_with_phone_number_cleansing.py

import os
import json
import re
import pyarrow as pa
import pyarrow.parquet as pq
from tqdm import tqdm

def standardize_phone_number(phone):

    digits = re.sub(r'\D', '', str(phone))
    

    if len(digits) >= 10:
        if len(digits) > 10 and digits.startswith('1'):
            digits = digits[-10:]
        elif len(digits) > 10:
            digits = digits[-10:]
        
        return digits
    
    return None

def combine_and_convert_to_parquet(
    input_dir='customer_data', 
    output_dir='customer_parquet', 
    target_file_size_mb=90,
    max_file_size_mb=100
):

    os.makedirs(output_dir, exist_ok=True)

    json_files = sorted([f for f in os.listdir(input_dir) if f.endswith('.json')])

    combined_data = []
    parquet_file_count = 0

    total_processed_files = 0
    phone_cleanup_stats = {
        'total_records': 0,
        'cleaned_records': 0,
        'invalid_records': 0
    }
    
    for json_filename in tqdm(json_files, desc="Processing JSON Files"):

        input_path = os.path.join(input_dir, json_filename)
        
        with open(input_path, 'r', encoding='utf-8') as f:
            data = json.load(f)

        cleaned_data = []
        for record in data:
            phone_cleanup_stats['total_records'] += 1

            cleaned_phone = standardize_phone_number(record['phone'])
            
            if cleaned_phone:
                record['phone'] = cleaned_phone
                cleaned_data.append(record)
                phone_cleanup_stats['cleaned_records'] += 1
            else:
                phone_cleanup_stats['invalid_records'] += 1

        combined_data.extend(cleaned_data)
        total_processed_files += 1

        current_size_mb = len(json.dumps(combined_data)) / (1024 * 1024)
        
        if current_size_mb >= target_file_size_mb or total_processed_files == len(json_files):
            parquet_file_count += 1
            output_path = os.path.join(output_dir, f'combined_customers_{parquet_file_count}.parquet')
            try:
                table = pa.Table.from_pylist(combined_data)
                pq.write_table(table, output_path, compression='snappy')
                file_size_mb = os.path.getsize(output_path) / (1024 * 1024)
                print(f"Created {output_path} - Size: {file_size_mb:.2f} MB")
                if file_size_mb > max_file_size_mb:
                    print(f"Warning: File {output_path} exceeds {max_file_size_mb} MB")
            except Exception as e:
                print(f"Error creating Parquet file: {e}")

            combined_data = []

    print("\n=== Phone Number Cleanup Statistics ===")
    print(f"Total Records: {phone_cleanup_stats['total_records']}")
    print(f"Cleaned Records: {phone_cleanup_stats['cleaned_records']}")
    print(f"Invalid Records: {phone_cleanup_stats['invalid_records']}")
    print(f"Cleanup Success Rate: {phone_cleanup_stats['cleaned_records'] / phone_cleanup_stats['total_records'] * 100:.2f}%")
    
    print(f"\nProcessed {total_processed_files} JSON files")
    print(f"Created {parquet_file_count} Parquet files in {output_dir}")

def demo_phone_standardization():

    test_numbers = [
        '123-456-7890',
        '(123) 456-7890',
        '1234567890',
        '+1 (123) 456-7890',
        '234-567-8901',
        '(800) CALL-NOW',
        '123.456.7890'
    ]
    
    for number in test_numbers:
        cleaned = standardize_phone_number(number)
        print(f"Original: {number} -> Cleaned: {cleaned}")

if __name__ == '__main__':
    combine_and_convert_to_parquet()