# ===== データ型のチェック ===== import pandas as pd df = pd.DataFrame({ ‘user_id’: [1, 2, 3], ‘name’: [‘山田’, ‘佐藤’, ‘鈴木’], ‘age’: [25, 30, 28], ‘email’: [‘yamada@ex.com’, ‘sato@ex.com’, ‘suzuki@ex.com’] }) # 期待されるスキーマ expected_schema = { ‘user_id’: ‘int64’, ‘name’: ‘object’, ‘age’: ‘int64’, ‘email’: ‘object’ } print(“=== 現在のスキーマ ===”) print(df.dtypes) print(“\n=== スキーマ検証 ===”) for col, expected_dtype in expected_schema.items(): if col not in df.columns: print(f”❌ カラム ‘{col}’ が存在しません”) elif str(df[col].dtype) != expected_dtype: print(f”❌ カラム ‘{col}’ の型が違います: {df[col].dtype} (期待: {expected_dtype})”) else: print(f”✅ カラム ‘{col}’ OK ({expected_dtype})”)

# ===== 必須カラムの確認 ===== import pandas as pd df = pd.DataFrame({ ‘user_id’: [1, 2, 3], ‘name’: [‘山田’, ‘佐藤’, ‘鈴木’] # emailとageがない！ }) # 必須カラムのリスト required_columns = [‘user_id’, ‘name’, ‘email’, ‘age’] print(“=== 必須カラムチェック ===”) missing_columns = [] for col in required_columns: if col in df.columns: print(f”✅ ‘{col}’ 存在”) else: print(f”❌ ‘{col}’ 不足”) missing_columns.append(col) if missing_columns: print(f”\n⚠️ 不足カラム: {missing_columns}”) else: print(“\n✅ すべての必須カラムが存在します”)

# ===== 再利用可能な検証関数 ===== import pandas as pd def validate_schema(df, schema): “”” DataFrameのスキーマを検証 Args: df: 検証するDataFrame schema: 期待されるスキーマ（辞書: {カラム名: データ型}） Returns: tuple: (成功フラグ, エラーリスト) “”” errors = [] # カラムの存在チェック for col in schema.keys(): if col not in df.columns: errors.append(f”カラム ‘{col}’ が存在しません”) # データ型チェック for col, expected_dtype in schema.items(): if col in df.columns: actual_dtype = str(df[col].dtype) if actual_dtype != expected_dtype: errors.append( f”カラム ‘{col}’ の型が違います: ” f”{actual_dtype} (期待: {expected_dtype})” ) # 余分なカラムの警告（エラーではない） extra_cols = set(df.columns) – set(schema.keys()) if extra_cols: print(f”ℹ️ 追加カラム（スキーマ外）: {extra_cols}”) # 結果 if errors: print(“❌ スキーマ検証失敗:”) for error in errors: print(f” – {error}”) return False, errors else: print(“✅ スキーマ検証成功”) return True, [] # ===== 使用例 ===== df = pd.DataFrame({ ‘user_id’: [1, 2, 3], ‘name’: [‘山田’, ‘佐藤’, ‘鈴木’], ‘age’: [25, 30, 28], ‘extra_col’: [‘x’, ‘y’, ‘z’] # スキーマ外のカラム }) schema = { ‘user_id’: ‘int64’, ‘name’: ‘object’, ‘age’: ‘int64’, ‘email’: ‘object’ # dfに存在しない } success, errors = validate_schema(df, schema)

# ===== 値が妥当な範囲内かチェック ===== import pandas as pd df = pd.DataFrame({ ‘user_id’: [1, 2, 3, 4, 5], ‘age’: [25, 30, -5, 150, 28], # -5と150は異常 ‘price’: [1000, 1500, 0, 2000, -100] # 0と-100は異常 }) print(“=== 元データ ===”) print(df) # 年齢の範囲チェック（0〜120） print(“\n=== 年齢の範囲チェック（0〜120）===”) age_invalid = df[(df[‘age’] < 0) | (df['age'] > 120)] if len(age_invalid) > 0: print(f”❌ 範囲外のデータ: {len(age_invalid)}件”) print(age_invalid[[‘user_id’, ‘age’]]) else: print(“✅ すべて範囲内”) # 価格の範囲チェック（1以上） print(“\n=== 価格の範囲チェック（1以上）===”) price_invalid = df[df[‘price’] < 1] if len(price_invalid) > 0: print(f”❌ 範囲外のデータ: {len(price_invalid)}件”) print(price_invalid[[‘user_id’, ‘price’]]) else: print(“✅ すべて範囲内”)

