實時數據表到歷史數據表的數據丟失

Question

顯示基於Web的監控項目的實時數據和歷史數據。有近16個採樣頻率爲50Hz的傳感器。傳感器的所有原始數據都必須存儲在數據庫中，每秒鐘可達到近900個數據。數據必須保存至少三年。數據庫是oracle 11g。

我的工作是爲傳感器硬件公司的工程師設計數據庫結構，他將編寫數據採集程序並將數據存儲到數據庫中。

設計了實時數據表和歷史數據表。從實時數據表中讀取實時數據，並從歷史數據表中讀取歷史數據。

實數據表如下，僅存儲一分鐘數據。

Create Table real_data(
record_time timestamp(3), 
ac_1 Float, 
ac_2 Float, 
ac_3 Float, 
ac_4 Float, 
ac_5 Float, 
ac_6 Float, 
ac_7 Float, 
ac_8 Float, 
ac_9 Float, 
ac_10 Float, 
ac_11 Float, 
ac_12 Float, 
ac_13 Float, 
ac_14 Float, 
ac_15 Float, 
ac_16 Float 
) 
Tablespace data_test; 

歷史數據表的結構是與真實數據，它由主鍵和分區

Create Table history_data(
record_time timestamp(3), 
ac_1 Float, 
ac_2 Float, 
ac_3 Float, 
ac_4 Float, 
ac_5 Float, 
ac_6 Float, 
ac_7 Float, 
ac_8 Float, 
ac_9 Float, 
ac_10 Float, 
ac_11 Float, 
ac_12 Float, 
ac_13 Float, 
ac_14 Float, 
ac_15 Float, 
ac_16 Float 
) 
Tablespace data_test 
PARTITION BY RANGE(record_time) 
INTERVAL(numtodsinterval(1，'day')) 
( 
    PARTITION P1 VALUES LESS THAN (TO_DATE('2016-08-01', 'YYYY-MM-DD')) 
); 

alter table history_data add constraint RECORD_DATE primary key (RECORD_TIME); 

間隔分區被選擇用於兩個原因相同：

1. sql查詢是基於web客戶端的時間記錄，如

   select ac_1來自ac_test where record_time> = to_timestamp（'2016-08-01 00:00:00'，'yyyy-mm-dd hh24：mi：ss'） and record_time < = to_timestamp（'2016-08-01 00 ：30：00'，'yyyy-mm-dd hh24：mi：ss'）;

2. 間隔分區的範圍是天。在一天數據測試期間，每天近430萬數據的成本爲近40秒。

執行作業以每一分鐘將實際數據傳送到歷史數據表。傳輸過程由oracle過程完成，傳輸時間由另一個表記錄：real_data_top_backup_date。

create or replace procedure copy_to_history_test is 
d_top_backup_date timestamp(3); 
begin 

select top_backup_date into d_top_backup_date from real_data_top_backup_date; 

Insert Into history_data Select * From real_data where record_time <d_top_backup_date; 

delete from real_data where record_time <d_top_backup_date; 

Update real_data_top_backup_date Set top_backup_date=(d_top_backup_date+1/24/60); 

commit; 

end copy_to_history_test; 

並編寫仿真程序來模擬傳感器數據採集和插入。

Declare 
time_index Number; 
start_time Timestamp(3); 
tmp_time Timestamp(3); 
tmp_value1 Float; 
tmp_value2 Float; 
tmp_value3 Float; 
tmp_value4 Float; 
tmp_value5 Float; 
tmp_value6 Float; 
tmp_value7 Float; 
tmp_value8 Float; 
tmp_value9 Float; 
tmp_value10 Float; 
tmp_value11 Float; 
tmp_value12 Float; 
tmp_value13 Float; 
tmp_value14 Float; 
tmp_value15 Float; 
tmp_value16 Float; 


Begin 

