DataLeap 資料資産實戰：如何實作存儲優化？

更多技術交流、求職機會，歡迎關注位元組跳動資料平台微信公衆号，回複【1】進入官方交流群

背景

• DataLeap 作為一站式資料中台套件，彙集了位元組内部多年積累的資料內建、開發、運維、治理、資産、安全等全套資料中台建設的經驗，助力企業客戶提升資料研發治理效率、降低管理成本。
• Data Catalog 是一種中繼資料管理的服務，會收集技術中繼資料，并在其基礎上提供更豐富的業務上下文與語義，通常支援中繼資料編目、查找、詳情浏覽等功能。目前 Data Catalog 作為火山引擎大資料研發治理套件 DataLeap 産品的核心功能之一，經過多年打磨，服務于位元組跳動内部幾乎所有核心業務線，解決了資料生産者和消費者對于中繼資料和資産管理的各項核心需求。
• Data Catalog 系統的存儲層，依賴 Apache Atlas，傳遞依賴 JanusGraph。JanusGraph 的存儲後端，通常是一個 Key-Column-Value 模型的系統，本文主要講述了使用 MySQL 作為 JanusGraph 存儲後端時，在設計上面的思考，以及在實際過程中遇到的一些問題。

起因

實際生産環境，我們使用的存儲系統維護成本較高，有一定的運維壓力，于是想要尋求替代方案。在這個過程中，我們試驗了很多存儲系統，其中 MySQL 是重點投入調研和開發的備選之一。

另一方面，除了位元組内部外，在 ToB 場景，MySQL 的運維成本也會明顯小于其他大資料元件，如果 MySQL 的方案跑通，我們可以在 ToB 場景多一種選擇。

基于以上兩點，我們投入了一定的人力調研和實作基于 MySQL 的存儲後端。

方案評估

在設計上，JanusGraph 的存儲後端是可插拔的，隻要做對應的适配即可，并且官方已經支援了一批存儲系統。結合位元組的技術棧以及我們的訴求，做了以下的評估。

各類存儲系統比較

• 因投入成本過高，我們不接受自己運維有狀态叢集，排除了 HBase 和 Cassandra；
• 從目前資料量與将來的可擴充性考慮，單機方案不可選，排除了 BerkeleyDB；
• 同樣因為人力成本，需要做極大量開發改造的方案暫時不考慮，排除了 Redis。

最終我們挑選了 MySQL 來推進到下一步。

MySQL 的理論可行性

• 可以支援 Key-Value（後續簡稱 KV 模型）或者 Key-Column-Value（後續簡稱 KCV 模型）的存儲模型，聚集索引 B+樹排序通路，支援基于 Key 或者 Key-Column 的 Range Query，所有查詢都走索引，且避免記憶體中重排序，效率初步判斷可接受。
• 中台内的其他系統，最大的 MySQL 單表已經到達億級别，且 MySQL 有成熟的分庫分表解決方案，判斷資料量可以支援。
• 在具體使用場景中，對于寫入的效率要求不高，因為大量的資料都是離線任務完成，判斷 MySQL 在寫入上的效率不會成為瓶頸。

總體設計

• 維護一張 Meta 表做 lookup 用，Meta 表中存儲租戶與 DataSource（庫）之間的映射關系，以及 Shards 等租戶級别的配置資訊。
• StoreManager 作為入口，在 openTransaction 的時候将租戶資訊注入到 StoreTransaction 中，并傳回租戶級别的 DataSource。
• StoreManager 中以 name 為 Key，維護一組 Store，Store 與存儲的資料類型有關，具有跨租戶能力
• 常見的 Store 有system_properies，tx_log，graphindex，edgestore等
• 對于 MySQL 最終的讀寫，都收斂在 Store，方法簽名中傳入 StoreTransaction，Store 從中取出租戶資訊和資料庫連接配接，進行資料讀寫。
• 對于單租戶來說，資料可以分表（shards），對于某個特定的 key 來說，存儲和讀取某個 shard，是根據 ShardManager 來決定
• 典型的 ShardManager 邏輯，是根據總 shard 數對 key 做 hash 決定，預設單分片。
• 對于每個 Store，表結構是 4 列（id, g_key, g_column, g_value)，除自增 ID 外，對應 key-column-value model 的資料模型，key+column 是一個聚集索引。
• Context 中的租戶資訊，需要在操作某個租戶資料之前設定，并在操作之後清除掉。

