1. 需求:
作為一家專注於三維高精度地圖服務的公司,內部有海量(PB級)的原始數據、中間數據、成功數據,需要存儲、管理、並定期歸檔。
按項目管理數據,數據分類航飛數據、控制點數據、中間數據、成果數據、其他數據。數據來源包括無人機數據、載荷數據、地面站數據、人工打點數據等。不同渠道彙集而來的數據。
採用類似百度網盤的形式,上傳、下載,支持斷點續傳、進度跟蹤。
支持細化到文件級別的權限控制,以及更多的文件(夾)屬性。
2. 分析:
系統重點在於數據存儲的選型,支持海量數據的存儲,能夠支持在複雜網絡下的數據上傳。選用CEPH作為數據存儲,RGW對象存儲,S3協議上傳下載,完美支持分片和斷點續傳。
系統難點在於文件級別的業務權限控制,以及文件(夾)更多的屬性支持。CEPH RGW本身支持權限控制,但是無法和業務權限做對接。對象存儲本身沒有文件夾的概念,無法對文件夾做分類、數量展示、大小展示。所以實現自定義索引服務,CEPH主要負責存儲,自定義索引服務實現展示與查詢。
由於上傳下載會經過Openresty,通過lua腳本,將流經的文件信息,經由kafka轉發到業務服務中進行業務處理應用。
上傳助手就是類百度網盤的桌面端軟件,採用
)實現。主要實現功能:項目展示、上傳、下載。
業務層包括網關服務、賬號服務、項目服務、文件索引服務等。採用Java Spring Boot Spring Cloud技術棧。其中重點服務是文件索引服務Index Server,負責海量文件的索引維護和查詢。
業務數據MySQL集羣 Redis集羣,海量文件存儲使用CEPH對象存儲,支持S3 API。
上傳助手使用普通的Put Object請求上傳文件,加上自定義的metadata字段(項目ID、用户ID等)即可完成數據的提交。
Openresty使用proxy模式將文件請求轉發到 CEPH RGW,由RGW完成後台數據存儲處理。
Openresty在RGW完成數據存儲以後,調用log_by_lua_file將對應請求的用户自定義metadata和文件屬性轉發到後台Kafka。
文件索引服務(Index Server)從Kafka中消費任務,拿到每個文件的信息。
文件索引服務(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 messageend--local is_args = ngx.var.is_argslocal request_method = ngx.var.request_methodlocal status_code = ngx.var.status-- 過濾Put Object成功的請求,記錄相應的metadata及請求ID,並轉發到kafkaif 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. 總結
通過此架構方案,在海量文件歸檔過程中,將文件基本信息異步導入到業務數據庫中,便於業務應用開發。
此架構一般也應用對象存儲的多媒體文件處理,比如圖片處理、視頻處理、加水印、鑑黃、事件通知等。