0
  • 聊天消息
  • 系统消息
  • 评论与回复
登录后你可以
  • 下载海量资料
  • 学习在线课程
  • 观看技术视频
  • 写文章/发帖/加入社区
会员中心
创作中心

完善资料让更多小伙伴认识你,还能领取20积分哦,立即完善>

3天内不再提示

跨机房ES同步实战

OSC开源社区 来源:OSCHINA 社区 作者:京东云开发者-谢泽 2022-12-13 15:10 次阅读

背景众所周知单个机房在出现不可抗拒的问题(如断电、断网等因素)时,会导致无法正常提供服务,会对业务造成潜在的损失。所以在协同办公领域,一种可以基于同城或异地多活机制的高可用设计,在保障数据一致性的同时,能够最大程度降低由于机房的仅单点可用所导致的潜在高可用问题,最大程度上保障业务的用户体验,降低单点问题对业务造成的潜在损失显得尤为重要。同城双活,对于生产的高可用保障,重大的意义和价值是不可言喻的。表面上同城双活只是简单的部署了一套生产环境而已,但是在架构上,这个改变的影响是巨大的,无状态应用的高可用管理、请求流量的管理、版本发布的管理、网络架构的管理等,其提升的架构复杂度巨大。结合真实的协同办公产品:京办(为北京市政府提供协同办公服务的综合性平台)生产环境面对的复杂的政务网络以及京办同城双活架构演进的案例,给大家介绍下京办持续改进、分阶段演进过程中的一些思考和实践经验的总结。本文仅针对 ES 集群在跨机房同步过程中的方案和经验进行介绍和总结。

架构

1.部署 Logstash 在金山云机房上,Logstash 启动多个实例(按不同的类型分类,提高同步效率),并且和金山云机房的 ES 集群在相同的 VPC2.Logstash 需要配置大网访问权限,保证 Logstash 和 ES 原集群和目标集群互通。3.数据迁移可以全量迁移和增量迁移,首次迁移都是全量迁移后续的增加数据选择增量迁移。4.增量迁移需要改造增加识别的增量数据的标识,具体方法后续进行介绍。ab134d5c-7a83-11ed-8abf-dac502259ad0.png原理

Logstash 工作原理

ab315e3c-7a83-11ed-8abf-dac502259ad0.pngLogstash 分为三个部分 input 、filter、ouput:1.input 处理接收数据,数据可以来源 ES,日志文件,kafka 等通道.2.filter 对数据进行过滤,清洗。3.ouput 输出数据到目标设备,可以输出到 ES,kafka,文件等。

增量同步原理

1. 对于 T 时刻的数据,先使用 Logstash 将 T 以前的所有数据迁移到有孚机房京东云 ES,假设用时∆T2. 对于 T 到 T+∆T 的增量数据,再次使用 logstash 将数据导入到有孚机房京东云的 ES 集群3. 重复上述步骤 2,直到∆T 足够小,此时将业务切换到华为云,最后完成新增数据的迁移适用范围:ES 的数据中带有时间戳或者其他能够区分新旧数据的标签

流程

ab61eb2e-7a83-11ed-8abf-dac502259ad0.png

准备工作

1.创建 ECS 和安装 JDK 忽略,自行安装即可2.下载对应版本的 Logstash,尽量选择与 Elasticsearch 版本一致,或接近的版本安装即可https://www.elastic.co/cn/downloads/logstash

1)源码下载直接解压安装包,开箱即用

2)修改对内存使用,logstash 默认的堆内存是 1G,根据 ECS 集群选择合适的内存,可以加快集群数据的迁移效率。

ab81d11e-7a83-11ed-8abf-dac502259ad0.png

3. 迁移索引

Logstash 会帮助用户自动创建索引,但是自动创建的索引和用户本身的索引会有些许差异,导致最终数据的搜索格式不一致,一般索引需要手动创建,保证索引的数据完全一致。

以下提供创建索引的 python 脚本,用户可以使用该脚本创建需要的索引。

create_mapping.py 文件是同步索引的 python 脚本,config.yaml 是集群地址配置文件。

注:使用该脚本需要安装相关依赖

yum install -y PyYAML
yum install -y python-requests

拷贝以下代码保存为 create_mapping.py:

import yaml
import requests
import json
import getopt
import sys