細節設計與疑難問題

細節設計

存儲模型

JanusGraph 要求 column-family 類型存儲（如 Cassandra, HBase），也就是說，資料存儲由一系列行組成，每行都由一個鍵（key）唯一辨別，每行由多個列值（column-value）對組成，也會對列進行排序和過濾，如果是非 column-family 的類型存儲，則需要另行适配，适配時資料模型有兩種方式：Key-Column-Value 和 Key-Value。

KCV 模型：

• 會将 key\column\value 在存儲中區分開來。
• 對應的接口為：KeyColumnValueStoreManager。

KV 模型：

• 在存儲中僅有 key 和 value 兩部分，此處的 key 相當于 KVC 模型中的 key+column；
• 如果要根據 column 進行過濾，需要額外的适配工作；
• 對應的接口為：KeyValueStoreManager，該接口有子類OrderedKeyValueStoreManager，提供了保證查詢結果有序性的接口；
• 同時提供了OrderedKeyValueStoreManagerAdapter接口，用于對 Key-Column-Value 模型進行适配，将其轉化為 Key-Value 模型。

MySQL 的存儲實作采用了 KCV 模型，每個表會有 4 列，一個自增的 ID 列，作為主鍵，同時還有 3 列分别對應模型中的 key\column\value，資料庫中的一條記錄相當于一個獨立的 KCV 結構，多行資料庫記錄代表一個點或者邊。

表中 key 和 column 這兩列會組成聯合索引，既保證了根據 key 進行查詢時的效率，也支援了對 column 的排序以及條件過濾。

多租戶

存儲層面：預設情況下，JanusGraph 會需要存儲edgestore, graphindex, system_properties, txlog等多種資料類型，每個類型在 MySQL 中都有各自對的表，且表名使用租戶名作為字首，如tenantA_edgestore，這樣即使不同租戶的資料在同一個資料庫，在存儲層面租戶之間的資料也進行了隔離，減少了互相影響，友善日常運維。（理論上每個租戶可以單獨配置設定一個資料庫）

具體實作：每個租戶都會有各自的 MySQL 連接配接配置，啟動之後會為各個租戶分别初始化資料庫連接配接，所有和 JanusGraph 的請求都會通過 Context 傳遞租戶資訊，以便在操作資料庫時選擇該租戶對應的連接配接。

具體代碼：

• MysqlKcvTx：實作了AbstractStoreTransaction，對具體的 MySQL 連接配接進行了封裝，負責和資料庫的互動，它的commit和rollback方法由封裝的 MySQL 連接配接真正完成。
• MysqlKcvStore：實作了KeyColumnValueStore，是具體執行讀寫操作的入口，每一個類型的 Store 對應一個MysqlKcvStore執行個體，MysqlKcvStore處理讀寫邏輯時，根據租戶資訊完全自主組裝 SQL 語句，SQL 語句會由MysqlKcvTx真正執行。
• MysqlKcvStoreManager：實作了KeyColumnValueStoreManager，作為管理所有 MySQL 連接配接和租戶的入口，也維護了所有 Store 和MysqlKcvStore對象的映射關系。在處理不同租戶對不同 Store 的讀寫請求時，根據租戶資訊，建立MysqlKcvTx對象，并将其配置設定給對應的MysqlKcvStore去執行。

public class MysqlKcvStoreManager implements KeyColumnValueStoreManager {
    
    @Override
    public StoreTransaction beginTransaction(BaseTransactionConfig config) throws BackendException {
        String tenant = TenantContext.getTenant();
        if (!tenantToDataSourceMap.containsKey(tenant)) {
            try {
                // 初始化單個租戶的DataSource
                initSingleDataSource(tenant);
            } catch (SQLException e) {
                log.error("init mysql database source failed due to", e);
                throw new BackendSQLException(String.format("init mysql database source failed due to", e.getMessage()));
            }
        }
        // 擷取資料庫連接配接
        Connection connection = tenantToDataSourceMap.get(tenant).getConnection(false);
        return new MysqlKcvTx(config, tenant, connection);
    }
}           

事務

幾乎所有與 JanusGraph 的互動都會開啟事務，而且事務對于多個線程并發使用是安全的，但是 JanusGraph 的事務并不都支援 ACID，是否支援會取決于底層存儲元件，對于某些存儲元件來說，提供可序列化隔離機制或者多行原子寫入代價會比較大。