--initiaze the variable 
time_index:=0;  
SELECT to_timestamp('2016-08-01 00:00:00:000', 'yyyy-mm-dd h24:mi:ss:ff') Into start_time FROM DUAL; 

     While time_index<(50*60*60*24*7) 
     Loop 
     -- add 20 millionseconds 
     SELECT start_time+numtodsinterval((0.02*time_index),'SECOND') Into tmp_time FROM DUAL; 
     -- dbms_output.put_line(tmp_time); 
     -- create random number 
     select dbms_random.value Into tmp_value1 from dual ; 
     select dbms_random.value Into tmp_value2 from dual ; 
     select dbms_random.value Into tmp_value3 from dual ; 
     select dbms_random.value Into tmp_value4 from dual ; 
     select dbms_random.value Into tmp_value5 from dual ; 
     select dbms_random.value Into tmp_value6 from dual ; 
     select dbms_random.value Into tmp_value7 from dual ; 
     select dbms_random.value Into tmp_value8 from dual ; 
     select dbms_random.value Into tmp_value9 from dual ; 
     select dbms_random.value Into tmp_value10 from dual ; 
     select dbms_random.value Into tmp_value11 from dual ; 
     select dbms_random.value Into tmp_value12 from dual ; 
     select dbms_random.value Into tmp_value13 from dual ; 
     select dbms_random.value Into tmp_value14 from dual ; 
     select dbms_random.value Into tmp_value15 from dual ; 
     select dbms_random.value Into tmp_value16 from dual ; 
     --dbms_output.put_line(tmp_value); 

     -- Insert Into ac_data (sensor_id,data,record_time) Values(sensor_index,tmp_value,tmp_time); 
     Insert Into real_data Values(tmp_time,tmp_value1,tmp_value2,tmp_value3,tmp_value4,tmp_value5,tmp_value6,tmp_value7,tmp_value8,tmp_value9,tmp_value10,tmp_value11,tmp_value12,tmp_value13,tmp_value14,tmp_value15,tmp_value16); 
     if mod(time_index,50)=0 then 
     commit; 
     dbms_lock.sleep(1); 
     End If; 

     time_index:=time_index+1; 
     End Loop; 

-- dbms_output.put_line(c); 
    Exception 
    WHEN OTHERS THEN 
    log_write('insert data failure!'); 
End; 

問題是，在傳輸數據過程中，接近0.1％的傳感器數據量將會丟失。我認爲傳輸數據（插入數據和刪除數據）的並行操作會導致數據丟失。如何處理這個問題？

在這種情況下，數據庫結構是否可行？數據庫有更好的設計嗎？

Answer 1

「近0.1％的量的傳感器的數據將丟失」

很有可能的。默認情況下，Oracle在語句級別上使用Read Committed隔離模型。隔離級別意味着由其他會話添加到表中的記錄不會包含在您的過程插入的記錄集中。但是，如果其他會話已提交這些行，它們將在刪除語句的範圍內。這種現象被稱爲「幻影讀取」。

所以關鍵是將記錄插入到「實時」表中。在您的測試用具中，插入件分批承諾提交50個mod(time_index,50)=0。如果該提交發生在一個會話中，而copy_to_history_test()正在運行，那麼記錄可能會崩潰。也許你的流程在生產中會有所不同。

「如何處理問題？」

此問題的標準方法是使用SERIALIZABLE隔離級別。對於運行copy_to_history_test()的會話進行設置意味着所有的語句將在事務處理期間使用相同的數據狀態執行。提供的記錄只能插入到實時表中，這種方法不應該給你帶來任何傷害。 （如果其他進程可以更新或刪除這些記錄，那麼你有更大的架構問題。）

所以，你的程序，現在應該是這樣的：

create or replace procedure copy_to_history_test is 
    d_top_backup_date timestamp(3); 
begin 

    SET TRANSACTION ISOLATION LEVEL SERIALIZABLE; -- isolate all these statements 

    select top_backup_date into d_top_backup_date from real_data_top_backup_date 
    FOR UPDATE OF top_backup_date; -- lock the table to any other session 
    Insert Into history_data Select * From real_data where record_time <d_top_backup_date; 
    delete from real_data where record_time <d_top_backup_date; 
    Update real_data_top_backup_date Set top_backup_date=(d_top_backup_date+1/24/60); 

    commit; -- reverts the isolation level 

end copy_to_history_test; 

請注意，我還鎖定了real_data_top_backup_date表。在多用戶環境中，最好預留更新記錄以防止因衝突導致交易失敗。

該文檔詳細介紹了隔離級別。 Find out more。

「數據庫的另一個更好的設計？」

那麼這取決於你想要達到的目標。 「實時」表格的要點是什麼？看起來你只會記錄它一分鐘。所以相關的問題是，爲什麼你不插入分區表？你有兩張桌子來證明這麼多的努力有什麼價值？

「如果傳感器數據只插入到歷史數據表中，實時圖形顯示不能保證，因爲檢索數據會隨着數據表的增長而變得更慢。」

你有沒有做過任何基準證明？選擇一分鐘的數據應該相當穩定，因爲您正在主鍵上執行索引範圍掃描。

無論如何，如果你的心臟被設置在兩個表結構，我建議你兩個表中同時插入記錄，使用INSERT ALL功能：

insert all 
    into real_data values (....) 
    into history_data values (....) 
select .... 

的多表語法要求INSERT ... SELECT結構，但是我們可以從DUAL中選擇局部變量，或者選擇適合您的用例的任何變量Find out more。

因爲您已經有了history_data中的記錄，您可以從copy_to_history_test()中刪除轉移，只需從real_data表中刪除即可。