defhelp():
    print
    """
    usage:
    -h/--help print this help.
    -c/--config config file path, default is config.yaml
    
    example:  
    python create_mapping.py -c config.yaml 
    """
defprocess_mapping(index_mapping, dest_index):
    print(index_mapping)
    # remove unnecessary keys
    del index_mapping["settings"]["index"]["provided_name"]
    del index_mapping["settings"]["index"]["uuid"]
    del index_mapping["settings"]["index"]["creation_date"]
    del index_mapping["settings"]["index"]["version"]

    # check alias
    aliases = index_mapping["aliases"]
    for alias inlist(aliases.keys()):
        if alias == dest_index:
            print(
                "source index "+ dest_index +" alias "+ alias +" is the same as dest_index name, will remove this alias.")
            del index_mapping["aliases"][alias]
    if index_mapping["settings"]["index"].has_key("lifecycle"):
        lifecycle = index_mapping["settings"]["index"]["lifecycle"]
        opendistro ={"opendistro":{"index_state_management":
                                         {"policy_id": lifecycle["name"],
                                          "rollover_alias": lifecycle["rollover_alias"]}}}
        index_mapping["settings"].update(opendistro)
        # index_mapping["settings"]["opendistro"]["index_state_management"]["rollover_alias"] = lifecycle["rollover_alias"]
        del index_mapping["settings"]["index"]["lifecycle"]
    print(index_mapping)
    return index_mapping
defput_mapping_to_target(url, mapping, source_index, dest_auth=None):
    headers ={'Content-Type':'application/json'}
    create_resp = requests.put(url, headers=headers, data=json.dumps(mapping), auth=dest_auth)
    if create_resp.status_code !=200:
        print(
            "create index "+ url +" failed with response: "+str(create_resp)+", source index is "+ source_index)
        print(create_resp.text)
        withopen(source_index +".json","w")as f:
            json.dump(mapping, f)
defmain():
    config_yaml ="config.yaml"
    opts, args = getopt.getopt(sys.argv[1:],'-h-c:',['help','config='])
    for opt_name, opt_value in opts:
        if opt_name in('-h','--help'):
            help()
            exit()
        if opt_name in('-c','--config'):
            config_yaml = opt_value

    config_file =open(config_yaml)
    config = yaml.load(config_file)
    source = config["source"]
    source_user = config["source_user"]
    source_passwd = config["source_passwd"]
    source_auth =None
    if source_user !="":
        source_auth =(source_user, source_passwd)
    dest = config["destination"]
    dest_user = config["destination_user"]
    dest_passwd = config["destination_passwd"]
    dest_auth =None
    if dest_user !="":
        dest_auth =(dest_user, dest_passwd)
    print(source_auth)
    print(dest_auth)

    # only deal with mapping list
    if config["only_mapping"]:
        for source_index, dest_index in config["mapping"].iteritems():
            print("start to process source index"+ source_index +", target index: "+ dest_index)
            source_url = source +"/"+ source_index
            response = requests.get(source_url, auth=source_auth)
            if response.status_code !=200:
                print("*** get ElasticSearch message failed. resp statusCode:"+str(
                    response.status_code)+" response is "+ response.text)
                continue
            mapping = response.json()
            index_mapping = process_mapping(mapping[source_index], dest_index)

            dest_url = dest +"/"+ dest_index
            put_mapping_to_target(dest_url, index_mapping, source_index, dest_auth)
            print("process source index "+ source_index +" to target index "+ dest_index +" successed.")
    else:
        # get all indices
        response = requests.get(source +"/_alias", auth=source_auth)
        if response.status_code !=200:
            print("*** get all index failed. resp statusCode:"+str(
                response.status_code)+" response is "+ response.text)
            exit()
        all_index = response.json()
        for index inlist(all_index.keys()):
            if"."in index:
                continue
            print("start to process source index"+ index)
            source_url = source +"/"+ index
            index_response = requests.get(source_url, auth=source_auth)
            if index_response.status_code !=200:
                print("*** get ElasticSearch message failed. resp statusCode:"+str(
                    index_response.status_code)+" response is "+ index_response.text)
                continue
            mapping = index_response.json()

            dest_index = index
            if index in config["mapping"].keys():
                dest_index = config["mapping"][index]
            index_mapping = process_mapping(mapping[index], dest_index)

            dest_url = dest +"/"+ dest_index
            put_mapping_to_target(dest_url, index_mapping, index, dest_auth)
            print("process source index "+ index +" to target index "+ dest_index +" successed.")

