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

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

1. 需求:

作为一家专注于三维高精度地图服务的公司,有海量(PB级)的原始数据、中间数据、成功数据,需要存储、管理、并定期归档。

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

2. 分析:

  1. 系统重点1 n Y z Z 2 o在于数据存储的选型,支持海量数据的存储,能够支持在复a q ! f S杂网络下的数据上传。选用CEPH作为] 1 + X = 1 ! ~数据存储,RGW对象存储,S3协议上传下载,完美支持分片和断点续传。
  2. 系统难点在于文件f / 5 * 7 k级别的业务权限控制,以及文件(夹)更多的属性支持。CEPH RGW本身支持权限控制,但是无法和业务权限做对接。对象存储本身没有文件夹的概念,x w w 7 T无法对文件夹做分类、数量展示、大小展示。所以实现自定义索引服务,CEPH主要负责存储,自定义索引服务实现展示# [ S k f C与查询。

3. 实现

系统重点在于海量数据上传的可靠性与海量数据索引的管理。

3.1 架构

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

  1. 上传助手就是类百度网盘I [ q G r K 0 %的桌面端软件,采用Electron JS
    )实现。主要实现功能:项目展示、上传、下载。
  2. 业务层包- y 1 c括网关服务、账号服务、项目服务、文件索引服务等。采用Java + Spring Boot + Spring Cloud技术栈。其中重点服务是文件索引服务Index Server,负] t 8 f y / a T责海量文件的索引维护和查询。
  3. 业务数据MySQL集群+Redis集群,海量文件存储使用CEPH对象存储,支持S3 API。

3.3 关键流程图

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

  1. 上传助手使用[ & B 5 -普通的Put Oj o $ 0 F rbjec3 W l E g Ct请求上传文件,加上自定义的metadata字段(项目ID、用户ID等)即可完成数据的提交。
  2. Openresty使; n { H 9 D H / F用proxy模式将文件请求转发到 C5 ; ( {EPH RGW,由RGW完成后台数据存储处理。
  3. Openresty在RGW完成数据存储以后A v [ X D & g + *,调用log_by_lua_file将对应请求的用户自定义metadata和文件属性转发到后台Kafka。
  4. 文件索引服o l x F h ; ;务(Index Server)从Kafka中消费任务,拿z 8 Z 2 s x Q U到每个文件的信息。
  5. 文件索引服务(Indei P c ~x Server). = ?对文件数据按业务要求进行处理后,存入MySQL数据库。

3.4 示例代码

log_by_lua_file.lua:从Openresty获取文件信息,并发往Kafka

localt u w U @  e P cjson = require "cjson"
local producer = require "resty.kafka.producer"
local broker_list = {
{ host = "172.16.0.20", port = 9092 },
}
functioC R 2 m 1 ln send_job_to_kafka()
local log_json = {}
local reX k f { z 7 U @ Aq_headers_ = ngx.req.get_headers()
for k, v in pai: N w M K W 9rs(req_headers_) do
ifG * + : w 0 P 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_m & Z f # x) do
if k == "etag" then
log_json["etag"] = st( @ 5ring.gsub(v, """, "")
break
end
end
log_json["uri"] = ngx.var.urP M c * fi
log_json["host"] = ngx.var.host
log_json["remoteAddr"] = ngx.var.M = G D = 8remote_addr
loJ ! c ? ]g_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} c y d ! V 6 & 7.var.status
-- 过滤Put Object成功的请求,记录相应的metadata及请求ID,并转{ K w A - / j *发到kafka
if request_method == "PUT" and stb } L xatus_code == "200" then
local bp = producer:new(broker_list, { producer_ty] . p ^ 9 R w ) 0pe = "async" })
local ok, err =s M  B 1 ? O T bpw [ {:send("cU r Zeph_lua_test", nil, send_jo& C c W 6b_to_kafka())
if not okq - # _ I thb @ yen
ngx.log(ngx.ERR, "kafka send err:", err)
return
end
ngx.log(ngx.ERR, 0 u Q S v S T 0 3"kafka send success:", ok)
end

4. 总结

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