千萬級支付對賬系統(tǒng)怎么玩(下篇)?

上篇文章我們講到對賬系統(tǒng)收集數(shù)據(jù)的流程，下面我們再來講下數(shù)據(jù)核對的流程。

這里再放一下支付對賬系統(tǒng)整個(gè)流程，忘記的同學(xué)可以看這個(gè)圖片再回憶一下。

數(shù)據(jù)導(dǎo)入DP

在 DP 核對之前，我們需要將對賬系統(tǒng)收集的數(shù)據(jù)，從 ?MySQL 導(dǎo)入 DP Hive 表中。

DP 任務(wù)調(diào)度開始，DP 平臺定時(shí)檢測對賬系統(tǒng)提供 HTTP 接口，判斷本次存疑流程是否處理完成。

如果完成，自動觸發(fā)將數(shù)據(jù)從 MYSQL 導(dǎo)入 DP Hive 表中。

數(shù)據(jù)導(dǎo)入之后，將會開始 DP 核對規(guī)程。這個(gè)過程就是整個(gè)對賬流程最關(guān)鍵的部分，這個(gè)流程核對兩端數(shù)據(jù)，檢查兩端是否存在差異數(shù)據(jù)。

DP 核對

數(shù)據(jù)導(dǎo)入結(jié)束，DP 平臺開始核對數(shù)據(jù)，這個(gè)過程分為兩個(gè)核對任務(wù)：

• 成功數(shù)據(jù)核對
• 存疑數(shù)據(jù)核對

成功數(shù)據(jù)核對

成功數(shù)據(jù)核對任務(wù)，核對的目的是為了核對出本端與對端支付單號與金額一致的數(shù)據(jù)。

這里的核對任務(wù)使用了 Hive SQL ，整個(gè) SQL 如下所示：

----?A
CREATE?TABLE?IF?NOT?EXISTS?dp.pay_check_success?(
????`batch_no`?bigint?comment?'批次號',
????`merchant_no`?string?comment?'三方商戶號',
????`sub_merchant_no`?string?comment?'三方子商戶號',
????`biz_id`?string?comment?'對賬業(yè)務(wù)關(guān)聯(lián)字段',
????`biz_amount`?bigint?comment?'金額',
????`biz_date`?string?comment?'業(yè)務(wù)日期',
????`biz_type`?int?comment?'業(yè)務(wù)類型',
????`status`?int?comment?'狀態(tài)標(biāo)識',
????`remark`?string?comment?'備注',
????`create_time`?string?comment?'創(chuàng)建時(shí)間',
????`update_time`?string?comment?'修改時(shí)間',
????`trade_date`?int?comment?'訂單交易日期',
????`channel_code`?int?comment?'渠道類型'
);

----B
insert
????overwrite?table?dp.pay_check_success
select
????tb1.batch_no?as?batch_no,
????tb1.merchant_no?as?merchant_no,
????tb1.sub_merchant_no?as?sub_merchant_no,
????tb1.biz_id?as?biz_id,
????tb1.biz_amount?as?biz_amount,
????tb1.biz_date?as?biz_date,
????tb1.biz_type?as?biz_type,
????tb1.status?as?status,
????tb1.remark?as?remark,
????tb1.trade_date?as?trade_date,
????tb1.channel_code?as?channel_code
from
????(
????????select
????????????tb2.batch_no?as?batch_no,
????????????tb1.merchant_no?as?merchant_no,
????????????tb1.sub_merchant_no?as?sub_merchant_no,
????????????tb1.biz_order_no?as?biz_id,
????????????tb1.trader_amount?as?biz_amount,
????????????'${DP_1_DAYS_AGO_Ymd}'?as?biz_date,
????????????'0'?as?status,
????????????''?as?remark,
????????????'${DP_1_DAYS_AGO_Ymd}'?as?trade_date,
????????????tb1.channel_code?as?channel_code
????????from
????????????dp.pay_check_record?tb1
????????????inner?join?dp.pay_check_channel_record?tb2?on?tb1.biz_order_no?=?tb2.biz_order_no
????????????and?tb1.trader_amount?=?tb2.trader_amount
????????????and?tb1.channel_code?=?tb2.channel_code
????????where
????????????tb1.is_check?=?0
????????????and?tb2.is_check?=?0
????????????and?tb1.bill_date?=?'${DP_1_DAYS_AGO_Ymd}'
????????????and?tb2.bill_date?=?'${DP_1_DAYS_AGO_Ymd}'
????????????and?tb1.is_filter?=?0
????)?tb1

