一種實時數據處理方法及系統的製作方法
2023-10-24 16:22:22 1
一種實時數據處理方法及系統的製作方法
【專利摘要】本發明公開了一種實時數據處理方法及系統，包括：實時接收數據並將數據放入隊列中；循環讀取隊列中的數據並將讀取的數據放入緩存中；判斷從隊列中讀取數據的時間是否超過預設的時間閾值，如果是，則將緩存中的數據進行聚合，並將聚合後的數據持久化。本發明的實時數據處理方法及系統，預設時間閾值，在處理隊列數據時，每隔固定的時間間隔跳出處理隊列的循環，能夠保證定期的持久化，也能夠保證將相同維度的數據聚合成儘量少的條數，可以平衡持久化壓力、持久化的周期和聚合效果這三者之間的矛盾關係，避免了由於數據量過大或者數據出入隊列速度不一致造成的種種不良後果。
【專利說明】一種實時數據處理方法及系統
【技術領域】
[0001]本發明涉及數據持久化【技術領域】，尤其涉及一種實時數據處理方法及系統。
【背景技術】
[0002]在實時數據處理系統中，一方面大量帶有維度的數據會按照固定的時間窗口進入隊列，另一方面隊列在處理數據時，期望能快速的將同一個時間窗口內具有相同維度的數據聚合成儘量少的條數，在維度特別多的情況下，聚合後的條目數量也是巨大的，考慮到持久化聚合後的數據對伺服器的壓力，也需要儘量減少每次持久化的數據量，也就是說在一個時間窗口內的數據尚未完全放入緩存時，需要先將已經放入緩存中的數據聚合併進行持久化。
[0003]現有的實時數據處理方法為:從隊列中取數據後，將取出的數據放入緩存中等待聚合併持久化，如果隊列為空，則將緩存中的數據聚合併進行持久化，如果隊列不為空，則只將隊列中的數據取出後放入緩存中。當從隊列中取數據的速度超過數據進入隊列的速度時，則現有技術可能會導致每取出一條數據後便立即進行一次聚合併持久化，持久化會特別頻繁。而當隊列中取數據的速度小於數據進入隊列的速度時，可能會導致隊列中數據大量積壓，隊列中數據如果一直取不完的話，則會導致緩存中的一直不能進行聚合併持久化，數據處理結果便不再是實時的，如果隊列中的數據在積壓很長時間後終於為空了，那麼將緩存中的數據聚合後的數據量通常也會更多，一次性存儲大量數據對伺服器的壓力也會瞬間變大。以上這兩種情況都不是期望的，所期望的是同一個時間窗口內的相同維度下的數據聚合後的條數儘量少，而且能夠儘快的持久化，並且每次存儲時的數據儘量少以減輕對伺服器的壓力。
【發明內容】
[0004]有鑑於此，本發明要解決的一個技術問題是提供一種實時數據處理方法，通過預設時間閾值，在處理隊列數據時每隔固定的時間間隔進行數據持久化。
[0005]一種實時數據處理方法，包括:實時接收數據並將所述數據放入隊列中；循環讀取所述隊列中的數據並將讀取的數據放入緩存中；判斷從所述隊列中讀取數據的時間是否超過預設的時間閾值，如果是，則將所述緩存中的數據進行聚合，並將聚合後的數據持久化。
[0006]根據本發明的一個實施例，進一步的，所述判斷從所述隊列中讀取數據的時間是否超過預設的時間閾值、如果是、則將所述緩存中的數據進行聚合、並將聚合後的數據持久化包括:將開始循環讀取所述隊列中數據的當前時間設置為第一時間Tl並記錄；在每次從所述隊列中取出一條數據並放入所述緩存時，判斷當前時間和Tl之間的時間差是否超過所述時間閾值，如果超過所述時間閾值，則跳出讀取所述隊列中數據的循環，將緩存中的數據進行聚合，並將聚合後的數據持久化；如果未超過所述時間閾值，則繼續循環進行從所述隊列中讀取下一條數據的操作。
[0007]根據本發明的一個實施例，進一步的，所述方法還包括:根據設定的聚合結果的最小更新周期T2和允許持久化的最小時間間隔T3設置所述時間閾值；其中，所述時間閾值小於等於T2，並且大於等於T3。
[0008]根據本發明的一個實施例，進一步的，所述根據設定的聚合結果的最小更新周期T2和允許持久化的最小時間間隔T3設置所述時間閾值包括:將所述時間閾值的分別設置為第一時間閾值和第二時間閾值，並進行所述數據的持久化處理；其中，第一時間閾值的值為T2，第二時間閾值的值為第一時間閾值的一半；分別獲取所述時間閾值為第一時間閾值的第一數據持久化結果和所述時間閾值為第二時間閾值時的第二數據持久化結果，並判斷所述第一數據持久化結果和所述第二數據持久化結果是否都符合預設的持久化檢測策略；所述數據持久化結果包括:聚合程度和存儲的行數；如果是，則將第二時間閾值的值賦給第一時間閾值，並將第二時間閾值的值設置為新的第一時間閾值的一半，進行數據的持久化處理並檢測數據持久化結果，依此循環執行設置第一時間閾值和第二時間閾值並進行所述數據的持久化處理，直到判斷所述第一數據持久化結果和所述第二數據持久化結果不都符合預設的持久化檢測策略時，則跳出執行設置第一時間閾值和第二時間閾值並進行所述數據的持久化處理的循環，確定所述時間閾值為此時的第一時間閾值；如果否，則確定所述時間閾值為T2。
[0009]根據本發明的一個實施例，進一步的，將緩存中的數據按照不同的維度進行聚合，將具有相同維度的數據聚合成一條數據，並將聚合完成後的數據存儲在資料庫或文件中。
[0010]本發明要解決的另一個技術問題是提供一種實時數據處理系統，通過預設時間閾值，在處理隊列數據時每隔固定的時間間隔進行數據持久化。
[0011]一種實時數據處理系統，包括:數據接收單元，用於實時接收數據並將所述數據放入隊列中；數據緩存單元，用於循環讀取所述隊列中的數據並將讀取的數據放入緩存中；持久化單元，用於判斷從所述隊列中讀取數據的時間是否超過預設的時間閾值，如果是，則將所述緩存中的數據進行聚合，並將聚合後的數據持久化。
[0012]根據本發明的一個實施例，進一步的，所述數據緩存單元包括數據讀取記錄子模塊，用於將開始循環讀取所述隊列中數據的當前時間設置為第一時間Tl並記錄；所述持久化單元包括數據聚合子模塊，用於在每次所述數據緩存單元從所述隊列中取出一條數據並放入所述緩存時，判斷當前時間和Tl之間的時間差是否超過所述時間閾值，如果超過所述時間閾值，則跳出讀取所述隊列中數據的循環，將緩存中的數據進行聚合，並將聚合後的數據持久化，如果未超過所述時間閾值，則所述數據緩存單元繼續循環進行從所述隊列中讀取下一條數據的操作。
[0013]根據本發明的一個實施例，進一步的，時間閾值設定單元，用於根據設定的聚合結果的最小更新周期T2和允許的持久化的最小時間間隔T3設置所述時間閾值；其中，所述時間閾值小於等於T2，並且大於等於T3。
[0014]根據本發明的一個實施例，進一步的，所述時間閾值設定單元包括:試驗閾值設定模塊，用於將所述時間閾值的分別設置為第一時間閾值和第二時間閾值，並進行所述數據的持久化處理；其中，第一時間閾值的值為T2，第二時間閾值的值為第一時間閾值的一半；持久化試驗模塊，用於分別獲取所述時間閾值為第一時間閾值的第一數據持久化結果和所述時間閾值為第二時間閾值時的第二數據持久化結果，並判斷所述第一數據持久化結果和所述第二數據持久化結果是否都符合預設的持久化檢測策略；如果是，則所述試驗閾值設定單元將第二時間閾值的值賦給第一時間閾值，並將第二時間閾值的值設置為新的第一時間閾值的一半，進行數據的持久化處理並檢測數據持久化結果，依此循環執行設置第一時間閾值和第二時間閾值並進行所述數據的持久化處理，直到判斷所述第一數據持久化結果和所述第二數據持久化結果不都符合預設的持久化檢測策略時，則跳出執行設置第一時間閾值和第二時間閾值並進行所述數據的持久化處理的循環；確定所述時間閾值為此時的第一時間閾值；如果否，則確定所述時間閾值為T2。
[0015]根據本發明的一個實施例，進一步的，所述持久化單元包括持久處理子模塊，用於將緩存中的數據按照不同的維度進行聚合，將具有相同維度的數據聚合成一條數據，並將聚合完成後的數據存儲在資料庫或文件中。
[0016]本發明的實時數據處理方法及系統，預設時間閾值，在處理隊列數據時，每隔固定的時間間隔跳出處理隊列的循環，能夠保證定期的持久化，也能夠保證將相同維度的數據聚合成儘量少的條數，可以平衡持久化壓力、持久化的周期和聚合效果這三者之間的矛盾關係，避免了由於數據量過大或者數據出入隊列速度不一致造成的種種不良後果。
【專利附圖】
【附圖說明】
[0017]為了更清楚地說明本發明實施例或現有技術中的技術方案，下面將對實施例或現有技術描述中所需要使用的附圖作一簡單地介紹，顯而易見地，下面描述中的附圖僅僅是本發明的一些實施例，對於本領域普通技術人員來講，在不付出創造性勞動性的前提下，還可以根據這些附圖獲得其他的附圖。
[0018]圖1為根據本發明的實時數據處理方法的一個實施例的流程圖；
[0019]圖2為根據本發明的實時數據處理系統的一個實施例的示意圖。
【具體實施方式】
[0020]下面參照附圖對本發明進行更全面的描述，其中說明本發明的示例性實施例。下面將結合本發明實施例中的附圖，對本發明實施例中的技術方案進行清楚、完整地描述，顯然，所描述的實施例僅僅是本發明一部分實施例，而不是全部的實施例。基於本發明中的實施例，本領域普通技術人員在沒有做出創造性勞動前提下所獲得的所有其他實施例，都屬於本發明保護的範圍。
[0021]圖1為根據本發明的實時數據處理方法的一個實施例的流程圖，如圖1所示:
[0022]步驟101，實時接收數據並將數據放入隊列中。
[0023]步驟102，循環讀取隊列中的數據並將讀取的數據放入緩存中。
[0024]步驟103，判斷從隊列中讀取數據的時間是否超過預設的時間閾值。
[0025]步驟104，如果是，則將緩存中的數據進行聚合，並將聚合後的數據持久化。
[0026]本發明的實時數據處理方法，通過預設時間閾值的方式，在處理隊列數據時，每隔固定的時間間隔跳出處理隊列的循環，這樣保證將固定時間間隔內的緩存中的數據進行一次聚合併進行持久化，在執行聚合和存儲時，不再考慮隊列中數據是否處理完，只要設置一個合理的時間閾值即可，既保證了定期的持久化，也能夠保證將相同維度的數據聚合成儘量少的條數。
[0027]在一個實施例中，可以啟動單線程或多線程完成數據持久化的操作，為了保證處理數據的先後順序是接收到數據的順序一致，所以將接收到的數據放入了隊列中。將單線程或多線程開始循環讀取隊列中數據的當前時間設置為第一時間Tl並記錄，然後開始讀取隊列中的數據並將讀取到的數據放入緩存集合中。
[0028]在每次從隊列中取出一條數據並放入緩存時，判斷當前時間和Tl之間的時間差是否超過時間閾值，如果超過時間閾值，則跳出當前從隊列中讀取數據的循環，將緩存中的數據進行聚合，並將聚合後的數據持久化；如果未超過時間閾值，則繼續循環進行從隊列中讀取下一條數據的操作。
[0029]在一個實施例中，還包括根據設定的聚合結果的最小更新周期T2和允許持久化的最小時間間隔T3設置時間閾值。已知需要聚合結果的最小更新周期是T2(即最小需要每個T都會有數據更新)和所允許的持久化的最小時間間隔Τ3，那麼聚合併持久化的時間周期肯定要小於等於Τ2，並且大於等於Τ3。
[0030]在一個實施例中，可以單獨啟動一個進程或任務，用於自動確定需要設置的時間閾值。啟動進程或任務，將時間閾值的分別設置為第一時間閾值和第二時間閾值，並進行數據的持久化處理。第一時間閾值的值為Τ2，第二時間閾值的值為第一時間閾值的一半。
[0031]分別獲取時間閾值為第一時間閾值和第二時間閾值時的數據持久化結果，並判斷2個數據持久化結果是否都符合預設的持久化檢測策略。數據持久化結果包括:聚合程度和存儲的行數。
[0032]如果是，則將第二時間閾值的值賦給第一時間閾值，並將第二時間閾值的值設置為新的第一時間閾值的一半，進行數據的持久化處理並檢測數據持久化結果，依此循環執行設置第一時間閾值和第二時間閾值並進行數據的持久化處理，直到判斷2個數據持久化結果不都符合預設的持久化檢測策略時，則跳出循環，確定時間閾值為此時的第一時間閾值；如果否，則確定時間閾值為Τ2。
[0033]在一個實施例中，分兩次分別測試將閾值設為Τ2和Τ2的一半，測試此兩種情況下最終數據持久化的結果，對比兩者存儲的行數以及聚合程度(是否將相同維度的數據聚合成一條)差別，如果持久化的壓力特別大，那麼更看重的是閾值為Τ2的二分之一時持久化的行數有沒有成倍的減少，如果是更關注的是聚合效果的話，那麼需要分析閾值為Τ2的二分之一時聚合程度是否依然理想。
[0034]閾值越大，聚合的越理想，存儲的壓力越大；閾值越小，聚合效果越差，存儲的壓力也越小。如果閾值為Τ2和二分之一 Τ2時聚合效果和持久化的行數都相差不大的話，那麼將閾值設為四分之一 Τ2，和二分之一 Τ2的持久化結果進行比較，以此類推....，確定設置的時間閾值的原則，即持久化檢測策略可以是:儘量減少一次持久化對伺服器的壓力，同時保證聚合效果要好，同時因為是實時數據處理，當然也要出結果的速度越快越好。
[0035]啟動用於時間閾值設定的程序或進程，設置各種參數，包括:Τ1、Τ2、持久化結果的判定策略，按照以上的測試方法在經過大量測試後，能夠自動找到期望的閾值。
[0036]在一個實施例中，將緩存中的數據按照不同的維度進行聚合，將具有相同維度的數據聚合成一條數據，並將聚合完成後的數據存儲在資料庫或文件中。可以根據具體的應用場景設置多個維度，形成一個維度體系，具備訪問和過濾事實的能力，包括一個完整的維度體系編碼、關鍵詞以及相關的表示。例如，時間維度包括年、季度、月、日等層次，地區維度包括國家、省、市等層次等等。將緩存中的數據按照不同的維度進行聚合，將聚合完成後的數據持久化，存儲方式可以採用現有的技術。
[0037]本發明的實時數據處理方法，能夠保證將固定時間間隔內的緩存中的數據進行一次聚合併進行持久化，在執行聚合和存儲時，不再考慮隊列中數據是否處理完，並能通過自動進行數據持久化的測試設置一個合理的時間閾值，既保證了定期的持久化，也能夠保證將相同維度的數據聚合成儘量少的條數。
[0038]如圖2所示，本發明提供一種實時數據處理系統，包括:數據接收單元21、數據緩存單元22和持久化單元23。數據接收單元21實時接收數據並將數據放入隊列中。數據緩存單元22循環讀取隊列中的數據並將讀取的數據放入緩存中。持久化單元23判斷從隊列中讀取數據的時間是否超過預設的時間閾值，如果是，則將緩存中的數據進行聚合，並將聚合後的數據持久化。
[0039]在一個實施例中，數據緩存單元22的數據讀取記錄子模塊將開始循環讀取隊列中數據的當前時間設置為第一時間Tl並記錄。持久化單元23的數據聚合子模塊在每次數據緩存單元從隊列中取出一條數據並放入緩存時，判斷當前時間和Tl之間的時間差是否超過時間閾值，如果超過時間閾值，則跳出當前從隊列中讀取數據的循環，將緩存中的數據進行聚合，並將聚合後的數據持久化，如果未超過時間閾值，則數據緩存單元22繼續循環進行從隊列中讀取下一條數據的操作。
[0040]在一個實施例中，時間閾值設定單元24根據設定的聚合結果的最小更新周期T2和允許的持久化的最小時間間隔T3設置時間閾值。時間閾值小於等於T2，並且大於等於T3。
[0041 ] 在一個實施例中，時間閾值設定單元24包括:試驗閾值設定模塊241和持久化試驗模塊242。試驗閾值設定模塊241將時間閾值的分別設置為第一時間閾值和第二時間閾值，並進行數據的持久化處理，第一時間閾值的值為T2，第二時間閾值的值為第一時間閾值的一半。
[0042]持久化試驗模塊242分別獲取時間閾值為第一時間閾值和第二時間閾值時的數據持久化結果，並判斷2個數據持久化結果是否都符合預設的持久化檢測策略。如果是，則試驗閾值設定單元241將第二時間閾值的值賦給第一時間閾值，並將第二時間閾值的值設置為新的第一時間閾值的一半，進行數據的持久化處理並檢測數據持久化結果，依此循環執行設置第一時間閾值和第二時間閾值並進行數據的持久化處理，直到判斷2個數據持久化結果不都符合預設的持久化檢測策略時，則跳出循環；確定時間閾值為此時的第一時間閾值；如果否，則確定時間閾值為T2。
[0043]在一個實時例中，持久化單元241的持久處理子模塊將緩存中的數據按照不同的維度進行聚合，將具有相同維度的數據聚合成一條數據，並將聚合完成後的數據存儲在資料庫或文件中。
[0044]本發明的實時數據處理方法及系統，預設時間閾值，在處理隊列數據時，每隔固定的時間間隔跳出處理隊列的循環，能夠保證定期的持久化，也能夠保證將相同維度的數據聚合成儘量少的條數，能夠平衡持久化壓力、持久化的周期和聚合效果這三者之間的矛盾關係，避免了由於數據量過大或者數據出入隊列速度不一致造成的種種不好後果。
[0045]可能以許多方式來實現本發明的方法和系統。例如，可通過軟體、硬體、固件或者軟體、硬體、固件的任何組合來實現本發明的方法和系統。用於方法的步驟的上述順序僅是為了進行說明，本發明的方法的步驟不限於以上具體描述的順序，除非以其它方式特別說明。此外，在一些實施例中，還可將本發明實施為記錄在記錄介質中的程序，這些程序包括用於實現根據本發明的方法的機器可讀指令。因而，本發明還覆蓋存儲用於執行根據本發明的方法的程序的記錄介質。
[0046]本發明的描述是為了示例和描述起見而給出的，而並不是無遺漏的或者將本發明限於所公開的形式。很多修改和變化對於本領域的普通技術人員而言是顯然的。選擇和描述實施例是為了更好說明本發明的原理和實際應用，並且使本領域的普通技術人員能夠理解本發明從而設計適於特定用途的帶有各種修改的各種實施例。
【權利要求】
1.一種實時數據處理方法，其特徵在於，包括: 實時接收數據並將所述數據放入隊列中； 循環讀取所述隊列中的數據並將讀取的數據放入緩存中； 判斷從所述隊列中讀取數據的時間是否超過預設的時間閾值，如果是，則將所述緩存中的數據進行聚合，並將聚合後的數據持久化。
2.如權利要求1所述的方法，其特徵在於，所述判斷從所述隊列中讀取數據的時間是否超過預設的時間閾值、如果是、則將所述緩存中的數據進行聚合、並將聚合後的數據持久化包括: 將開始循環讀取所述隊列中數據的當前時間設置為第一時間Tl並記錄； 在每次從所述隊列中取出一條數據並放入所述緩存時，判斷當前時間和Tl之間的時間差是否超過所述時間閾值，如果超過所述時間閾值，則跳出讀取所述隊列中數據的循環，將緩存中的數據進行聚合，並將聚合後的數據持久化；如果未超過所述時間閾值，則繼續循環進行從所述隊列中讀取下一條數據的操作。
3.如權利要求2所述的方法，其特徵在於，所述方法還包括: 根據設定的聚合結果的最小更新周期T2和允許持久化的最小時間間隔T3設置所述時間閾值；其中，所述時間閾值小於等於T2，並且大於等於T3。
4.如權利要求3所述的方法，其特徵在於，所述根據設定的聚合結果的最小更新周期T2和允許持久化的最小時間間隔T3設置所述時間閾值包括: 將所述時間閾值的分別設置為第一時間閾值和第二時間閾值，並進行所述數據的持久化處理；其中，第一時間閾值的值為T2，第二時間閾值的值為第一時間閾值的一半； 分別獲取所述時間閾值為第一時間閾值的第一數據持久化結果和所述時間閾值為第二時間閾值時的第二數據持久化結果，並判斷所述第一數據持久化結果和所述第二數據持久化結果是否都符合預設的持久化檢測策略；所述數據持久化結果包括:聚合程度和存儲的行數； 如果是，則將第二時間閾值的值賦給第一時間閾值，並將第二時間閾值的值設置為新的第一時間閾值的一半，進行數據的持久化處理並檢測數據持久化結果，依此循環執行設置第一時間閾值和第二時間閾值並進行所述數據的持久化處理，直到判斷所述第一數據持久化結果和所述第二數據持久化結果不都符合預設的持久化檢測策略時，則跳出此執行設置第一時間閾值和第二時間閾值並進行所述數據的持久化處理的循環，確定所述時間閾值為此時的第一時間閾值； 如果否，則確定所述時間閾值為T2。
5.如權利要求1所述的方法，其特徵在於，所述將所述緩存中的數據進行聚合、並將聚合後的數據持久化包括: 將緩存中的數據按照不同的維度進行聚合，將具有相同維度的數據聚合成一條數據，並將聚合完成後的數據存儲在資料庫或文件中。
6.一種實時數據處理系統，其特徵在於，包括: 數據接收單元，用於實時接收數據並將所述數據放入隊列中； 數據緩存單元，用於循環讀取所述隊列中的數據並將讀取的數據放入緩存中； 持久化單元，用於判斷從所述隊列中讀取數據的時間是否超過預設的時間閾值，如果是，則將所述緩存中的數據進行聚合，並將聚合後的數據持久化。
7.如權利要求6所述的系統，其特徵在於: 所述數據緩存單元包括數據讀取記錄子模塊，用於將開始循環讀取所述隊列中數據的當前時間設置為第一時間Tl並記錄； 所述持久化單元包括數據聚合子模塊，用於在每次所述數據緩存單元從所述隊列中取出一條數據並放入所述緩存時，判斷當前時間和Tl之間的時間差是否超過所述時間閾值，如果超過所述時間閾值，則跳出讀取所述隊列中數據的循環，將緩存中的數據進行聚合，並將聚合後的數據持久化，如果未超過所述時間閾值，則所述數據緩存單元繼續循環進行從所述隊列中讀取下一條數據的操作。
8.如權利要求7所述的系統，其特徵在於，還包括: 時間閾值設定單元，用於根據設定的聚合結果的最小更新周期T2和允許的持久化的最小時間間隔T3設置所述時間閾值；其中，所述時間閾值小於等於T2，並且大於等於T3。
9.如權利要求8所述的系統，其特徵在於: 所述時間閾值設定單元包括: 試驗閾值設定模塊，用於將所述時間閾值的分別設置為第一時間閾值和第二時間閾值，並進行所述數據的持久化處理；其中，第一時間閾值的值為T2，第二時間閾值的值為第一時間閾值的一半； 持久化試驗模塊，用於分別獲取所述時間閾值為第一時間閾值的第一數據持久化結果和所述時間閾值為第二時間閾值時的第二數據持久化結果，並判斷所述第一數據持久化結果和所述第二數據持久化結果是否都符合預設的持久化檢測策略；如果是，則所述試驗閾值設定單元將第二時間閾值的值賦給第一時間閾值，並將第二時間閾值的值設置為新的第一時間閾值的一半，進行數據的持久化處理並檢測數據持久化結果，依此循環執行設置第一時間閾值和第二時間閾值並進行所述數據的持久化處理，直到判斷所述第一數據持久化結果和所述第二數據持久化結果不都符合預設的持久化檢測策略時，則跳出此執行設置第一時間閾值和第二時間閾值並進行所述數據的持久化處理的循環；確定所述時間閾值為此時的第一時間閾值；如果否，則確定所述時間閾值為T2。
10.如權利要求6至9任意一項所述的系統，其特徵在於: 所述持久化單元包括持久處理子模塊，用於將緩存中的數據按照不同的維度進行聚合，將具有相同維度的數據聚合成一條數據，並將聚合完成後的數據存儲在資料庫或文件中。
【文檔編號】G06F17/30GK104317958SQ201410645385
【公開日】2015年1月28日 申請日期:2014年11月12日 優先權日:2014年11月12日
【發明者】郭濤, 王鵬 申請人:北京國雙科技有限公司