【DevOps】Elasticsearch 数据跨集群同步方案

from datetime import datetime
from elasticsearch import Elasticsearch
 
# 初始化源和目标集群的 Elasticsearch 客户端
src_es = Elasticsearch(hosts=["源集群地址"])
dest_es = Elasticsearch(hosts=["目标集群地址"])
 
def sync_data(index_name, src_index_name=None, dest_index_name=None):
    """
    同步数据从源集群到目标集群
    :param index_name: 同步的索引名
    :param src_index_name: 源集群索引名，如果不同
    :param dest_index_name: 目标集群索引名，如果不同
    """
    # 如果源和目标索引名相同，则直接使用index_name
    src_index_name = src_index_name or index_name
    dest_index_name = dest_index_name or index_name
 
    # 获取源集群索引的映射
    src_mapping = src_es.indices.get_mapping(index=src_index_name)
 
    # 在目标集群创建索引，并设置映射
    dest_es.indices.create(index=dest_index_name, body=src_mapping[src_index_name]['mappings'], ignore=400)
 
    # 查询源集群的索引，获取所有数据
    scroll = src_es.scroll(
        index=src_index_name,
        scroll='5m',
        size=1000,
        body={'query': {'match_all': {}}}
    )
 
    # 批量写入目标集群
    for hit in scroll:
        dest_es.index(index=dest_index_name, id=hit['_id'], document=hit['_source'])
 
# 调用函数同步数据
sync_data('my_index', src_index_name='my_source_index', dest_index_name='my_dest_index')

这个简化版本的代码实例展示了如何使用Elasticsearch Python API从一个Elasticsearch集群同步数据到另一个集群。它演示了如何获取索引映射、创建索引和使用滚动查询来批量获取和索引数据。这个例子可以作为开发者实现自己的数据同步解决方案的参考。