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

上篇文章我們講到對(duì)賬系統(tǒng)收集數(shù)據(jù)的流程，下面我們?cè)賮碇v下數(shù)據(jù)核對(duì)的流程。

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

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

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

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

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

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

DP 核對(duì)

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

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

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

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

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

----?A
CREATE?TABLE?IF?NOT?EXISTS?dp.pay_check_success?(
????`batch_no`?bigint?comment?'批次號(hào)',
????`merchant_no`?string?comment?'三方商戶號(hào)',
????`sub_merchant_no`?string?comment?'三方子商戶號(hào)',
????`biz_id`?string?comment?'對(duì)賬業(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)識(shí)',
????`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 分為兩部分，第一部分將會(huì)在 DP 中創(chuàng)建一張 pay_check_success，記錄核對(duì)成功的數(shù)據(jù)。

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

查找核對(duì)成功的數(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ái)系統(tǒng)變量。DP_1_DAYS_AGO_Ymd 代表當(dāng)前日期的前一天

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

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

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

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

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

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

CREATE?TABLE?IF?NOT?EXISTS?dp.check_dp_buffer_record?(
????`biz_id`?string?comment?'訂單號(hào)',
????`order_type`?string?comment?'訂單類型?0本端訂單?1渠道訂單',
????`bill_date`?int?comment?'對(duì)賬日期',
????`biz_type`?int?comment?'業(yè)務(wù)類型',
????`channel_code`?int?comment?'渠道類型',
????`amount`?string?comment?'金額',
????`merchant_no`?string?comment?'商戶號(hào)',
????`sub_merchant_no`?string?comment?'三方子商戶號(hào)',
????`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 分為兩部分，第一部分將會(huì)在 DP 中創(chuàng)建一張 check_dp_buffer_record，記錄核對(duì)差異的的數(shù)據(jù)。

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

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

• 本端單邊賬，即本端存在數(shù)據(jù)，但是對(duì)端不存在數(shù)據(jù)
• 渠道端單邊賬，即對(duì)端存在數(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í)際邏輯可以簡(jiǎn)化為下面 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ù)，如果渠道單號(hào)不存在，則認(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ù)，如果本端單號(hào)不存在，則為渠道數(shù)據(jù)存在，本端數(shù)據(jù)不存在。當(dāng)然也有可能是兩端數(shù)據(jù)都存在，但是金額不相等。

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

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

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

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

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

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

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

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

DP 核對(duì)過程是無法區(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ù)處理流程將會(huì)查找 check_dp_buffer_record 表所有未核對(duì)的記錄，然后依次遍歷。

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

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

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

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

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

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

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

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

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

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

這里邏輯簡(jiǎn)單解釋為，就是利用 Scala 腳本代碼對(duì)數(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)前存在對(duì)賬系統(tǒng)內(nèi)部差錯(cuò)數(shù)據(jù)表中。

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

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

總結(jié)

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

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

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

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

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