整個(gè) SQL 分為兩部分，第一部分將會在 DP 中創(chuàng)建一張 pay_check_success，記錄核對成功的數(shù)據(jù)。

第二部分，將核對成功的數(shù)據(jù)插入上面創(chuàng)建的 pay_check_success 表中。

查找核對成功的數(shù)據(jù) SQL 如下：

????????select
????????????tb2.batch_no?as?batch_no,
????????????tb1.merchant_no?as?merchant_no,
????????????tb1.sub_merchant_no?as?sub_merchant_no,
????????????tb1.biz_order_no?as?biz_id,
????????????tb1.trader_amount?as?biz_amount,
????????????'${DP_1_DAYS_AGO_Ymd}'?as?biz_date,
????????????'0'?as?status,
????????????''?as?remark,
????????????'${DP_1_DAYS_AGO_Ymd}'?as?trade_date,
????????????tb1.channel_code?as?channel_code
????????from
????????????dp.pay_check_record?tb1
????????????inner?join?dp.pay_check_channel_record?tb2?on?tb1.biz_order_no?=?tb2.biz_order_no
????????????and?tb1.trader_amount?=?tb2.trader_amount
????????????and?tb1.channel_code?=?tb2.channel_code
????????where
????????????tb1.is_check?=?0
????????????and?tb2.is_check?=?0
????????????and?tb1.bill_date?=?'${DP_1_DAYS_AGO_Ymd}'
????????????and?tb2.bill_date?=?'${DP_1_DAYS_AGO_Ymd}'
????????????and?tb1.is_filter?=?0

上述 SQL 存在一些 DP 平臺系統(tǒng)變量。DP_1_DAYS_AGO_Ymd 代表當(dāng)前日期的前一天

主要邏輯非常簡單，利用 sql 內(nèi)連接查詢的功能，可以查找單號,金額,渠道編碼一致的數(shù)據(jù)。

成功數(shù)據(jù)核對任務(wù)結(jié)束，將會把剛才在 DP 中創(chuàng)建的 ?pay_check_success 同步回對賬系統(tǒng)的 MYSQL 數(shù)據(jù)庫中。

存疑數(shù)據(jù)核對

存疑數(shù)據(jù)核對任務(wù)，核對的目的是為了核對出本端與對端支付單號或金額不一致的數(shù)據(jù)。

這些數(shù)據(jù)將會當(dāng)做存疑數(shù)據(jù)，這些數(shù)據(jù)將會在第二階段存疑數(shù)據(jù)處理。

這里的核對任務(wù)也是使用了 Hive SQL ，整個(gè) SQL 跟上面比較類似，SQL 如下所示：

CREATE?TABLE?IF?NOT?EXISTS?dp.check_dp_buffer_record?(
????`biz_id`?string?comment?'訂單號',
????`order_type`?string?comment?'訂單類型?0本端訂單?1渠道訂單',
????`bill_date`?int?comment?'對賬日期',
????`biz_type`?int?comment?'業(yè)務(wù)類型',
????`channel_code`?int?comment?'渠道類型',
????`amount`?string?comment?'金額',
????`merchant_no`?string?comment?'商戶號',
????`sub_merchant_no`?string?comment?'三方子商戶號',
????`trade_date`?int?comment?'交易日期',
????`create_time`?string?comment?'創(chuàng)建時(shí)間',
????`update_time`?string?comment?'修改時(shí)間'
);