if __name__ =='__main__':
    main()

配置文件保存为 config.yaml:

# 源端ES集群地址,加上http://
source: http://ip:port
source_user: "username"
source_passwd: "password"
# 目的端ES集群地址,加上http://
destination: http://ip:port
destination_user: "username"
destination_passwd: "password"

# 是否只处理这个文件中mapping地址的索引
# 如果设置成true,则只会将下面的mapping中的索引获取到并在目的端创建
# 如果设置成false,则会取源端集群的所有索引,除去(.kibana)
# 并且将索引名称与下面的mapping匹配,如果匹配到使用mapping的value作为目的端的索引名称
# 如果匹配不到,则使用源端原始的索引名称
only_mapping: true

# 要迁移的索引,key为源端的索引名字,value为目的端的索引名字
mapping:
    source_index: dest_index

以上代码和配置文件准备完成,直接执行 python create_mapping.py 即可完成索引同步。

索引同步完成可以取目标集群的 kibana 上查看或者执行 curl 查看索引迁移情况:

GET _cat/indices?v

ab95c520-7a83-11ed-8abf-dac502259ad0.png

全量迁移Logstash 配置位于 config 目录下。用户可以参考配置修改 Logstash 配置文件,为了保证迁移数据的准确性,一般建议建立多组 Logstash,分批次迁移数据,每个 Logstash 迁移部分数据。配置集群间迁移配置参考:abc4a7aa-7a83-11ed-8abf-dac502259ad0.png



input{
    elasticsearch{
        # 源端地址
        hosts =>  ["ip1:port1","ip2:port2"]
        # 安全集群配置登录用户名密码
        user => "username"
        password => "password"
        # 需要迁移的索引列表,以逗号分隔,支持通配符
        index => "a_*,b_*"
        # 以下三项保持默认即可,包含线程数和迁移数据大小和logstash jvm配置相关
        docinfo=>true
        slices => 10
        size => 2000
        scroll => "60m"
    }
}

filter {
  # 去掉一些logstash自己加的字段
  mutate {
    remove_field => ["@timestamp", "@version"]
  }
}

output{
    elasticsearch{
        # 目的端es地址
        hosts => ["http://ip:port"]
        # 安全集群配置登录用户名密码
        user => "username"
        password => "password"
 # 目的端索引名称,以下配置为和源端保持一致
        index => "%{[@metadata][_index]}"
        # 目的端索引type,以下配置为和源端保持一致
        document_type => "%{[@metadata][_type]}"
        # 目标端数据的_id,如果不需要保留原_id,可以删除以下这行,删除后性能会更好
        document_id => "%{[@metadata][_id]}"
        ilm_enabled => false
        manage_template => false
    }

    # 调试信息,正式迁移去掉
    stdout { codec => rubydebug { metadata => true }}
}

增量迁移

预处理:

1.@timestamp在 elasticsearch2.0.0beta 版本后弃用

https://www.elastic.co/guide/en/elasticsearch/reference/2.4/mapping-timestamp-field.html

2. 本次对于京办从金山云机房迁移到京东有孚机房,所涉及到的业务领域多,各个业务线中所代表新增记录的时间戳字段不统一,所涉及到的兼容工作量大,于是考虑通过 elasticsearch 中预处理功能 pipeline 进行预处理添加统一增量标记字段:gmt_created_at,以减少迁移工作的复杂度(各自业务线可自行评估是否需要此步骤)。

PUT _ingest/pipeline/gmt_created_at
{
  "description":"Adds gmt_created_at timestamp to documents",
  "processors":[
    {
      "set":{
        "field":"_source.gmt_created_at",
        "value":"{{_ingest.timestamp}}"
      }
    }
  ]
}

3. 检查 pipeline 是否生效

