القائمة
Implement data quality checks, validation rules, and monitoring. Use when ensuring data quality, validating data pipelines, or implementing data governance.
التثبيت باستخدام Codex أو Claude انسخ هذا Prompt والصقه في Codex أو Claude أو مساعد آخر ليراجع صفحة Skill ويثبّتها لك.
استنادا إلى تصنيف SOC المهني
Creates and validates Azure Resource Manager (ARM) templates for infrastructure deployment. Use when creating ARM templates, deploying Azure infrastructure as code, or validating Azure templates.
Analyzes Azure costs and provides optimization recommendations including reserved instances, rightsizing, and unused resources. Use when optimizing Azure spending or analyzing Azure costs.
Creates dbt models with proper layering (staging, marts), incremental strategies, and documentation. Use when creating dbt models, organizing data transformations, or implementing incremental models.
Generates dbt tests including schema tests, data quality tests, and freshness checks. Use when adding tests to dbt models or implementing data quality validation.
Configures Google Cloud Build pipelines with caching, parallel builds, and optimization. Use when setting up Cloud Build, optimizing build performance, or configuring CI/CD pipelines.
Analyzes GCP costs and provides optimization recommendations including committed use discounts, rightsizing, and unused resources. Use when optimizing GCP spending or analyzing GCP costs.
| name | data-quality-checker |
| description | Implement data quality checks, validation rules, and monitoring. Use when ensuring data quality, validating data pipelines, or implementing data governance. |
Implement comprehensive data quality checks and validation.
Use Great Expectations for validation, implement schema checks, monitor data quality metrics, set up alerts.
import great_expectations as gx
context = gx.get_context()
# Create expectation suite
suite = context.add_expectation_suite("data_quality_suite")
# Add expectations
validator = context.get_validator(
batch_request=batch_request,
expectation_suite_name="data_quality_suite"
)
# Schema validation
validator.expect_table_columns_to_match_ordered_list(
column_list=["id", "name", "email", "created_at"]
)
# Null checks
validator.expect_column_values_to_not_be_null("email")
# Value ranges
validator.expect_column_values_to_be_between("age", min_value=0, max_value=120)
# Uniqueness
validator.expect_column_values_to_be_unique("email")
# Run validation
results = validator.validate()
def validate_data_quality(df):
issues = []
# Check for nulls
null_counts = df.isnull().sum()
if null_counts.any():
issues.append(f"Null values found: {null_counts[null_counts > 0]}")
# Check for duplicates
duplicates = df.duplicated().sum()
if duplicates > 0:
issues.append(f"Found {duplicates} duplicate rows")
# Check data freshness
max_date = df['created_at'].max()
if (datetime.now() - max_date).days > 1:
issues.append("Data is stale")
return issues
def calculate_quality_metrics(df):
return {
'completeness': 1 - (df.isnull().sum().sum() / df.size),
'uniqueness': df.drop_duplicates().shape[0] / df.shape[0],
'validity': (df['email'].str.contains('@').sum() / len(df)),
'timeliness': (datetime.now() - df['created_at'].max()).days
}