RDBMSとのアーキテクチャ概念対応

Elasticsearch 6.0以降、公式ガイドラインでは単一インデックス内への多様なデータ統合を推奨しており、「タイプ」層は段階的に廃止されています。現在の設計ではインデックスをテーブル等同義とし、命名規則やプレフィックスで論理的分離を行うのが一般的です。

接続検証とクラスタメタデータの取得

from elasticsearch import Elasticsearch, BasicAuth

# 認証情報とエンドポイントを設定してクライアント初期化
auth_info = BasicAuth('admin_user', 'secret_password')
elastic_client = Elasticsearch(['https://127.0.0.1:9200'], basic_auth=auth_info, verify_certs=False)

# コネクション死活確認
connection_status = elastic_client.ping()
print(f'接続成功: {connection_status}')

# クラスタ基本バージョン情報
cluster_metadata = elastic_client.info()
print(cluster_metadata['version']['number'])

インデックス一覧取得と統計カウント

# JSON形式で全インデックスの状態情報を取得
index_catalog = elastic_client.cat.indices(format='json')
for record in index_catalog:
    print(f"{'インデックス:'} {record['index']}, {'ステータス:'} {record['health']}")

# 全文書数のカウント（構造化された値のみ抽出）
total_docs = elastic_client.cat.count(index='production_data', format='json', h=['count'])
print(f'対象インデックスの総レコード数: {total_docs[0].get("count", 0)}')

スキーマ定義によるインデックス構築

マッピング設計において、type はフィールドのデータ型を示します。代表的な型と制御パラメータは以下の通りです。

• text: 前方一致・完全一致が可能。トークン分割され、メインコンテンツ検索に最適。
• keyword: 等価照合・集計・ソート向け。構造化済み文字列（ID、タグ、ステータス）に適す。
• date/integer/float: 各データ種別に対応し、形式フォーマット指定が可能です。

追加制御パラメータ:

# カスタムマッピング定義
schema_definition = {
    "mappings": {
        "user_profile": {
            "properties": {
                "entity_id": { "type": "keyword" },
                "full_name": { "type": "keyword" },
                "account_status": { "type": "keyword", "index": False },
                "registration_age": { "type": "integer", "ignore_malformed": True },
                "join_timestamp": {
                    "type": "date",
                    "format": "yyyy-MM-dd HH:mm:ss||epoch_second",
                    "store": True
                }
            }
        }
    }
}

# スキーマ付きインデックス生成（既存時は400エラーを無視）
create_response = elastic_client.indices.create(
    index='corporate_db',
    body=schema_definition,
    ignore=400
)
print(create_response)

マッピング修正とインデックス維持処理

# 新規フィールドの動的追加
field_expansion = {
    "user_profile": {
        "properties": {
            "last_login_epoch": { "type": "long" }
        }
    }
}
elastic_client.indices.put_mapping(index='corporate_db', doc_type='user_profile', body=field_expansion)

# 直近の書き込みをリアルタイム検索対象へ反映
elastic_client.indices.refresh(index='corporate_db')

# 不要インデックスの削除（存在しない場合は404を静観）
elastic_client.indices.delete(index='legacy_backup_v1', ignore=[400, 404])

ドキュメント登録・更新・一意参照

import time, uuid

# 新規登録（自動採番または明示ID付与）
generate_uid = str(uuid.uuid4())[:16]
target_record = {
    "entity_id": generate_uid,
    "full_name": "Akira Tanaka",
    "account_status": "active",
    "registration_age": 28,
    "join_timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
}

register_result = elastic_client.index(
    index='corporate_db',
    doc_type='user_profile',
    id=generate_uid,
    body=target_record,
    refresh=True
)
print(register_result['result'])  # created / updated

# 同一IDでの完全置換
updated_record = {
    "entity_id": generate_uid,
    "full_name": "Akira T.",
    "account_status": "inactive",
    "registration_age": 29,
    "join_timestamp": "2023-11-15 09:00:00"
}
elastic_client.index(index='corporate_db', doc_type='user_profile', id=generate_uid, body=updated_record, refresh=True)

# UUIDキーによる単一ドキュメント取得
fetch_result = elastic_client.get(index='corporate_db', doc_type='user_profile', id=generate_uid)
print(fetch_result['_source'])

条件指定検索と複雑な問い合わせ処理

# 検索ペイロード構成（モジュール分けして可読性向上）
filter_clause = {
    "query": {
        "match_phrase": {
            "content_body": "技術文書"  # 完全フレーズ一致
        }
    }
}

pagination_controls = {
    "size": 20,
    "from": 0,
    "search_after": [],  # 最終ページの末尾ソート値を埋め込むことで深層スクロール実現
    "sort": [
        { "_seq_no": { "order": "desc" } }
    ]
}

highlight_settings = {
    "require_field_match": False,
    "fields": {
        "content_body": {},
        "document_title": {}
    }
}

complete_query = filter_clause | pagination_controls | {"highlight": highlight_settings}

execution_output = elastic_client.search(index='corporate_db', body=complete_query)

# 結果解析
hits_container = execution_output['hits']['hits']
for item in hits_container:
    doc_meta = item['_source']
    matched_chunks = item.get('highlight', {})
    print(doc_meta.get('document_title'), matched_chunks.get('content_body'))