JanusGraph 中的每個圖形操作都發生在事務的上下文中，根據 TinkerPop 的事務規範，每個線程執行圖形上的第一個操作時便會打開針對圖形資料庫的事務，所有圖形元素都與檢索或者建立它們的事務範圍相關聯，在使用commit或者rollback方法顯式的關閉事務之後，與該事務關聯的圖形元素都将過時且不可用。

JanusGraph 提供了AbstractStoreTransaction接口，該接口包含commit和rollback的操作入口，在 MySQL 存儲的實作中，MysqlKcvTx實作了AbstractStoreTransaction，對具體的 MySQL 連接配接進行了封裝，在其commit和rollback方法中調用 SQL 連接配接的commit和rollback方法，以此實作對于 JanusGraph 事務的支援。

public class MysqlKcvTx extends AbstractStoreTransaction {


    private static final Logger log = LoggerFactory.getLogger(MysqlKcvTx.class);


    private final Connection connection;
    @Getter
    private final String tenant;


    public MysqlKcvTx(BaseTransactionConfig config, String tenant, Connection connection) {
        super(config);
        this.tenant = tenant;
        this.connection = connection;
    }


    @Override
    public synchronized void commit() {
        try {
            if (Objects.nonNull(connection)) {
                connection.commit();
                connection.close();
            }
            if (log.isDebugEnabled()) {
                log.debug("tx has been committed");
            }
        } catch (SQLException e) {
            log.error("failed to commit transaction", e);
        }
    }


    @Override
    public synchronized void rollback() {
        try {
            if (Objects.nonNull(connection)) {
                connection.rollback();
                connection.close();
            }
            if (log.isDebugEnabled()) {
                log.debug("tx has been rollback");
            }
        } catch (SQLException e) {
            log.error("failed to rollback transaction", e);
        }
    }


    public Connection getConnection() {
        return connection;
    }
}           

資料庫連接配接池

Hikari 是 SpringBoot 内置的資料庫連接配接池，快速、簡單，做了很多優化，如使用 FastList 替換 ArrayList，自行研發無所集合類 ConcurrentBag，位元組碼精簡等，在性能測試中表現的也比其他競品要好。

Druid 是另一個也非常優秀的資料庫連接配接池，為監控而生，内置強大的監控功能，監控特性不影響性能。功能強大，能防 SQL 注入，内置 Loging 能診斷 Hack 應用行為。

關于兩者的對比很多，此處不再贅述，雖然 Hikari 的性能号稱要優于 Druid，但是考慮到 Hikari 監控功能比較弱，最終在實作的時候還是選擇了 Druid。

疑難問題

連接配接逾時

現象：在進行資料導入測試時，服務報錯" The last packet successfully received from the server was X milliseconds ago"，導緻資料寫入失敗。

原因：存在超大 table（有 8000 甚至 10000 列），這些 table 的中繼資料處理非常耗時（10000 列的可能需要 30 分鐘），而且在處理過程中有很長一段時間和資料庫并沒有互動，資料庫連接配接一直空閑。

解決辦法：

• 調整 mysql server 端的 wait_timeout 參數，已調整到 3600s。
• 調整 client 端資料庫配置中連接配接的最小空閑時間，已調整到 2400s。

分析過程：

1. 懷疑是 mysql client 端沒有增加空閑清理或者保活機制，conneciton 線上程池中長時間沒有使用，mysql 服務端已經關閉該連結導緻。嘗試修改用戶端 connection 空閑時間，增加 validationQuery 等常見措施，無果；
2. 根據打點發現單條消息處理耗時過高，疑似線程卡死；
3. 新增打點發現線程沒卡死，隻是在執行一些非常耗時的邏輯，這時候已經擷取到了資料庫連接配接，但是在執行那些耗時邏輯的過程中和資料庫沒有任何互動，長時間沒有使用資料庫連接配接，最終導緻連接配接被回收；
4. 調高了 MySQL server 端的 wait_timeout，以及 client 端的最小空閑時間，問題解決。

并行寫入死鎖

現象：線程 thread-p-3-a-0 和線程 thread-p-7-a-0 在執行過程中都出現 Deadlock。

具體日志如下：

[thread-p-3-a-0] ERROR org.janusgraph.diskstorage.mysql.MysqlKcvStore 313 - failed to insert query:INSERT INTO default_edgestore (g_key, g_column, g_value) VALUES (?,?,?) ON DUPLICATE KEY UPDATE g_value=?, params: key=A800000000001500, column=55A0, value=008000017CE616D0DD03674495
com.mysql.cj.jdbc.exceptions.MySQLTransactionRollbackException: Deadlock found when trying to get lock; try restarting transaction


