长春市教育局大数据中心需要准备的技术问题

一、ElasticSearch的备份与还原

============================================================================================================================================================
1. 问题

Elasticsearch 副本提供了高可靠性；它们让你可以容忍零星的节点丢失而不会中断服务。但是，副本并不提供对灾难性故障的保护。对这种情况，你需要的是对集群真正的备份——在某些东西确实出问题的时候有一个完整的拷贝。

2.解决方案

通过快照的方式，将Elasticsearch集群中的数据，备份到HDFS上，这样数据即存在于Elasticsearch(简称ES)集群当中，又存在于HDFS上。当ES集群出现不可恢复性的故障时，可以将数据从HDFS上快速恢复。

也可以选择NFS方式进行备份数据：https://www.cnblogs.com/keithtt/p/7189493.html


3、需要进行的测试
（1）如何进行快照备份
（2）如何进行快照还原
（3）在历史日志时间太久，还不想彻底删除掉，想要归档，可以使用快照方式吗。如果以后还想用，怎么还原回来？

https://www.jianshu.com/p/d2a189f704a2

可以采用rename_pattern和rename_replacement用于重命名index，因为无法将index还原到open的index。

如何删除N个月前的数据？

https://www.cnblogs.com/kasumi/p/6479733.html

https://www.cnblogs.com/zhzhang/p/6946264.html

1、已知：可以全库备份，也可以指定index备份。

2、能不能备份指定条件的呢？不知道...意义不大，研究不研究无所谓。

3、如果按这个方案，那么HDFS和ES的服务器在物理上要如何规划？比如3台物理服务器（什么样的配置CPU、内存、磁盘??）

组成VMWARE ESXI虚拟机集群，2台NTB的存储，HDFS和ES应该如何避免放在一个篮子里？这里需要规划和绘制示意图，要求教育局购买。 存储的容量如何规划？？？比如100T ,教育局会问，能用多久？以后扩容怎么办？

是不是需要单独购买硬件防火墙？？DDOS攻击怎么办？？

============================================================================================================================================================
二、整体的技术架构图

三、各业务系统数据汇集的办法

 

国内同类型软件的实现方法：

http://www.ruisoft.com/solution-smartcity-01.html

http://www.chinawiserv.com/home/news/detail/id/526

各业务系统需要提供全量和增量数据上报，增量数据上报肯定会要求上报最近新增、修改、删除的数据，最简单的办法是使用触发器，下面给出测试过的各主流数据库触发器方案：

https://www.cnblogs.com/littlehb/p/5614780.html

特殊需要说明的是：对于物理性删除的delete情况，我们需要规划一个deleted表来记录删除的主键ID。

（1） SqlServer

如果业务库是SqlServer的话，可以参考这篇文章：

http://www.cnblogs.com/iampkm/p/4082916.html  (时间戳方案)

https://blog.csdn.net/yenange/article/details/49636215 （CDC方案）

（2） Mongodb

如果业务库是Mongodb的话，可以参考这篇文章：

https://blog.csdn.net/u013066244/article/details/80004153

（3）Mysql

如果业务库是Mysql的话，还需要特殊注意下这个：

如果Mysql数据库版本低于5.6，可以考虑使用触发器方式：https://blog.csdn.net/heweimingming/article/details/51315895

（4） Oracle方案

https://blog.csdn.net/strawberry1019/article/details/79422071

伪列的方案不是很优雅，不建议使用，因为alter table无效。

四、数据汇集有两种方式：

（1）业务系统可能采取提供webservice ,http restful等形式，要求东师理想自己进行数据采集。 这个需要一个个接入。（推荐使用这种！！！）

          缺点：需要单独编码实现，不编码无法完成。

          优点：东师接入可以通过程序控制数据质量，及时发现问题。

（2） 业务系统通过东师理想大数据中心的API接口自行编码进行数据上报。

         优点：没有优势。

         缺点：需要单独编码实现，不编码无法完成。

（3）业务系统不提供webservice,https restful等形式，主动进行数据上报。（前置机）

         优点：东师理想提供前置机代码，通过简单配置即可完成上报工作，业务系统无开发量，简单。

         缺点：需要将前置机部署到业务系统中，业务系统会比较反感。

一、全量和增量

其实全量和增量我们可以看做从哪个时间戳开始的问题，从遥远的古代开始，就是从头开始。从昨天晚上18点开始，就可以理解为增量。其实真正的意思是一样的。这个开始的位置，是由服务器维护的，业务系统在上报前需要申请开始位置。

二、定时和实时

