Loading...
Loading...
Profile datasets to understand schema, quality, and characteristics. Use when analyzing data files (CSV, JSON, Parquet), discovering dataset properties, assessing data quality, or when user mentions data profiling, schema detection, data analysis, or quality metrics. Provides basic and intermediate profiling including distributions, uniqueness, and pattern detection.
npx skill4agent add fusionet24/aiskills data-profilerimport pandas as pd
def detect_schema(df: pd.DataFrame) -> dict:
"""
Detect schema information from DataFrame.
Returns:
Dictionary with column names, types, nullable status
"""
schema = {
'columns': []
}
for col in df.columns:
col_info = {
'name': col,
'type': str(df[col].dtype),
'nullable': df[col].isnull().any(),
'null_count': int(df[col].isnull().sum()),
'null_percentage': float(df[col].isnull().sum() / len(df) * 100)
}
# Infer semantic type
if pd.api.types.is_numeric_dtype(df[col]):
col_info['semantic_type'] = 'numeric'
elif pd.api.types.is_datetime64_dtype(df[col]):
col_info['semantic_type'] = 'datetime'
elif pd.api.types.is_bool_dtype(df[col]):
col_info['semantic_type'] = 'boolean'
else:
col_info['semantic_type'] = 'string'
schema['columns'].append(col_info)
return schemadef get_basic_statistics(df: pd.DataFrame) -> dict:
"""
Get basic statistical summary.
Returns:
Dictionary with row count, column count, and column stats
"""
stats = {
'row_count': len(df),
'column_count': len(df.columns),
'columns': {}
}
for col in df.columns:
col_stats = {}
if pd.api.types.is_numeric_dtype(df[col]):
col_stats = {
'min': float(df[col].min()) if not df[col].isna().all() else None,
'max': float(df[col].max()) if not df[col].isna().all() else None,
'mean': float(df[col].mean()) if not df[col].isna().all() else None,
'median': float(df[col].median()) if not df[col].isna().all() else None,
'std': float(df[col].std()) if not df[col].isna().all() else None
}
elif pd.api.types.is_string_dtype(df[col]) or df[col].dtype == object:
non_null = df[col].dropna()
if len(non_null) > 0:
col_stats = {
'min_length': int(non_null.astype(str).str.len().min()),
'max_length': int(non_null.astype(str).str.len().max()),
'avg_length': float(non_null.astype(str).str.len().mean())
}
stats['columns'][col] = col_stats
return statsdef analyze_distributions(df: pd.DataFrame, max_unique: int = 50) -> dict:
"""
Analyze value distributions for each column.
Args:
df: DataFrame to analyze
max_unique: Max unique values to show frequencies for
Returns:
Dictionary with distribution info per column
"""
distributions = {}
for col in df.columns:
dist_info = {}
# Value counts
value_counts = df[col].value_counts()
unique_count = len(value_counts)
dist_info['unique_count'] = unique_count
dist_info['unique_percentage'] = (unique_count / len(df)) * 100
# Show top values if not too many unique
if unique_count <= max_unique:
dist_info['value_frequencies'] = {
str(k): int(v) for k, v in value_counts.head(20).items()
}
# Percentiles for numeric columns
if pd.api.types.is_numeric_dtype(df[col]):
percentiles = df[col].quantile([0.25, 0.50, 0.75, 0.90, 0.95, 0.99])
dist_info['percentiles'] = {
f'p{int(p*100)}': float(v) for p, v in percentiles.items()
}
distributions[col] = dist_info
return distributionsdef analyze_uniqueness(df: pd.DataFrame) -> dict:
"""
Analyze uniqueness characteristics of columns.
Returns:
Dictionary with uniqueness metrics per column
"""
uniqueness = {}
for col in df.columns:
unique_info = {}
total_count = len(df[col])
non_null_count = df[col].notna().sum()
unique_count = df[col].nunique()
duplicate_count = total_count - unique_count
unique_info['distinct_count'] = unique_count
unique_info['duplicate_count'] = duplicate_count
unique_info['uniqueness_ratio'] = unique_count / total_count if total_count > 0 else 0
unique_info['is_unique_key'] = (unique_count == non_null_count)
unique_info['has_duplicates'] = (duplicate_count > 0)
# Find duplicated values (top 10)
if duplicate_count > 0:
duplicates = df[col].value_counts()
duplicates = duplicates[duplicates > 1].head(10)
unique_info['top_duplicates'] = {
str(k): int(v) for k, v in duplicates.items()
}
uniqueness[col] = unique_info
return uniquenessimport re
def detect_patterns(df: pd.DataFrame) -> dict:
"""
Detect common patterns in string columns.
Returns:
Dictionary with detected patterns per column
"""
patterns = {}
# Common regex patterns
pattern_regexes = {
'email': r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$',
'phone_us': r'^\(?[0-9]{3}\)?[-.\s]?[0-9]{3}[-.\s]?[0-9]{4}$',
'url': r'^https?:\/\/(www\.)?[-a-zA-Z0-9@:%._\+~#=]{1,256}\.[a-zA-Z0-9()]{1,6}\b',
'ipv4': r'^(?:[0-9]{1,3}\.){3}[0-9]{1,3}$',
'date_iso': r'^\d{4}-\d{2}-\d{2}',
'date_us': r'^\d{1,2}\/\d{1,2}\/\d{2,4}$',
'uuid': r'^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$',
'numeric': r'^\d+(\.\d+)?$',
'alphanumeric': r'^[a-zA-Z0-9]+$'
}
for col in df.columns:
if pd.api.types.is_string_dtype(df[col]) or df[col].dtype == object:
col_patterns = {}
non_null = df[col].dropna().astype(str)
if len(non_null) == 0:
continue
# Test each pattern
for pattern_name, pattern_regex in pattern_regexes.items():
matches = non_null.str.match(pattern_regex, flags=re.IGNORECASE)
match_count = matches.sum()
match_percentage = (match_count / len(non_null)) * 100
if match_percentage > 50: # If >50% match, consider it a pattern
col_patterns[pattern_name] = {
'match_count': int(match_count),
'match_percentage': float(match_percentage)
}
if col_patterns:
patterns[col] = col_patterns
return patternsdef calculate_quality_metrics(df: pd.DataFrame) -> dict:
"""
Calculate overall data quality metrics.
Returns:
Dictionary with quality scores and metrics
"""
quality = {
'overall_completeness': 0.0,
'columns': {}
}
total_cells = len(df) * len(df.columns)
non_null_cells = df.notna().sum().sum()
quality['overall_completeness'] = (non_null_cells / total_cells) * 100 if total_cells > 0 else 0
for col in df.columns:
col_quality = {}
# Completeness
total_count = len(df[col])
non_null_count = df[col].notna().sum()
col_quality['completeness'] = (non_null_count / total_count) * 100 if total_count > 0 else 0
# Consistency (e.g., no mixed types)
if pd.api.types.is_numeric_dtype(df[col]):
col_quality['type_consistency'] = 100.0 # All numeric
elif pd.api.types.is_string_dtype(df[col]) or df[col].dtype == object:
# Check if all non-null values are same type
non_null = df[col].dropna()
if len(non_null) > 0:
types = non_null.apply(type)
col_quality['type_consistency'] = (types.value_counts().max() / len(non_null)) * 100
quality['columns'][col] = col_quality
return qualityimport pandas as pd
from typing import Dict, Any
def profile_dataset(df: pd.DataFrame) -> Dict[str, Any]:
"""
Complete dataset profiling with all metrics.
Args:
df: pandas DataFrame to profile
Returns:
Comprehensive profiling results dictionary
"""
profile = {
'metadata': {
'row_count': len(df),
'column_count': len(df.columns),
'memory_usage_mb': df.memory_usage(deep=True).sum() / (1024 * 1024)
},
'schema': detect_schema(df),
'statistics': get_basic_statistics(df),
'distributions': analyze_distributions(df),
'uniqueness': analyze_uniqueness(df),
'patterns': detect_patterns(df),
'quality': calculate_quality_metrics(df)
}
return profile
# Usage example
if __name__ == "__main__":
# Load dataset
df = pd.read_csv("data.csv")
# Profile
profile_results = profile_dataset(df)
# Print summary
print(f"Dataset Profile:")
print(f" Rows: {profile_results['metadata']['row_count']}")
print(f" Columns: {profile_results['metadata']['column_count']}")
print(f" Completeness: {profile_results['quality']['overall_completeness']:.2f}%")dataset_profile:
metadata:
row_count: 10000
column_count: 15
memory_usage_mb: 2.5
schema:
columns:
- name: customer_id
type: string
nullable: false
null_count: 0
semantic_type: string
- name: revenue
type: float64
nullable: true
null_count: 250
null_percentage: 2.5
semantic_type: numeric
quality:
overall_completeness: 95.5
columns:
customer_id:
completeness: 100.0
type_consistency: 100.0
revenue:
completeness: 97.5
type_consistency: 100.0
recommendations:
- "customer_id has 98% uniqueness - suitable as primary key"
- "revenue has 2.5% nulls - consider default value or filtering"
- "date column matches ISO 8601 format - ready for datetime conversion"