スキーマの検証とアサーション
列の範囲、NULL以外の制約、一意キーに対するアサーションチェックを作成し、すべてのパイプラインの開始時に実行します。
「スキーマの検証とアサーション」はCoddyKit上の無料Pandas & NumPy Academyレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはPandas & NumPy Academy学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 Pandas & NumPy Academyコースには全4レッスンが含まれています。
このレッスンの一部はまだ翻訳されておらず、英語で表示されています。
Why Schema Validation Matters
A data pipeline processes new data automatically, often without human review. If the schema of the input file changes — a column is renamed, a date format shifts, or a new category appears — the pipeline should fail loudly rather than produce silently wrong output. Schema validation with assertions is the mechanism that enforces contracts between data producers and data consumers, catching problems at the earliest possible moment.
import pandas as pd
import numpy as np
df = pd.read_parquet('sales_treated.parquet')
print('Loaded:', df.shape)
print(df.dtypes)Required Column Checks
The most fundamental validation is checking that all required columns are present. Store the expected column set in a config and assert that it is a subset of the actual columns. This check catches renames and drops immediately at pipeline start, before any downstream code attempts to access missing columns and raises a confusing KeyError deep in the pipeline.
REQUIRED_COLUMNS = {'order_id', 'order_date', 'customer_id',
'product', 'category', 'quantity', 'unit_price', 'revenue'}
missing = REQUIRED_COLUMNS - set(df.columns)
assert not missing, f'Missing required columns: {missing}'
print('All required columns present.')Column Data Type Assertions
After required columns, validate their data types. A date column loaded as object means pd.to_datetime() was not called. An ID column stored as float instead of int64 often means there are NaN values that prevented integer storage. Write type assertions using assert df['col'].dtype == expected_type for the most critical columns.
assert pd.api.types.is_datetime64_any_dtype(df['order_date']), \
'order_date must be datetime'
assert pd.api.types.is_numeric_dtype(df['revenue']), \
'revenue must be numeric'
assert df['order_id'].dtype == object or pd.api.types.is_integer_dtype(df['order_id']), \
'order_id must be string or int'
print('Dtype checks passed.')Non-Null Constraints
Key columns like order_id, order_date, and revenue should never be null. Assert df['col'].notna().all() for each of them. To make the assertion message actionable, include the count of null values so the pipeline operator knows the scale of the problem rather than just that an assertion failed.
NOT_NULL_COLS = ['order_id', 'order_date', 'revenue', 'customer_id']
for col in NOT_NULL_COLS:
null_count = df[col].isna().sum()
assert null_count == 0, f'{col} has {null_count} null values'
print('Non-null checks passed.')Range Checks for Numeric Columns
Numeric columns often have valid business ranges. Revenue must be non-negative. Quantity must be a positive integer. Unit price must be greater than zero. Assert these constraints explicitly — a negative revenue row that slips through will undercount totals in every downstream aggregation without any error. Range checks catch data entry errors and upstream system bugs early.
assert (df['revenue'] >= 0).all(), 'Negative revenue found'
assert (df['quantity'] > 0).all() or df['is_return'].any(), \
'Non-positive quantity without return flag'
assert (df['unit_price'] > 0).all(), 'Zero or negative unit price found'
print('Range checks passed.')Unique Key Assertions
After deduplication, order_id should be unique. Assert df['order_id'].is_unique to verify this invariant is maintained across every pipeline run. Uniqueness violations after deduplication indicate a bug in the deduplication logic or a newly introduced data source that was not cleaned before merging into the main dataset.
assert df['order_id'].is_unique, \
f'order_id not unique: {df.duplicated(subset=["order_id"]).sum()} duplicates'
print('Uniqueness check passed.')Categorical Value Assertions
For columns with a finite set of valid values — like region or category — assert that every value is in the allowed set. This catches rogue values that appear after a system migration or an upstream data entry change. Define the valid sets in your pipeline config so they are easy to update when the business expands into new regions.
VALID_REGIONS = {'North', 'South', 'East', 'West', 'Central'}
VALID_CATEGORIES = {'electronics', 'apparel', 'home', 'sports', 'beauty', 'other'}
assert df['region'].isin(VALID_REGIONS).all(), \
f'Invalid regions: {df[~df["region"].isin(VALID_REGIONS)]["region"].unique()}'
assert df['category'].isin(VALID_CATEGORIES).all(), \
'Invalid categories found'
print('Categorical checks passed.')Date Range Assertions
Validate that all dates fall within the expected period for the dataset. An order dated in the future is impossible; an order dated before the company was founded indicates a corrupted record. Define MIN_DATE and MAX_DATE in the config and assert the column falls within bounds. This also catches Unix-epoch-zero dates (1970-01-01) from bad timestamp conversions.
MIN_DATE = pd.Timestamp('2020-01-01')
MAX_DATE = pd.Timestamp('today')
assert (df['order_date'] >= MIN_DATE).all(), 'Date before minimum found'
assert (df['order_date'] <= MAX_DATE).all(), 'Future date found'
print(f'Date range: {df["order_date"].min()} to {df["order_date"].max()}')Row Count Guard
A sudden change in row count from the previous pipeline run is a strong signal of an upstream issue. Store the previous run's row count in a config file and assert that the new count is within a tolerance band — for example, within ±20 % of the historical count. A dataset that loses 50 % of its rows between runs almost certainly indicates a broken data export.
EXPECTED_MIN_ROWS = 5000
EXPECTED_MAX_ROWS = 200000
assert EXPECTED_MIN_ROWS <= len(df) <= EXPECTED_MAX_ROWS, \
f'Row count {len(df)} outside expected range [{EXPECTED_MIN_ROWS}, {EXPECTED_MAX_ROWS}]'
print(f'Row count check passed: {len(df)} rows')Packaging Checks into a Validate Function
Collect all assertions into a single validate(df) function that can be called at the start of every pipeline stage. Each check raises an AssertionError with a descriptive message if it fails. This pattern makes the pipeline self-documenting: the validation function is the machine-readable schema contract for the DataFrame at that stage.
def validate_sales_df(df):
assert not (REQUIRED_COLUMNS - set(df.columns)), 'Missing columns'
assert df['order_id'].is_unique, 'Duplicate order IDs'
assert (df['revenue'] >= 0).all(), 'Negative revenue'
assert df['region'].isin(VALID_REGIONS).all(), 'Invalid region'
print(f'Validation passed: {len(df)} rows, {len(df.columns)} columns')
validate_sales_df(df)Logging Validation Failures Gracefully
In production pipelines, a hard assert crash is acceptable during development but undesirable in a scheduled job. Replace bare assertions with try/except AssertionError blocks that log the error message and optionally send an alert before exiting. This lets the monitoring system capture the failure reason rather than just an unhandled exception traceback.
import logging
logging.basicConfig(level=logging.INFO)
def validate_with_logging(df):
checks = [
(lambda d: d['order_id'].is_unique, 'Duplicate order IDs'),
(lambda d: (d['revenue'] >= 0).all(), 'Negative revenue found'),
]
for check_fn, msg in checks:
try:
assert check_fn(df), msg
except AssertionError as e:
logging.error(f'VALIDATION FAILED: {e}')
raise
validate_with_logging(df)
print('All checks passed.')Quick Check
Test your understanding of Data Analysis concepts from this lesson.
Lesson Recap
In this lesson you learned: asserting required columns, dtypes, non-null constraints, and range validity, checking unique keys, categorical values, and date ranges, and packaging all checks into a reusable validate() function with logging. Next up we explore custom aggregations using apply() on columns and rows.
よくある質問
「スキーマの検証とアサーション」レッスンは無料ですか？
はい。「スキーマの検証とアサーション」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、Pandas & NumPy Academyコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 Pandas & NumPy Academyコースには全4レッスンが含まれています。
「スキーマの検証とアサーション」で何を学びますか？
列の範囲、NULL以外の制約、一意キーに対するアサーションチェックを作成し、すべてのパイプラインの開始時に実行します。 ブラウザで直接実行するハンズオンコードでPandas & NumPy Academyを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
Pandas & NumPy Academyを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのPandas & NumPy Academyは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。
「スキーマの検証とアサーション」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このPandas & NumPy Academyレッスンでコードを書いて実行できますか？
はい。すべてのPandas & NumPy Academyレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- 重複の検出と削除
- 外れ値の検出と処理
- 不統一なカテゴリの標準化
- スキーマの検証とアサーション