insert
????overwrite?table?dp.check_dp_buffer_record
select
????tb1.biz_id?as?biz_id,
????tb1.order_type?as?order_type,
????tb1.bill_date?as?bill_date,
????tb1.biz_type?as?biz_type,
????tb1.channel_code?as?channel_code,
????tb1.amount?as?amount,
????tb1.merchant_no?as?merchant_no,
????tb1.sub_merchant_no?as?sub_merchant_no,
????tb1.trade_date?as?trade_date,
????'${DP_0_DAYS_AGO_Y_m_d_HMS}',
????'${DP_0_DAYS_AGO_Y_m_d_HMS}'
FROM
????(
????????select
????????????tb1.biz_order_no?as?biz_id,
????????????0?as?order_type,
????????????tb1.bill_date?as?bill_date,
????????????10?as?biz_type,
????????????tb1.channel_code?as?channel_code,
????????????tb1.trade_amount?as?amount,
????????????tb1.merchant_no?as?merchant_no,
????????????tb1.sub_merchant_no?as?sub_merchant_no,
????????????'${DP_1_DAYS_AGO_Ymd}'?as?trade_date
????????FROM
????????????(
????????????????select
????????????????????biz_order_no,
????????????????????bill_date,
????????????????????channel_code,
????????????????????trade_amount,
????????????????????merchant_no,
????????????????????sub_merchant_no
????????????????from
????????????????????ods.pay_check_record
????????????????where
????????????????????and?bill_date?=?'${DP_1_DAYS_AGO_Ymd}'
????????????????????and?is_filter?=?0
????????????????????and?is_check?=?0
????????????)?tb1
????????????LEFT?JOIN?(
????????????????select
????????????????????biz_order_no,
????????????????????trade_amount,
????????????????????channel_code
????????????????from
????????????????????ods.pay_check_channel_record
????????????????where
????????????????????and?bill_date?=?'${DP_1_DAYS_AGO_Ymd}'
????????????????????and?is_check?=?0
????????????)?tb2?ON?tb1.biz_order_no?=?tb2.biz_order_no
????????????and?tb1.trade_amount?=?tb2.trade_amount
????????????and?tb1.channel_code?=?tb2.channel_code
????????where
????????????tb2.biz_order_no?IS?NULL
????????union
????????select
????????????tb1.biz_order_no?as?biz_id,
????????????1?as?order_type,
????????????tb1.bill_date?as?bill_date,
????????????10?as?biz_type,
????????????tb1.channel_code?as?channel_code,
????????????tb1.trade_amount?as?amount,
????????????tb1.merchant_no?as?merchant_no,
????????????tb1.sub_merchant_no?as?sub_merchant_no,
????????????'${DP_1_DAYS_AGO_Ymd}'?as?trade_date
????????FROM
????????????(
????????????????select
????????????????????biz_order_no,
????????????????????bill_date,
????????????????????channel_code,
????????????????????trade_amount,
????????????????????merchant_no,
????????????????????sub_merchant_no
????????????????from
????????????????????ods.pay_check_chnnel_bill
????????????????where
????????????????????and?bill_date?=?'${DP_1_DAYS_AGO_Ymd}'
????????????????????and?is_check?=?0
????????????)?tb1
????????????LEFT?JOIN?(
????????????????select
????????????????????biz_order_no,
????????????????????channel_code,
????????????????????trade_amount
????????????????from
????????????????????ods.pay_check_record
????????????????where
????????????????????and?bill_date?=?'${DP_1_DAYS_AGO_Ymd}'
????????????????????and?is_filter?=?0
????????????????????and?is_check?=?0
????????????)?tb2?ON?tb1.biz_order_no?=?tb2.biz_order_no
????????????and?tb1.trade_amount?=?tb2.trade_amount
????????????and?tb1.channel_code?=?tb2.channel_code
????????where
????????????tb2.biz_order_no?IS?NULL
????)?tb1;

整個(gè) SQL 分為兩部分，第一部分將會在 DP 中創(chuàng)建一張 check_dp_buffer_record，記錄核對差異的的數(shù)據(jù)。

第二部分，將核對差異的數(shù)據(jù)插入上面創(chuàng)建的 check_dp_buffer_record 表中。

查找差異數(shù)據(jù)較為麻煩，需要分成兩部分收集：：

• 本端單邊賬，即本端存在數(shù)據(jù)，但是對端不存在數(shù)據(jù)
• 渠道端單邊賬，即對端存在數(shù)據(jù)，本端不存在數(shù)據(jù)

兩邊數(shù)據(jù)查找到之后，使用 ?SQL union 功能，將兩端數(shù)據(jù)聯(lián)合。

我們先來看下本端單邊張的邏輯的：

select
????tb1.biz_order_no?as?biz_id,
????0?as?order_type,
????tb1.bill_date?as?bill_date,
????10?as?biz_type,
????tb1.channel_code?as?channel_code,
????tb1.trade_amount?as?amount,
????tb1.merchant_no?as?merchant_no,
????tb1.sub_merchant_no?as?sub_merchant_no,
????'${DP_1_DAYS_AGO_Ymd}'?as?trade_date
FROM
????(
????????select
????????????biz_order_no,
????????????bill_date,
????????????channel_code,
????????????trade_amount,
????????????merchant_no,
????????????sub_merchant_no
????????from
????????????ods.pay_check_record
????????where
????????????and?bill_date?=?'${DP_1_DAYS_AGO_Ymd}'
????????????and?is_filter?=?0
????????????and?is_check?=?0
????)?tb1
????LEFT?JOIN?(
????????select
????????????biz_order_no,
????????????trade_amount,
????????????channel_code
????????from
????????????ods.pay_check_channel_record
????????where
????????????and?bill_date?=?'${DP_1_DAYS_AGO_Ymd}'
????????????and?is_check?=?0
????)?tb2?ON?tb1.biz_order_no?=?tb2.biz_order_no
????and?tb1.trade_amount?=?tb2.trade_amount
????and?tb1.channel_code?=?tb2.channel_code
where
????tb2.biz_order_no?IS?NULL