[thread-p-7-a-0] ERROR org.janusgraph.diskstorage.mysql.MysqlKcvStore 313 - failed to insert query:INSERT INTO default_edgestore (g_key, g_column, g_value) VALUES (?,?,?) ON DUPLICATE KEY UPDATE g_value=?, params: key=A800000000001500, column=55A0, value=008000017CE616D8E1036F3495
com.mysql.cj.jdbc.exceptions.MySQLTransactionRollbackException: Deadlock found when trying to get lock; try restarting transaction


[thread-p-3-a-0] ERROR org.janusgraph.diskstorage.mysql.MysqlKcvStore 313 - failed to insert query:INSERT INTO default_edgestore (g_key, g_column, g_value) VALUES (?,?,?) ON DUPLICATE KEY UPDATE g_value=?, params: key=5000000000000080, column=55A0, value=008000017CE616F3C10442108A
com.mysql.cj.jdbc.exceptions.MySQLTransactionRollbackException: Deadlock found when trying to get lock; try restarting transaction


[thread-p-7-a-0] ERROR org.janusgraph.diskstorage.mysql.MysqlKcvStore 313 - failed to insert query:INSERT INTO default_edgestore (g_key, g_column, g_value) VALUES (?,?,?) ON DUPLICATE KEY UPDATE g_value=?, params: key=5000000000000080, column=55A0, value=008000017CE61752B50556208A
com.mysql.cj.jdbc.exceptions.MySQLTransactionRollbackException: Deadlock found when trying to get lock; try restarting transaction           

原因：

1. 結合日志分析，兩個線程并發執行，需要對同樣的多個記錄加鎖，但是順序不一緻，進而導緻了死鎖。
2. 55A0這個 column 對應的 property 是"__modificationTimestamp"，該屬性是 atlas 的系統屬性，當對圖庫中的點或者邊有更新時，對應點或者邊的"__modificationTimestamp"屬性會被更新。在并發導入資料的時候，加劇了資源競争，是以會偶發死鎖問題。

解決辦法：

業務中并沒有用到"__modificationTimestamp"這個屬性，通過修改 Atlas 代碼，僅在建立點和邊的時候為該屬性指派，後續更新時不再更新該屬性，問題得到解決。

性能測試

環境搭建

在位元組内部 JanusGraph 主要用作 Data Catalog 服務的存儲層，關于 MySQL 作為存儲的性能測試并沒有在 JanusGraph 層面進行，而是模拟 Data Catalog 服務的業務使用場景和資料，使用業務接口進行測試，主要會關注接口的響應時間。

接口邏輯有所裁剪，在不影響核心讀寫流程的情況下，屏蔽掉對其他服務的依賴。

模拟單租戶表單分片情況下，庫表中繼資料建立、更新、查詢，表之間血緣關系的建立、查詢，以此反映在圖庫單次讀寫和多次讀寫情況下 MySQL 的表現。

整個測試環境搭建在火山引擎上，總共使用 6 台 8C32G 的機器，硬體條件如下：

測試場景如下：

測試結論

總計 10 萬個表（庫數量為個位數，可忽略）

在 10 萬個表且模拟了表之間血緣關系的情況下，graphindex表的資料量已有 7000 萬，edgestore表的資料量已有 1 億 3000 萬，業務接口的響應時間基本在預期範圍内，可滿足中小規模 Data Catalog 服務的存儲要求。

總結

MySQL 作為 JanusGraph 的存儲，有部署簡單，友善運維等優勢，也能保持良好的擴充性，在中小規模的 Data Catalog 存儲服務中也能保持較好的性能水準，可以作為一個存儲選擇。

市面上也有比較成熟的 MySQL 分庫分表方案，未來可以考慮将其引入，以滿足更大規模的存儲需求。

火山引擎 Data Catalog 産品是基于位元組跳動内部平台，經過多年業務場景和産品能力打磨，在公有雲進行部署和釋出，期望幫助更多外部客戶創造資料價值。目前公有雲産品已包含内部成熟的産品功能同時擴充若幹 ToB 核心功能，正在逐漸對齊業界領先 Data Catalog 雲産品各項能力。

點選跳轉 大資料研發治理套件-火山引擎了解更多