# ===== 欠損値のチェック ===== import pandas as pd import numpy as np df = pd.DataFrame({ ‘user_id’: [1, 2, 3, 4], ‘name’: [‘山田’, np.nan, ‘鈴木’, ‘田中’], # 1件欠損 ‘email’: [‘yamada@ex.com’, ‘sato@ex.com’, None, ‘tanaka@ex.com’], # 1件欠損 ‘age’: [25, 30, np.nan, np.nan] # 2件欠損 }) print(“=== 元データ ===”) print(df) # 必須カラムのNULLチェック print(“\n=== 必須カラムのNULLチェック ===”) required_columns = [‘user_id’, ‘name’, ‘email’] for col in required_columns: null_count = df[col].isnull().sum() total_count = len(df) null_pct = (null_count / total_count * 100) if null_count > 0: print(f”❌ ‘{col}’: {null_count}件の欠損 ({null_pct:.1f}%)”) else: print(f”✅ ‘{col}’: 欠損なし”) # 全体の欠損値サマリー print(“\n=== 全体の欠損値サマリー ===”) null_summary = df.isnull().sum() null_summary = null_summary[null_summary > 0] print(null_summary)

# ===== 重複データのチェック ===== import pandas as pd df = pd.DataFrame({ ‘user_id’: [1, 2, 3, 2, 4], # user_id=2が重複 ‘email’: [‘a@ex.com’, ‘b@ex.com’, ‘c@ex.com’, ‘b@ex.com’, ‘d@ex.com’], ‘name’: [‘山田’, ‘佐藤’, ‘鈴木’, ‘佐藤’, ‘田中’] }) print(“=== 元データ ===”) print(df) # 特定カラムの重複チェック print(“\n=== user_idの重複チェック ===”) dup_user_id = df[df[‘user_id’].duplicated(keep=False)] if len(dup_user_id) > 0: print(f”❌ 重複あり: {len(dup_user_id)}件”) print(dup_user_id) else: print(“✅ 重複なし”) # 行全体の重複チェック print(“\n=== 行全体の重複チェック ===”) dup_rows = df[df.duplicated(keep=False)] if len(dup_rows) > 0: print(f”❌ 重複行: {len(dup_rows)}件”) print(dup_rows) else: print(“✅ 重複行なし”)

# ===== 包括的な品質チェック関数 ===== import pandas as pd import numpy as np def data_quality_report(df): “”” DataFrameの品質レポートを生成 “”” print(“=” * 60) print(“データ品質レポート”) print(“=” * 60) # 基本情報 print(f”\n【1. 基本情報】”) print(f” 行数: {len(df):,}”) print(f” 列数: {len(df.columns)}”) print(f” メモリ使用量: {df.memory_usage(deep=True).sum() / 1024:.1f} KB”) # 欠損値 print(f”\n【2. 欠損値】”) null_counts = df.isnull().sum() null_pct = (null_counts / len(df) * 100).round(2) null_df = pd.DataFrame({ ‘欠損数’: null_counts, ‘欠損率(%)’: null_pct }) null_df = null_df[null_df[‘欠損数’] > 0] if len(null_df) > 0: print(null_df) else: print(” 欠損値なし ✅”) # 重複 print(f”\n【3. 重複】”) dup_count = df.duplicated().sum() if dup_count > 0: print(f” 重複行数: {dup_count} ❌”) else: print(f” 重複行なし ✅”) # データ型 print(f”\n【4. データ型】”) for col in df.columns: print(f” {col}: {df[col].dtype}”) # 数値カラムの統計 numeric_cols = df.select_dtypes(include=[np.number]).columns if len(numeric_cols) > 0: print(f”\n【5. 数値カラムの統計】”) stats = df[numeric_cols].describe().T[[‘min’, ‘max’, ‘mean’, ‘std’]] print(stats.round(2)) print(“\n” + “=” * 60) # ===== 使用例 ===== df = pd.DataFrame({ ‘user_id’: [1, 2, 3, 2, 4], ‘age’: [25, np.nan, 30, 28, 35], ‘price’: [1000, 1500, 2000, 1500, 2500], ‘name’: [‘山田’, ‘佐藤’, ‘鈴木’, ‘佐藤’, ‘田中’] }) data_quality_report(df)

# ===== Great Expectationsの基本的な使い方 ===== import pandas as pd import great_expectations as gx # データ作成 df = pd.DataFrame({ ‘user_id’: [1, 2, 3, 4, 5], ‘age’: [25, 30, 28, 35, 40], ‘email’: [‘a@ex.com’, ‘b@ex.com’, ‘c@ex.com’, ‘d@ex.com’, ‘e@ex.com’] }) # コンテキストを作成（一時的なもの） context = gx.get_context() # DataFrameをデータソースとして追加 datasource = context.sources.add_pandas(“pandas_datasource”) data_asset = datasource.add_dataframe_asset(name=”my_data”) batch_request = data_asset.build_batch_request(dataframe=df) # Validatorを取得（検証器） validator = context.get_validator(batch_request=batch_request) # 検証ルールを定義して実行 print(“=== 検証実行 ===”) # カラムの存在チェック result1 = validator.expect_column_to_exist(“user_id”) print(f”user_idが存在: {result1.success}”) # NULL値がないことを確認 result2 = validator.expect_column_values_to_not_be_null(“user_id”) print(f”user_idにNULLなし: {result2.success}”) # 範囲チェック result3 = validator.expect_column_values_to_be_between(“age”, min_value=0, max_value=120) print(f”ageが0-120の範囲内: {result3.success}”) # 全体の検証実行 results = validator.validate() print(f”\n=== 総合結果 ===”) print(f”成功: {‘✅’ if results.success else ‘❌’}”)

