擴展Druid.io支持GaussDB T 數據庫[轉]

轉自?http://3ms.huawei.com/km/blogs/details/6220889

為了滿足某產品的訴求，需要擴展druid.io使之支持高斯數據庫。之前druid.io使用mysql關系數據庫來存儲元數據，現在主要是將元數據存儲方式改為使用高斯數據庫存儲。

在Druid.io官網上找到Druid.io模塊擴展的官方文檔：

https://druid.apache.org/docs/latest/development/modules.html

通過分析代碼了解到Druid.io為了使得代碼具有很強的擴展性使用了Guice框架來進行開發，下面我們來看看如何基于Guice框架擴展Druid.io支持高斯數據庫。

1、創建一個高斯元數據存儲的模塊

在源代碼druid\extensions-core目錄下創建一個名為gaussdb-metadata-storage的maven模塊。

創建類ZenithMetadataStorageModule繼承SQLMetadataStorageDruidModule這個元數據存儲基類，并實現DruidModule這個模塊接口：

package?org.gaussdb.metadata.storage; import?java.util.List; import?com.fasterxml.jackson.databind.Module; import?com.google.common.collect.ImmutableList; import?com.google.inject.Binder; import?com.google.inject.Key; import?io.druid.guice.LazySingleton; import?io.druid.guice.PolyBind; import?io.druid.guice.SQLMetadataStorageDruidModule; import?io.druid.initialization.DruidModule; import?io.druid.metadata.MetadataStorageActionHandlerFactory; import?io.druid.metadata.MetadataStorageConnector; import?io.druid.metadata.MetadataStorageProvider; import?io.druid.metadata.NoopMetadataStorageProvider; import?io.druid.metadata.SQLMetadataConnector; public?class?ZenithMetadataStorageModule?extends?SQLMetadataStorageDruidModule?implements?DruidModule { ????public?static?final?String?TYPE?=?"gaussdb"; ???? ????public?ZenithMetadataStorageModule() ????{ ????????super(TYPE); ????????//?TODO?Auto-generated?constructor?stub ????} ????@Override ????public?List?getJacksonModules() ????{ ????????//?TODO?Auto-generated?method?stub ????????return?ImmutableList.of(); ????} ???? ????@Override ????public?void?configure(Binder?binder) ????{ ??????super.configure(binder); ??????PolyBind ??????????.optionBinder(binder,?Key.get(MetadataStorageProvider.class)) ??????????.addBinding(TYPE) ??????????.to(NoopMetadataStorageProvider.class) ??????????.in(LazySingleton.class); ??????PolyBind ??????????.optionBinder(binder,?Key.get(MetadataStorageConnector.class)) ??????????.addBinding(TYPE) ??????????.to(ZenithConnector.class) ??????????.in(LazySingleton.class); ??????PolyBind ??????????.optionBinder(binder,?Key.get(SQLMetadataConnector.class)) ??????????.addBinding(TYPE) ??????????.to(ZenithConnector.class) ??????????.in(LazySingleton.class); ??????PolyBind.optionBinder(binder,?Key.get(MetadataStorageActionHandlerFactory.class)) ??????????????.addBinding(TYPE) ??????????????.to(ZenithMetadataStorageActionHandlerFactory.class) ??????????????.in(LazySingleton.class); ????} }

同時需要引入對高斯驅動的依賴

????????????com.huawei.gauss ????????????com.huawei.gauss.jdbc.ZenithDriver ????????????V300R001C00SPC100B210

2、注冊高斯元數據存儲模塊

2.1、修改配置

Druid.io所有核心擴展可開箱即用。通過將其名稱添加到common.runtime.properties配置文件的druid.extensions.loadList屬性來加載綁定擴展。

druid.extensions.loadList=["gaussdb-metadata-storage"]

并增加高斯數據庫的參數配置如下：

#?For?GaussDB: druid.metadata.storage.type=gaussdb druid.metadata.storage.connector.connectURI=jdbc:zenith:@10.243.49.xx:1611 druid.metadata.storage.connector.user=druid druid.metadata.storage.connector.password=xxx

注意此處的druid.metadata.storage.type與ZenithMetadataStorageModule代碼中的TYPE是對應的。

2.2、需要在jar的META-INF / services目錄中打包一個額外的文件注冊模塊。

通過在src / main / resources目錄中創建文件，它應該是一個文本文件，類似：

META-INF/services/org.apache.druid.initialization.DruidModule，

其中包含實現DruidModule的包限定類的新行分隔列表，類似：

org.gaussdb.metadata.storage.ZenithMetadataStorageModule

此時，當jar被添加到類路徑或作為擴展時，Druid.io會注意到該文件并將實例化Module的實例。 模塊應該有一個默認的構造函數，但是如果你需要訪問運行時配置屬性，它可以有一個帶@Inject的方法來獲取從Guice注入的一個Properties對象。

3、驗證元數據存儲的實現

3.1、打包

distribution工程是專門用來打包的，為了保證druid.io打包的時候將該jar包打入，需要修改distribution工程的pom文件。

在exec-maven-plugin這個插件中增加參數：

-c io.druid.extensions:gaussdb-metadata-storage

最終在druid-0.12.0-bin.tar.gz\druid-0.12.0\extensions\gaussdb-metadata-storage中可以看到jar包及高斯驅動包。

3.2、適配高斯語法

目前為止我們已經可以部署druid了，但是運行的時候會報很多錯誤和異常，這些錯誤一般是由于代碼中存在與高斯語法不一致的地方導致的，druid.io使用jdbi進行關系數據庫的操作，下面一一列出對這塊操作的修改。

1、ZenithConnector類中：

package?org.gaussdb.metadata.storage; import?java.sql.SQLException; import?java.util.List; import?java.util.Map; import?org.apache.commons.dbcp2.BasicDataSource; import?org.skife.jdbi.v2.DBI; import?org.skife.jdbi.v2.Handle; import?org.skife.jdbi.v2.tweak.HandleCallback; import?com.google.common.base.Supplier; import?com.google.inject.Inject; import?io.druid.java.util.common.StringUtils; import?io.druid.java.util.common.logger.Logger; import?io.druid.metadata.MetadataStorageConnectorConfig; import?io.druid.metadata.MetadataStorageTablesConfig; import?io.druid.metadata.SQLMetadataConnector; public?class?ZenithConnector?extends?SQLMetadataConnector { ????private?static?final?Logger?log?=?new?Logger(ZenithConnector.class); ???? ????private?static?final?String?PAYLOAD_TYPE?=?"BLOB"; ???? ????private?static?final?String?SERIAL_TYPE?=?"Integer?AUTO_INCREMENT"; ???? ????private?static?final?String?QUOTE_STRING?=?""; ???? ????public?static?final?int?DEFAULT_STREAMING_RESULT_SIZE?=?100; ???? ????private?final?DBI?dbi; ???? ????@Inject ????public?ZenithConnector(Supplier?config, ????????????Supplier?tablesConfigSupplier) ????{ ????????super(config,?tablesConfigSupplier); ???????? ????????final?BasicDataSource?datasource?=?getDatasource(); ???????? ????????datasource.setDriverClassLoader(getClass().getClassLoader()); ????????datasource.setDriverClassName("com.huawei.gauss.jdbc.ZenithDriver"); ???????? ????????this.dbi?=?new?DBI(datasource); ???????? ????????log.info("Configured?Zenith?as?metadata?storage"); ????} ???? ????@Override ????protected?String?getPayloadType() ????{ ????????return?PAYLOAD_TYPE; ????} ???? ????@Override ????protected?String?getSerialType() ????{ ????????return?SERIAL_TYPE; ????} ???? ????@Override ????public?String?getQuoteString() ????{ ????????return?QUOTE_STRING; ????} ???? ????@Override ????protected?int?getStreamingFetchSize() ????{ ????????return?DEFAULT_STREAMING_RESULT_SIZE; ????} ???? ????@Override ????public?boolean?tableExists(Handle?handle,?String?tableName) ????{???????? ????????try ????????{ ????????????String?owner?=?handle.getConnection().getMetaData().getUserName(); ????????????List>?list?=?handle ????????????????????.createQuery( ????????????????????????????"select?*?from?dba_tables?where?owner=?:owner?and?table_name=?:tableName") ????????????????????.bind("owner",?owner.toUpperCase()) ????????????????????.bind("tableName",?tableName.toUpperCase()).list(); ????????????log.info("ZenithConnector----"?+?list.size()); ????????????return?!list.isEmpty(); ????????} ????????catch?(SQLException?e) ????????{ ????????????log.error("table?is?not?Exists");???????????? ????????} ????????return?false; ????} ???? ????@Override ????public?Void?insertOrUpdate(final?String?tableName,?final?String?keyColumn, ????????final?String?valueColumn,?final?String?key,?final?byte[]?value) ????????throws?Exception ????{ ????????return?getDBI().withHandle(new?HandleCallback() ????????{ ????????????@Override ????????????public?Void?withHandle(Handle?handle)?throws?Exception ????????????{ ????????????????handle.createStatement(StringUtils.format( ????????????????????????"INSERT?INTO?%1$s?(%2$s,?%3$s)?VALUES?(:key,?:value)?ON?DUPLICATE?KEY?UPDATE?%3$s?=?:value", ????????????????????????tableName, ????????????????????????keyColumn, ????????????????????????valueColumn)) ????????????????????????.bind("key",?key) ????????????????????????.bind("value",?value) ????????????????????????.execute(); ????????????????return?null; ????????????} ????????}); ????} ???? ????@Override ????public?DBI?getDBI() ????{ ????????return?dbi; ????} ???? }

上面代碼中有幾個關鍵地方解釋下：

PAYLOAD_TYPE?=?"BLOB"?；//??高斯下CLOB方式不行，必須用BLOB SERIAL_TYPE?=?"Integer?AUTO_INCREMENT";??//?主鍵自增方式 QUOTE_STRING?=?"?";??//?高斯對空字符串的處理必須轉換為空格 DEFAULT_STREAMING_RESULT_SIZE?=?100;??//?高斯中必須不能為0

另外高斯中對查詢數據表是否存在時需要使用用戶名參數，可以從連接對象中獲取：

-

Java 代碼

String?owner?=?handle.getConnection().getMetaData().getUserName();

2、解決高斯Do not support addBatch問題

druid-server工程中SQLMetadataConnector類：

public?void?createTable(final?String?tableName,?final?Iterable?sql) ??{ ????try?{ ??????retryWithHandle( ??????????new?HandleCallback() ??????????{ ????????????@Override ????????????public?Void?withHandle(Handle?handle) ????????????{ ???????????? if?(!tableExists(handle,?tableName))?{ ????????????????????log.info("Creating?table[%s]",?tableName); //??????????????????final?Batch?batch?=?handle.createBatch(); ??????????????????for?(String?s?:?sql)?{ //??????????????????????log.info(s); //????????????????????batch.add(s); ??????????????????????handle.execute(s); ??????????????????} //??????????????????batch.execute(); ??????????????????}?else?{ ????????????????????log.info("Table[%s]?already?exists",?tableName); ??????????????????} ??????????????????return?null; ????????????} ??????????} ??????); ????}

3、解決字段名與保留關鍵字沖突問題：

druid-server工程中SQLMetadataConnector類：

將sql語句中所有start改為starttime，end改為endtime

public?void?createPendingSegmentsTable(final?String?tableName) ??{ ????createTable( ????????tableName, ????????ImmutableList.of( ????????????StringUtils.format( ????????????????"CREATE?TABLE?%1$s?(\n" ????????????????+?"??id?VARCHAR(255)?NOT?NULL,\n" ????????????????+?"??dataSource?VARCHAR(255)?NOT?NULL,\n" ????????????????+?"??created_date?VARCHAR(255)?NOT?NULL,\n" ????????????????+?"??starttime?VARCHAR(255)?NOT?NULL,\n" ????????????????+?"??endtime?VARCHAR(255)?NOT?NULL,\n" ????????????????+?"??sequence_name?VARCHAR(255)?NOT?NULL,\n" ????????????????+?"??sequence_prev_id?VARCHAR(255),\n" ????????????????+?"??sequence_name_prev_id_sha1?VARCHAR(255)?NOT?NULL,\n" ????????????????+?"??payload?%2$s?NOT?NULL,\n" ????????????????+?"??PRIMARY?KEY?(id),\n" ????????????????+?"??UNIQUE?(sequence_name_prev_id_sha1)\n" ????????????????+?")", //????????????????tableName,?getPayloadType(),?getQuoteString() ????????????????tableName,?getPayloadType() ????????????) ????????) ????); ??}

druid-server工程中IndexerSQLMetadataStorageCoordinator類：

start和end字符串與高斯關鍵字沖突，將sql語句中所有start改為starttime，end改為endtime

4、sequence_prev_id改為可以為空。

-

Java 代碼

final?String?previousSegmentIdNotNull?=?previousSegmentId?==?null???""?:?previousSegmentId;

此處的previousSegmentIdNotNull會作為sequence_prev_id插入，但高斯不支持null或“”字符串的插入，所以sequence_prev_id字段要設置為可以為空。

5、字段名大小寫問題：

druid-server工程中SQLMetadataSegmentManager類396行：

stringObjectMap中的key轉為大寫再取值，因為高斯數據庫查詢返回的字段名是大寫的。

云數據庫 GaussDB(for openGauss)

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。