基于Openresty+CEPH实现海量数据管理系统

「持续更新中，欢迎关注...」

1. 需求：

做为一家专一于三维高精度地图服务的公司，内部有海量（PB级）的原始数据、中间数据、成功数据，须要存储、管理、并按期归档。

1. 按项目管理数据，数据分类航飞数据、控制点数据、中间数据、成果数据、其余数据。数据来源包括无人机数据、载荷数据、地面站数据、人工打点数据等。不一样渠道聚集而来的数据。
2. 采用相似百度网盘的形式，上传、下载，支持断点续传、进度跟踪。
3. 支持细化到文件级别的权限控制，以及更多的文件(夹)属性。

2. 分析：

1. 系统重点在于数据存储的选型，支持海量数据的存储，可以支持在复杂网络下的数据上传。选用CEPH做为数据存储，RGW对象存储，S3协议上传下载，完美支持分片和断点续传。
2. 系统难点在于文件级别的业务权限控制，以及文件（夹）更多的属性支持。CEPH RGW自己支持权限控制，可是没法和业务权限作对接。对象存储自己没有文件夹的概念，没法对文件夹作分类、数量展现、大小展现。因此实现自定义索引服务，CEPH主要负责存储，自定义索引服务实现展现与查询。
3. 因为上传下载会通过Openresty，经过lua脚本，将流经的文件信息，经由kafka转发到业务服务中进行业务处理应用。

3. 实现

3.1 架构

1. 上传助手就是类百度网盘的桌面端软件，采用Electron JS

)实现。主要实现功能：项目展现、上传、下载。

1. 业务层包括网关服务、帐号服务、项目服务、文件索引服务等。采用Java + Spring Boot + Spring Cloud技术栈。其中重点服务是文件索引服务Index Server，负责海量文件的索引维护和查询。
2. 业务数据MySQL集群+Redis集群，海量文件存储使用CEPH对象存储，支持S3 API。

3.3 关键流程图

1. 上传助手使用普通的Put Object请求上传文件，加上自定义的metadata字段（项目ID、用户ID等）便可完成数据的提交。
2. Openresty使用proxy模式将文件请求转发到 CEPH RGW，由RGW完成后台数据存储处理。
3. Openresty在RGW完成数据存储之后，调用log_by_lua_file将对应请求的用户自定义metadata和文件属性转发到后台Kafka。
4. 文件索引服务（Index Server）从Kafka中消费任务，拿到每一个文件的信息。
5. 文件索引服务（Index Server）对文件数据按业务要求进行处理后，存入MySQL数据库。

3.4 示例代码

log_by_lua_file.lua：从Openresty获取文件信息，并发往Kafka

local cjson = require "cjson"
local producer = require "resty.kafka.producer"
local broker_list = {
    { host = "172.16.0.20", port = 9092 },
}
function send_job_to_kafka()
    local log_json = {}
    local req_headers_ = ngx.req.get_headers()

    for k, v in pairs(req_headers_) do
        if k == "content-length" then
            log_json["contentLength"] = tostring(v)
        end
        if k == "u-id" then
            log_json["uId"] = tostring(v)
        end
        if k == "p-id" then
            log_json["pId"] = tostring(v)
        end
    end

    local resp_headers_ = ngx.resp.get_headers()
    for k, v in pairs(resp_headers_) do
        if k == "etag" then
            log_json["etag"] = string.gsub(v, "\"", "")
            break
        end
    end

    log_json["uri"] = ngx.var.uri
    log_json["host"] = ngx.var.host
    log_json["remoteAddr"] = ngx.var.remote_addr
    log_json["status"] = ngx.var.status
    local message = cjson.encode(log_json);
    ngx.log(ngx.ERR, "message is[", message, "]")
    return message
end

--local is_args = ngx.var.is_args
local request_method = ngx.var.request_method
local status_code = ngx.var.status

-- 过滤Put Object成功的请求，记录相应的metadata及请求ID，并转发到kafka
if request_method == "PUT" and status_code == "200" then
    local bp = producer:new(broker_list, { producer_type = "async" })
    local ok, err = bp:send("ceph_lua_test", nil, send_job_to_kafka())
    if not ok then
        ngx.log(ngx.ERR, "kafka send err:", err)
        return
    end
    ngx.log(ngx.ERR, "kafka send success:", ok)
end

4. 总结

1. 经过此架构方案，在海量文件归档过程当中，将文件基本信息异步导入到业务数据库中，便于业务应用开发。
2. 此架构通常也应用对象存储的多媒体文件处理，好比图片处理、视频处理、加水印、鉴黄、事件通知等。