# ===== 完全な品質チェックシステム ===== import pandas as pd import numpy as np from datetime import datetime def comprehensive_data_validation(df, config): “”” 包括的なデータ検証を実行 Args: df: 検証するDataFrame config: 検証設定（辞書） – schema: {カラム名: データ型} – required_columns: [必須カラムのリスト] – ranges: {カラム名: (最小値, 最大値)} – unique_columns: [一意であるべきカラムのリスト] Returns: dict: 検証結果 “”” results = { ‘timestamp’: datetime.now().strftime(‘%Y-%m-%d %H:%M:%S’), ‘total_rows’: len(df), ‘total_columns’: len(df.columns), ‘checks’: [], ‘passed’: 0, ‘failed’: 0 } print(“=” * 60) print(“データ品質検証レポート”) print(“=” * 60) print(f”実行日時: {results[‘timestamp’]}”) print(f”行数: {results[‘total_rows’]:,}”) print(f”列数: {results[‘total_columns’]}”) # 1. スキーマチェック print(“\n【1. スキーマチェック】”) if ‘schema’ in config: for col, expected_dtype in config[‘schema’].items(): if col not in df.columns: print(f” ❌ カラム ‘{col}’ が存在しません”) results[‘failed’] += 1 elif str(df[col].dtype) != expected_dtype: print(f” ❌ ‘{col}’ の型: {df[col].dtype} (期待: {expected_dtype})”) results[‘failed’] += 1 else: print(f” ✅ ‘{col}’ OK ({expected_dtype})”) results[‘passed’] += 1 # 2. 欠損値チェック print(“\n【2. 欠損値チェック】”) if ‘required_columns’ in config: for col in config[‘required_columns’]: if col in df.columns: null_count = df[col].isnull().sum() if null_count > 0: print(f” ❌ ‘{col}’ に欠損値: {null_count}件 ({null_count/len(df)*100:.1f}%)”) results[‘failed’] += 1 else: print(f” ✅ ‘{col}’ 欠損値なし”) results[‘passed’] += 1 # 3. 範囲チェック print(“\n【3. 範囲チェック】”) if ‘ranges’ in config: for col, (min_val, max_val) in config[‘ranges’].items(): if col in df.columns: invalid = df[(df[col] < min_val) | (df[col] > max_val)] invalid_count = len(invalid) if invalid_count > 0: print(f” ❌ ‘{col}’ が範囲外: {invalid_count}件 ({min_val}〜{max_val})”) results[‘failed’] += 1 else: print(f” ✅ ‘{col}’ 範囲内 ({min_val}〜{max_val})”) results[‘passed’] += 1 # 4. 重複チェック print(“\n【4. 重複チェック】”) if ‘unique_columns’ in config: for col in config[‘unique_columns’]: if col in df.columns: dup_count = df[col].duplicated().sum() if dup_count > 0: print(f” ❌ ‘{col}’ に重複: {dup_count}件”) results[‘failed’] += 1 else: print(f” ✅ ‘{col}’ 重複なし”) results[‘passed’] += 1 # サマリー print(“\n” + “=” * 60) print(“検証サマリー”) print(“=” * 60) total_checks = results[‘passed’] + results[‘failed’] print(f”チェック総数: {total_checks}”) print(f”成功: {results[‘passed’]} ({results[‘passed’]/total_checks*100:.0f}%)”) print(f”失敗: {results[‘failed’]} ({results[‘failed’]/total_checks*100:.0f}%)”) overall_success = results[‘failed’] == 0 print(f”\n総合結果: {‘✅ 合格’ if overall_success else ‘❌ 不合格’}”) results[‘overall_success’] = overall_success return results # ========== 使用例 ========== # テストデータ（意図的に問題を含む） df = pd.DataFrame({ ‘user_id’: [1, 2, 3, 2, 4], # 重複あり ‘name’: [‘山田’, np.nan, ‘鈴木’, ‘佐藤’, ‘田中’], # 欠損値あり ‘age’: [25, 30, -5, 28, 35], # 範囲外あり ‘email’: [‘a@ex.com’, ‘b@ex.com’, ‘c@ex.com’, ‘d@ex.com’, ‘e@ex.com’] }) # 検証設定 validation_config = { ‘schema’: { ‘user_id’: ‘int64’, ‘name’: ‘object’, ‘age’: ‘int64’, ‘email’: ‘object’ }, ‘required_columns’: [‘user_id’, ‘name’, ‘email’], ‘ranges’: { ‘age’: (0, 120) }, ‘unique_columns’: [‘user_id’, ‘email’] } # 検証実行 results = comprehensive_data_validation(df, validation_config)