这类的大数据汇集中心，与东师理想的大数据中心不同。东师理想的大数据中心，采用的是准实时的mysql+canal数据上报方式，延迟时间一般在1秒至3秒间（指正常工作情况下，不出现故障的情况下）。而长春市教育局的大数据中心，个人理解业务系统的数据汇集可以是定时的，定时定长最小为10分钟，这已经足够让其完美的运行，现在流行的ETL工具等，都是定期调度进行数据变更汇集的思路，这无可厚非。

上面这些东西，都在一个主题：业务数据的增（改）、删。

我们的设计思路是，不管是增加，还是修改，这条数据都要重新上报覆盖掉旧的即可，所以，无所谓增加还是修改。对于删除，如果是伪删除的话，即标识b_deleted=1这样的删除，其实就是修改，不是真删除，不必讨论。如果是真删除，我们要求必须提供在指定时间戳后，获取删除掉了哪些主键，然后我们发送到大数据中心，对数据进行修改。

五、python前置机的功能组成

两个线程

（1） 第一个线程是：正常的增量数据上报，数据扫描时间设置为10分钟。

（2） 第二个线程是：心跳，向数据中心报告其存在，1分钟一次心跳。每次心跳时，向大数据中心获取是否有任务让其执行，比如：将某一个记录上报，从某个时间点开始重新上报（这个可以是增量，也可以是全量，看起始时间点）。

这样做的目的是架构最简单，最清晰，最不易出错。

 

业务系统提供一个只读帐号给python程序，python程序以源码形式提供给各业务系统，python程序不会把业务系统不允许的数据内容进行采集。因python程序也可能需要不断的完善，所有应该是有两个py文件，一个是update.py ，另一个是senddata,py. update.py负责向oss检查是否本地的senddata.py与oss上的同名文件是否md5值一致，不一致则下载回来，保证senddata.py 的自动更新。

六、核心 代码

1、全量获取es中的数据

#!user/bin/env python3
# -*- coding: gbk -*-

# https://es.xiaoleilu.com/060_Distributed_Search/20_Scan_and_scroll.html

from elasticsearch import Elasticsearch
from elasticsearch import helpers

es_servers = [{
    'host': 'data.edusoa.com',
    'port': 9200
}]
es = Elasticsearch(hosts=es_servers)
index_v="t_wkds_info"
doc_type_v="doc"

# 方式1：ES的API中提供了scan和scroll，这个方法有点类型传统数据库中的游标。
# query={"query" : {"match_all" : {}}}
# scanResp = helpers.scan(es, query, scroll="10m", index=index_v, doc_type=doc_type_v, timeout="10m")
#
# for resp in scanResp:
#     print(len(scanResp))
#     print(resp)

# 方式2：按地区Area_Code='aly'为例,以PK进行遍历
query={
    "query": {
        "bool": {
            "must": [
                {
                    "term": {
                        "area_code": "aly"
                    }
                },
                {
                    "range": {
                        "pk": {
                            "gt": 0
                        }
                    }
                }
            ]
        }
    },
    "size": 500
}

# 起始值

startId=0
allCount=0

while startId>-1:
    query['query']['bool']['must'][1]['range']['pk']['gt']=startId
    _searched = es.search(index=index_v, doc_type=doc_type_v, body=query)
    # 输出查询到的结果
    count=0
    for hit in _searched['hits']['hits']:
        print(hit['_source'], flush=True)
        count=count+1
        allCount=allCount+1
    if count>0:
        startId=_searched['hits']['hits'][count-1]['_source']['pk']
        print("allCount="+str(allCount))
    else:
        startId=-1

print("成功结束！")

2、操作Kafka的代码

https://www.cnblogs.com/littlehb/p/8438401.html

Python3.6 连接Mssql

 https://www.cnblogs.com/xuhongfei/p/9145233.html

 https://www.lfd.uci.edu/~gohlke/pythonlibs/#pymssql

下载：pymssql-2.1.4.dev5-cp36-cp36m-win_amd64.whl

Python3.6连接oracle数据库

https://www.cnblogs.com/jsonhc/p/7283620.html

下载：cx_Oracle-6.3.1-cp36-cp36m-win_amd64.whl

https://pypi.org/project/cx_Oracle/

https://blog.csdn.net/mhmds/article/details/53079322

Oracle 11g Client x64

https://download.oracle.com/otn/nt/oracle11g/112010/win64_11gR2_client.zip

Python 3.6 MongoDb

https://www.cnblogs.com/JansXin/p/7885767.html

https://blog.csdn.net/ialexanderi/article/details/77744190

https://blog.csdn.net/u013066244/article/details/80004153