GET _ingest/pipeline/*

4. 各个 index 设置对应 settings 增加 pipeline 为默认预处理

PUT index_xxxx/_settings
{
  "settings": {
    "index.default_pipeline": "gmt_created_at"
  }
}

5. 检查新增 settings 是否生效

GET index_xxxx/_settings

ac0e3f46-7a83-11ed-8abf-dac502259ad0.png

增量迁移脚本

schedule-migrate.conf

index:可以使用通配符的方式

query: 增量同步的 DSL,统一 gmt_create_at 为增量同步的特殊标记

schedule: 每分钟同步一把,"* * * * *"

input {
elasticsearch {
        hosts =>["ip:port"]
        # 安全集群配置登录用户名密码
        user =>"username"
        password =>"password"
        index =>"index_*"
        query =>'{"query":{"range":{"gmt_create_at":{"gte":"now-1m","lte":"now/m"}}}}'
        size =>5000
        scroll =>"5m"
        docinfo =>true
        schedule =>"* * * * *"
      }
}
filter {
     mutate {
      remove_field =>["source", "@version"]
   }
}
output {
    elasticsearch {
        # 目的端es地址
        hosts =>["http://ip:port"]
        # 安全集群配置登录用户名密码
        user =>"username"
        password =>"password"
        index =>"%{[@metadata][_index]}"
        document_type =>"%{[@metadata][_type]}"
        document_id =>"%{[@metadata][_id]}"
        ilm_enabled =>false
        manage_template =>false
    }

# 调试信息,正式迁移去掉
stdout { codec => rubydebug { metadata =>true}}
}

问题:

mapping 中存在 join 父子类型的字段,直接迁移报 400 异常ac4a1f70-7a83-11ed-8abf-dac502259ad0.png  
[2022-09-20T20:02:16,404][WARN ][logstash.outputs.elasticsearch] Could not index event to Elasticsearch. {:status=>400, 
:action=>["index", {:_id=>"xxx", :_index=>"xxx", :_type=>"joywork_t_work", :routing=>nil}, #], 
:response=>{"index"=>{"_index"=>"xxx", "_type"=>"xxx", "_id"=>"xxx", "status"=>400, 
"error"=>{"type"=>"mapper_parsing_exception", "reason"=>"failed to parse", 
"caused_by"=>{"type"=>"illegal_argument_exception", "reason"=>"[routing] is missing for join field [task_user]"}}}}}

解决方法:

https://discuss.elastic.co/t/an-routing-missing-exception-is-obtained-when-reindex-sets-the-routing-value/155140https://github.com/elastic/elasticsearch/issues/26183

结合业务特征,通过在 filter 中加入小量的 ruby 代码,将_routing 的值取出来,放回 logstah event 中,由此问题得以解决。

示例:

ac725e7c-7a83-11ed-8abf-dac502259ad0.png


审核编辑 :李倩


声明:本文内容及配图由入驻作者撰写或者入驻合作网站授权转载。文章观点仅代表作者本人,不代表电子发烧友网立场。文章及其配图仅供工程师学习之用,如有内容侵权或者其他违规问题,请联系本站处理。 举报投诉
  • 数据
    +关注

    关注

    8

    文章

    6888

    浏览量

    88823
  • 数据迁移
    +关注

    关注

    0

    文章

    68

    浏览量

    6938

原文标题:跨机房 ES 同步实战

文章出处:【微信号:OSC开源社区,微信公众号:OSC开源社区】欢迎添加关注!文章转载请注明出处。

收藏 人收藏

    评论

    相关推荐

    机房精密空调安装指南

    1、确认精密空调型号和规格是否符合机房需求,确保精密空调能够满足机房的制冷、除湿、加湿等需求。
    的头像 发表于 10-25 17:44 142次阅读
    <b class='flag-5'>机房</b>精密空调安装指南

    控温高手:小型机房精密空调

    随着信息技术的快速发展,各类电子设备在小型机房中的应用越来越广泛。为了保证设备稳定运行,机房环境必须保持恒温恒湿。因此,小型机房恒温恒湿专用空调解决方案显得尤为重要。
    的头像 发表于 10-12 18:19 153次阅读
    控温高手:小型<b class='flag-5'>机房</b>精密空调

    动环监控主机应用机房

    信息化高速发展的时代,数据中心作为信息社会的“心脏”,其稳定运行与高效管理直接关系到各行各业的发展。随着云计算、大数据、物联网等技术的不断融合与创新,智慧机房的概念应运而生,动环监控主机的引入,正是
    的头像 发表于 09-14 15:59 251次阅读

    RobustRIO-E模块 时钟同步&分发,实现声音与振动板卡间及机箱时钟同步

    同步时钟发生器 + 同步时钟分发器
    的头像 发表于 09-14 15:00 239次阅读
    RobustRIO-E模块 时钟<b class='flag-5'>同步</b>&分发,实现声音与振动板卡间及<b class='flag-5'>跨</b>机箱时钟<b class='flag-5'>同步</b>

    机房监控,机房监控系统百科

    一、引言 随着信息技术的飞速发展,数据中心和机房作为支撑企业运营和存储关键数据的基础设施,其重要性日益凸显。机房环境的稳定性、安全性及设备的运行状态直接影响到企业的业务连续性和数据安全性。因此,建立
    的头像 发表于 09-02 14:32 288次阅读

    机房监控,机房监控百科

    机房监控是现代数据中心管理不可或缺的一部分,它直接关系到系统的稳定运行、数据的安全保护以及故障的快速响应。一个完善的机房监控系统能够实时监测机房内的环境参数、设备状态及安全情况,确保数据中心高效、可靠地运行。以下是一篇关于
    的头像 发表于 08-22 17:34 259次阅读

    机房托管费用贵吗?机房托管要考虑哪些因素?

     机房托管费用受多种因素影响,包括地理位置、设备规模、服务水平、安全性要求等。不同配置和服务质量的托管价格差异较大,一般1U服务器托管费用一年在2000到5000元之间。Rak部落为您整理发布机房托管费用的差异,希望对您选择机房
    的头像 发表于 08-16 11:34 424次阅读

    机房托管的好处

    机房托管提供的专业级服务和便利性,是其受到企业青睐的重要原因。下面将详细探讨机房托管的多重优势,并了解这种服务模式如何帮助企业提升运营效率和安全保障。Rak部落为您整理发布机房托管的好处。
    的头像 发表于 08-08 10:08 181次阅读

    机房配线架是干嘛的

    机房配线架是机房内用于管理和连接网络线路的重要设备。它的主要作用和功能包括以下几个方面: 一、定义与作用 定义:配线架是用于终端用户线或中继线,并能对它们进行调配连接的设备。它是管理子系统中最
    的头像 发表于 07-17 10:06 284次阅读

    机房动力环境监控方案

    机房动力环境监控系统是现代信息技术基础设施中不可或缺的一环。它负责对机房内的电力、环境等关键参数进行实时监测、预警和记录,确保机房设备的稳定运行和数据安全。 机房动力环境监控系统为企业
    的头像 发表于 07-15 16:37 243次阅读

    机房动环监控系统组成设备

    在信息化高速发展的今天,机房作为信息存储、处理与传输的重要场所,其安全稳定运行显得尤为重要。机房动环监控系统作为确保机房环境及设备稳定运行的关键技术之一,通过在线监测与远程运维,实现对机房
    的头像 发表于 06-20 16:42 375次阅读

    机房ups电源含义,选择合适的类型

    在现代社会中,信息技术的快速发展使得机房成为了许多组织和企业的核心设施。而机房的正常运行离不开稳定可靠的电力供应。为了应对电力故障和突发事件,机房UPS(不间断电源)应运而生。那么,什么是机房
    的头像 发表于 04-12 14:30 717次阅读
    <b class='flag-5'>机房</b>ups电源含义,选择合适的类型

    机房巡检机器人功能和作用

    现在信息量愈发增大,机房中心所管理的信息数据亦随之增多,且规模渐趋庞大复杂。为确保机房数据安全,需每日派遣人员进行多次人工巡检,而运维巡检存在高密度、重复性及熬夜巡检。巡检数据的及时性和准确性依旧
    的头像 发表于 04-02 08:46 669次阅读
    <b class='flag-5'>机房</b>巡检机器人功能和作用

    数据中心机房建设的痛点:机房顶面防水的重要性

    随着数字时代的来临,数据中心机房承载着大量的精密设备,而其中一个常被忽视的痛点就是机房顶面防水的问题。本文将深入探讨机房顶面防水的重要性,以及在机房选址和总体设计中的关键考虑因素。
    的头像 发表于 02-02 10:23 807次阅读

    动环监控系统应用智慧机房

    数据机房作为企业信息化的核心,有大量重要的数据和资料需要管理。为了机房运行的稳定,通过建设智慧机房以实现可视化监测、环境与设备巡查、测温、人员管理等功能。 所谓的智慧机房,是指具有自动
    的头像 发表于 12-01 17:49 703次阅读