SQL 看起來比較復(fù)雜，實(shí)際邏輯可以簡化為下面 SQL :

select
????*
from
????innerTab?t1
????LEFT?JOIN?channelTab?t2?ON?t1.biz_order_no?=?t2.biz_order_no
????and?t1.trade_amount?=?t2.trade_amount
????and?t1.channel_code?=?t2.channel_code
where
????t2.biz_order_no?is?null;

這里主要利用 SQL 左連接的功能，本端數(shù)據(jù) left join 渠道數(shù)據(jù)，如果渠道單號不存在，則認(rèn)為本端數(shù)據(jù)存在，渠道數(shù)據(jù)不存在，當(dāng)然也有可能是兩端數(shù)據(jù)都存在，但是金額不相等。

這種情況記為本端數(shù)據(jù)存疑，orderType 為 0。

渠道端單邊賬收集邏輯：

select
????tb1.biz_order_no?as?biz_id,
????1?as?order_type,
????tb1.bill_date?as?bill_date,
????10?as?biz_type,
????tb1.channel_code?as?channel_code,
????tb1.trade_amount?as?amount,
????tb1.merchant_no?as?merchant_no,
????tb1.sub_merchant_no?as?sub_merchant_no,
????'${DP_1_DAYS_AGO_Ymd}'?as?trade_date
FROM
????(
????????select
????????????biz_order_no,
????????????bill_date,
????????????channel_code,
????????????trade_amount,
????????????merchant_no,
????????????sub_merchant_no
????????from
????????????ods.pay_check_chnnel_bill
????????where
????????????and?bill_date?=?'${DP_1_DAYS_AGO_Ymd}'
????????????and?is_check?=?0
????)?tb1
????LEFT?JOIN?(
????????select
????????????biz_order_no,
????????????channel_code,
????????????trade_amount
????????from
????????????ods.pay_check_record
????????where
????????????and?bill_date?=?'${DP_1_DAYS_AGO_Ymd}'
????????????and?is_filter?=?0
????????????and?is_check?=?0
????)?tb2?ON?tb1.biz_order_no?=?tb2.biz_order_no
????and?tb1.trade_amount?=?tb2.trade_amount
????and?tb1.channel_code?=?tb2.channel_code
where
????tb2.biz_order_no?IS?NULL

邏輯與本端單邊賬收集類似，渠道數(shù)據(jù) left join 本端數(shù)據(jù)，如果本端單號不存在，則為渠道數(shù)據(jù)存在，本端數(shù)據(jù)不存在。當(dāng)然也有可能是兩端數(shù)據(jù)都存在，但是金額不相等。

這里記為渠道存疑數(shù)據(jù)，orderType 為 1

成功數(shù)據(jù)核對以及存疑數(shù)據(jù)核對結(jié)束，DP 平臺將會自動把數(shù)據(jù)從 Hive 表中導(dǎo)入到 MYSQL。

數(shù)據(jù)導(dǎo)出結(jié)束，DP 平臺將會調(diào)用對賬系統(tǒng)的相關(guān)接口，通知對賬系統(tǒng) DP 核對流程結(jié)束。

DP 核對流程是整個(gè)對賬流程核心流程，目前千萬級數(shù)據(jù)的情況下，大概能在一個(gè)小時(shí)之內(nèi)搞定。

DP 核對流程結(jié)束之后，對賬系統(tǒng)開始下個(gè)流程-二次存疑數(shù)據(jù)處理

二次存疑數(shù)據(jù)處理

前面流程我們講到存疑處理，為什么這里還需要二次存疑數(shù)據(jù)處理呢？

這因?yàn)?DP 核對存疑數(shù)據(jù)收集的過程，我們使用業(yè)務(wù)單號與金額去互相匹配，那如果不存在，有可能是因?yàn)閮啥藬?shù)據(jù)有一端不存在，還有可能是因?yàn)閮啥藬?shù)據(jù)數(shù)據(jù)都存在，但是金額卻不相等。

