实时数据表到历史数据表的数据丢失

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表中删除即可。