基于CSV动态确定Oracle列类型 解决大数据Excel运算卡顿难题

Great question—handling messy, large CSVs with missing values and needing to infer proper Oracle data types is a super common pain point when moving away from Excel. Let’s walk through practical, scalable solutions here, since manually defining schemas isn’t sustainable for your long-term goal of handling diverse spreadsheets.

Core Problem Recap #

You can’t rely on just the second row (or even a handful of rows) because missing values mean the actual data type for a column might only show up later in the file. For example, your ID column has 10R91 and 10M99—those can’t be integers, so you need to catch that before setting the column to NUMBER.

The Right Approach: Type Inference with Priority & Sampling/Full Traversal #

The key idea is to assign a priority hierarchy to data types (stricter types first) and validate each non-null value in the column against that hierarchy, downgrading the column type when a value doesn’t fit the current strictest type.

Type Priority Hierarchy (Strict → Flexible) #

We want to use the most restrictive type possible for each column, since that’s better for query performance and data integrity:

1. INTEGER (Oracle: NUMBER(10) or adjust precision based on your data)
2. FLOAT/DECIMAL (Oracle: NUMBER(19,4) or tweak scale/precision as needed)
3. DATE (Oracle: DATE—support common formats like YYYY.MM.DD, MM/DD/YYYY, etc.)
4. STRING (Oracle: VARCHAR2(n) where n is the maximum length of values in the column)

Should You Traverse All Rows or Sample? #

• Full traversal: 100% accurate, and manageable in Python if you read the CSV line-by-line (no need to load the entire 1M-row file into memory). The built-in csv module is efficient enough for this—you won’t hit memory limits as long as you process one row at a time.
• Sampling: Faster, and works for most real-world datasets. Sample a large random subset (e.g., 10,000 rows plus the first 1,000 rows) to minimize the chance of missing edge cases. If you’re worried about rare type mismatches, add a post-import validation step to flag rows that don’t fit the inferred schema.

Step-by-Step Implementation in Python #

Let’s put this into code. We’ll use memory-efficient line-by-line reading and write helper functions to detect each data type.

1. Helper Functions for Type Detection #

First, functions to check if a value fits a specific type:

import csv
from datetime import datetime

def is_integer(val):
    try:
        int(val)
        return True
    except (ValueError, TypeError):
        return False

def is_float(val):
    try:
        float(val)
        return True
    except (ValueError, TypeError):
        return False

def is_date(val, formats=["%Y.%m.%d", "%Y-%m-%d", "%m/%d/%Y"]):
    for fmt in formats:
        try:
            datetime.strptime(val, fmt)
            return True
        except ValueError:
            continue
    return False

2. Traverse CSV to Infer Column Types #

We’ll start each column at the strictest type (integer), then iterate through rows and downgrade types as needed. We’ll also track maximum string lengths for VARCHAR2 columns:

def infer_csv_schema(csv_path, sample_size=None):
    schema = {}
    max_str_length = {}
    
    with open(csv_path, 'r', newline='', encoding='utf-8') as f:
        reader = csv.DictReader(f, delimiter='\t')  # Adjust delimiter if your CSV uses commas
        headers = reader.fieldnames
        
        # Initialize schema and max length tracking for each column
        for col in headers:
            schema[col] = 'integer'
            max_str_length[col] = 0
        
        row_count = 0
        for row in reader:
            # Stop early if using sampling
            if sample_size and row_count >= sample_size:
                break
            
            for col in headers:
                val = row[col].strip()
                # Skip empty values (NULLs don't affect type inference)
                if not val or val == '""':
                    continue
                
                # Update max string length for VARCHAR2 sizing
                current_len = len(val)
                if current_len > max_str_length[col]:
                    max_str_length[col] = current_len
                
                # Downgrade type if current value doesn't fit
                current_type = schema[col]
                if current_type == 'integer':
                    if not is_integer(val):
                        if is_float(val):
                            schema[col] = 'float'
                        elif is_date(val):
                            schema[col] = 'date'
                        else:
                            schema[col] = 'string'
                elif current_type == 'float':
                    if not is_float(val):
                        if is_date(val):
                            schema[col] = 'date'
                        else:
                            schema[col] = 'string'
                elif current_type == 'date':
                    if not is_date(val):
                        schema[col] = 'string'
                # String is the final type—no further downgrades
            
            row_count += 1
    
    # Map inferred types to Oracle-compatible data types
    oracle_schema = {}
    for col in headers:
        typ = schema[col]
        if typ == 'integer':
            oracle_schema[col] = f'"{col}" NUMBER(10)'  # Adjust precision for larger IDs
        elif typ == 'float':
            oracle_schema[col] = f'"{col}" NUMBER(19,4)'  # Tweak scale for more decimals
        elif typ == 'date':
            oracle_schema[col] = f'"{col}" DATE'
        else:
            # Use max length + buffer, or default to 255 if sampling missed long values
            str_len = max(max_str_length[col], 255)
            oracle_schema[col] = f'"{col}" VARCHAR2({str_len})'
    
    return oracle_schema

3. Generate CREATE TABLE Statement #

Once you have the inferred schema, generate the SQL to create your Oracle table:

def create_table_sql(table_name, oracle_schema):
    columns_clause = ', '.join(oracle_schema.values())
    return f'CREATE TABLE {table_name} ({columns_clause})'

# Usage example
# Use sample_size for faster inference, or remove it for full traversal
schema = infer_csv_schema('./data.csv', sample_size=10000)
sql = create_table_sql('SOME_TABLE_NAME', schema)
print(sql)

Key Optimizations & Notes #

• Memory Efficiency: Using csv.DictReader line-by-line ensures you never load the entire 1M-row CSV into memory—critical for large files.
• Date Format Flexibility: Update the formats list in is_date() to match your actual date formats (e.g., add %d.%m.%Y if needed).
• Oracle Type Tweaks: Adjust NUMBER precision/scale based on your data (e.g., use NUMBER(15) if IDs can be larger than 10 digits).
• Post-Import Validation: If you use sampling, run a quick query after importing to check for mismatches (e.g., SELECT * FROM SOME_TABLE_NAME WHERE NOT REGEXP_LIKE(ID, '^[0-9]+$') if ID was inferred as integer).
• Column Name Handling: The code wraps column names in double quotes to handle spaces or special characters—remove the quotes if your column names are safe for unquoted identifiers.

内容的提问来源于stack exchange，提问作者MattSom