DP 核對過程是無法區(qū)分出這兩種情況，所以增加一個(gè)二次存疑數(shù)據(jù)處理流程，單獨(dú)區(qū)分出這兩類數(shù)據(jù)。

回到二次存疑數(shù)據(jù)處理流程，當(dāng)天產(chǎn)生的所有存疑數(shù)據(jù)都從 DP 中導(dǎo)入到 check_dp_buffer_record 表。

二次存疑數(shù)據(jù)處理流程將會查找 check_dp_buffer_record 表所有未核對的記錄，然后依次遍歷。

遍歷過程中將會嘗試在 check_dp_buffer_record 表中查找相反方向的存疑數(shù)據(jù)。

這個(gè)可能不好理解，舉個(gè)例子：

假如有一筆訂單，本端是 100 元，渠道端是 10 元。這種情況兩筆記錄都會出現(xiàn)在 ?check_dp_buffer_record 表。

遍歷到本端這筆的時(shí)候，這筆類型是本端存疑，type為 0。使用者本端單號從check_dp_buffer_record 查找渠道端存疑（type 為 1）的數(shù)據(jù)。

上面的情況可以找到，證明這筆存疑數(shù)據(jù)其實(shí)是金額不相等，這里需要將數(shù)據(jù)移動到差錯(cuò)表。

那如果是正常一端缺失的數(shù)據(jù)，那自然去相反方向查找是找不到的，這種數(shù)據(jù)是正常存疑數(shù)據(jù)，移動內(nèi)部存疑表。

對賬系統(tǒng)二次存疑數(shù)據(jù)處理結(jié)束之后，開始下一個(gè)階段數(shù)據(jù)匯總。

數(shù)據(jù)匯總

數(shù)據(jù)匯總階段就是為了統(tǒng)計(jì)當(dāng)天每個(gè)有多少成功功對賬數(shù)據(jù)，多少存疑數(shù)據(jù)，統(tǒng)計(jì)結(jié)束通過看板給相關(guān)運(yùn)營人員展示統(tǒng)計(jì)數(shù)據(jù)。

由于數(shù)據(jù)量大的問題，這里使用的是 DP 平臺 Sprak 任務(wù)進(jìn)行任務(wù)統(tǒng)計(jì)。

這里邏輯簡單解釋為，就是利用 Scala 腳本代碼對數(shù)據(jù)進(jìn)行相關(guān)求和，這里代碼沒有普遍性，就不展示具體的邏輯了。

差錯(cuò)數(shù)據(jù)推送

數(shù)據(jù)匯總結(jié)束之后，開始下一個(gè)階段，差錯(cuò)數(shù)據(jù)推送給差錯(cuò)系統(tǒng)。

上面存疑數(shù)據(jù)處理的流程中轉(zhuǎn)化的差錯(cuò)數(shù)據(jù)，當(dāng)前存在對賬系統(tǒng)內(nèi)部差錯(cuò)數(shù)據(jù)表中。

目前我們差錯(cuò)數(shù)據(jù)是是另外一個(gè)差錯(cuò)系統(tǒng)單獨(dú)處理，所以對賬系統(tǒng)需要把差錯(cuò)數(shù)據(jù)表數(shù)據(jù)推送給差錯(cuò)系統(tǒng)。

這里的邏輯比較簡單，查找所有待處理的差錯(cuò)數(shù)據(jù)，遍歷發(fā)送 NSQ 消息給差錯(cuò)系統(tǒng)。

總結(jié)

千萬級數(shù)據(jù)對賬整個(gè)流程看起，其實(shí)相關(guān)操作流程都不是很難。

那我個(gè)人認(rèn)為這里難點(diǎn)在于第一需要一套完整大數(shù)據(jù)平臺體系，第二改變原有對賬方式，思考如何將對賬系統(tǒng)與大數(shù)據(jù)平臺一起串起來。

希望這篇文章對正好碰到該類問題同仁起到相關(guān)幫助。

預(yù)告一下，上面我們講的其實(shí)都是業(yè)務(wù)明細(xì)的對賬，這一部分其實(shí)在財(cái)務(wù)領(lǐng)域被叫做，賬賬對賬。

那實(shí)際上我們還需要核對當(dāng)天應(yīng)到資金與實(shí)到資金是否一致，下一篇文章我們聊聊這個(gè)。