在學習 paimon sink 的過程中本來只想快速梳理下 paimon 的 sink 時對 DataStream 操作的拓撲, 但是過程中發(fā)現(xiàn) paimon 會有很多概念，并且這些概念都做了很好的抽象，一口吃不了大胖子，慢慢的邊啃邊理解吧。這篇記錄在學習過程中的 paimon table，包含 paimon table 的創(chuàng)建和 table 本身。

paimon table 創(chuàng)建

在 sql 中 table 的創(chuàng)建少不了 catalog, catalog 的創(chuàng)建又離不開 flink 的 CatalogFactory。

Paimon 對于 flink CatalogFactory 的實現(xiàn)

org.apache.flink.table.factories.CatalogFactory

Paimon 對于 flink CatalogFactory 實現(xiàn)類的不同點

IDENTIFIER 不同

org.apache.paimon.flink.FlinkGenericCatalogFactory：IDENTIFIER = "paimon"
```
CREATE CATALOG catalogName WITH ( 'type'='paimon', ....)
```
org.apache.paimon.flink.FlinkCatalogFactory: IDENTIFIER = "paimon-generic";
```
CREATE CATALOG catalogName WITH ( 'type'='paimon-generic', ....)
```

核心方法 CatalogFactory#createCatalog 的實現(xiàn)不一樣

FlinkGenericCatalogFactory#createCatalog
1. 創(chuàng)建 flink hive connector 的 HiveCatalogFactory
2. 然后創(chuàng)建 flink hive connector 的 HiveCatalog
3. new paimon 的 FlinkCatalog
4. 返回 new FlinkGenericCatalog (paimonCatalog, hiveCatalog)
FlinkCatalogFactory#createCatalog 直接 new paimon 的 FlinkCatalog
所以 FlinkGenericCatalogFactory 創(chuàng)建出來的是 FlinkGenericCatalog
FlinkCatalogFactory 創(chuàng)建出來的是 FlinkCatalog
FlinkGenericCatalog 和 FlinkCatalog 都實現(xiàn)了 flink Catalog, 那么他們的區(qū)別是什么呢？

Paimon 對于 Flink Catalog 的實現(xiàn)

org.apache.flink.table.catalog.Catalog

從剛剛 createCatalog 方法中可以看到他們區(qū)別是 FlinkGenericCatalog 不僅僅含有 paimon 的 FlinkCatalog 還包含 flink hive connector 的 HiveCatalog，從 FlinkGenericCatalog 的實現(xiàn)來看，很多操作都會同時操作兩個 catalog, 其中 HiveCatalog 是對 hive HMS 進行請求操作，F(xiàn)linkCatalog 是對 paimon 進行操作，方法例舉如下。

public CatalogBaseTable getTable(ObjectPath tablePath) {
     try {
         return paimon.getTable(tablePath); //  paimon 的 FlinkCatalog
     } catch (TableNotExistException e) {
         return flink.getTable(tablePath);  // hive connector 的 HiveCatalog
     }
}

public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists) {
     String connector = table.getOptions().get(CONNECTOR.key());
     if (FlinkCatalogFactory.IDENTIFIER.equals(connector)) { //是 paimon 表
         paimon.createTable(tablePath, table, ignoreIfExists); // 就用 paimon 的 FlinkCatalog 進行操作
     } else {
         flink.createTable(tablePath, table, ignoreIfExists); // 否則就用  hive connector 的 HiveCatalog
     }
}

public List<String> listTables(String databaseName) {
     // flink list tables contains all paimon tables // 都包含？那他是怎么把 paimon 表同步到 HMS 的
     return flink.listTables(databaseName); // 為什么這里只用  hive connector 的 HiveCatalog 就行？
}

所以看起來 FlinkGenericCatalog 有如下特點

是可以自動識別是否為 paimon 表，優(yōu)先用 paimon catalog 去嘗試。
這個 catalog 可以兼容普通的 hive 表和 paimon 表

對于 listTables 等一些操作為什么只要用 HiveCatalog 就行？
既然都包含，那么 paimon 的 FlinkCatalog 是怎么把表同步到 HMS 的？

Paimon 的 FlinkCatalog

上面得知 FlinkCatalog 是 flink Catalog 的一個實現(xiàn)

FlinkCatalog extends org.apache.flink.table.catalog.AbstractCatalog {
  private final org.apache.paimon.catalog.Catalog catalog;
  public List<String> listTables(String databaseName) {
      return catalog.listTables(databaseName);
    } 
 ... ...
}

從上得知 FlinkCatalog 其實只是一個包裝真正進行 Catalog 操作的還是 org.apache.paimon.catalog.Catalog 需要注意它的包名不是 flink 的 org.apache.flink.table.catalog.Catalog
文中我有意的補充包路徑或者強調是 paimon 的什么什么或者是 flink 的什么什么就是想要做區(qū)分，在 Paimon 中好多類名看起來很容易誤以為是 flink 的類，其實不是。。。。
所以要看 Paimon 的 org.apache.paimon.flink.FlinkCatalog 還得看 paimon 的 org.apache.paimon.catalog.Catalog

Paimon 的 Catalog

org.apache.paimon.catalog.Catalog

Paimon 的 Catalog 是如何創(chuàng)建的

直接說結論吧還是通過 flink SPI 機制根據(jù)配置、IDENTIFIER 和 org.apache.paimon.catalog.CatalogFactory 實現(xiàn)類去先找到 factory 然后在 factory 進行創(chuàng)建 paimon org.apache.paimon.catalog.Catalog 的具體實現(xiàn)
看一眼 org.apache.paimon.catalog.CatalogFactory 類圖

org.apache.paimon.catalog.CatalogFactory

Paimon Catalog 創(chuàng)建示例

// 創(chuàng)建 org.apache.paimon.catalog.FileSystemCatalog
CREATE CATALOG my_catalog WITH (
    'type' = 'paimon',
    'warehouse' = 'hdfs:///path/to/warehouse'
);

// 創(chuàng)建 org.apache.paimon.hive.HiveCatalog
CREATE CATALOG my_hive WITH (
'type' = 'paimon',
'metastore' = 'hive',
-- 'uri' = 'thrift://<hive-metastore-host-name>:<port>', default use 'hive.metastore.uris' in HiveConf
-- 'hive-conf-dir' = '...', this is recommended in the kerberos environment
-- 'hadoop-conf-dir' = '...', this is recommended in the kerberos environment
-- 'warehouse' = 'hdfs:///path/to/warehouse', default use 'hive.metastore.warehouse.dir' in HiveConf
);

// 創(chuàng)建 org.apache.paimon.jdbc.JdbcCatalog
CREATE CATALOG my_jdbc WITH (
'type' = 'paimon',
'metastore' = 'jdbc',
'uri' = 'jdbc:mysql://<host>:<port>/<databaseName>',
'jdbc.user' = '...', 
'jdbc.password' = '...', 
'catalog-key'='jdbc',
'warehouse' = 'hdfs:///path/to/warehouse'
);

可能大部分場景用的是 org.apache.paimon.hive.HiveCatalog 所以來看看這個的實現(xiàn)，剛好也可以解答 paimon 的 FlinkGenericCatalog 是怎么把表同步到 HMS 的，在創(chuàng)建 FlinkGenericCatalog 時會創(chuàng)建 Paimon 的 org.apache.paimon.catalog.Catalog 會注入核心參數(shù) 'metastore'='hive', 使得 FlinkGenericCatalog 中的 Paimon Catalog 最終是 org.apache.paimon.hive.HiveCatalog
```
options.set(CatalogOptions.METASTORE, "hive");
org.apache.paimon.catalog.CatalogFactory.createCatalog(
                            CatalogContext.create(options, new FlinkFileIOLoader()), cl),
```

Paimon 的 org.apache.paimon.hive.HiveCatalog

HiveCatalog 維護了一個 hive client 會對表的變更進行同步

HiveCatalog extends AbstractCatalog implements Catalog{
      private final IMetaStoreClient client;  // hive client
      protected void alterTableImpl(Identifier identifier, List<SchemaChange> changes) {
        final SchemaManager schemaManager = schemaManager(identifier);
        // first commit changes to underlying files
        TableSchema schema = schemaManager.commitChanges(changes);
        try {
            // sync to hive hms 表變更同步到 hive
            Table table = client.getTable(identifier.getDatabaseName(), identifier.getObjectName());
            updateHmsTablePars(table, schema);
            updateHmsTable(table, identifier, schema);
            client.alter_table(
                    identifier.getDatabaseName(), identifier.getObjectName(), table, true);
        } catch (Exception te) {
            schemaManager.deleteSchema(schema.id());
            throw new RuntimeException(te);
        }
    }
}

到這里就說明了 paimon table 的創(chuàng)建前揍，從 Flink 的 CatalogFactory 到 Flink 的 Catalog, 再到 Paimon 基于 Flink 的 Catalog, Paimon 基于 Flink 的 Catalog 是一個殼子實際是用的 Paimon 的 Catalog, Paimon 的 Catalog 又是通過 Paimon 的 CatalogFactory 創(chuàng)建而來。接下來看看主角 Paimon 的 Catalog 創(chuàng)建的 Paimon Table, Table 的創(chuàng)建是 org.apache.paimon.catalog.AbstractCatalog 實現(xiàn)的

org.apache.paimon.catalog.AbstractCatalog implements Catalog  {
  public Table getTable(Identifier identifier) throws TableNotExistException {
        if (isSystemDatabase(identifier.getDatabaseName())) { // 先忽略
        } else if (isSpecifiedSystemTable(identifier)) { //先忽略
        } else {
            return getDataTable(identifier);
        }
    }
}

private FileStoreTable getDataTable(Identifier identifier) throws TableNotExistException {
        TableSchema tableSchema = getDataTableSchema(identifier);
        return FileStoreTableFactory.create(
                fileIO,
                getDataTableLocation(identifier),
                tableSchema,
                new CatalogEnvironment(
                        Lock.factory(
                                lockFactory().orElse(null), lockContext().orElse(null), identifier),
                        metastoreClientFactory(identifier).orElse(null),
                        lineageMetaFactory));
    }

然后在 FileStoreTableFactory.create 方法中根據(jù)是否有主鍵會創(chuàng)建 AppendOnlyFileStoreTable 或者 PrimaryKeyFileStoreTable

    public static FileStoreTable create(
         FileIO fileIO, // 這又是個啥玩意？
         Path tablePath,
         TableSchema tableSchema,
         Options dynamicOptions,
         CatalogEnvironment catalogEnvironment) {
     FileStoreTable table =
             tableSchema.primaryKeys().isEmpty()
                     ? new AppendOnlyFileStoreTable(
                             fileIO, tablePath, tableSchema, catalogEnvironment)
                     : new PrimaryKeyFileStoreTable(
                             fileIO, tablePath, tableSchema, catalogEnvironment);
     return table.copy(dynamicOptions.toMap());
 }

到這里 Paimon Table 就已經(jīng)創(chuàng)建完成了，Table 提供了表的讀取寫入和表操作的一些抽象，涉及面較多
簡單看看 Table 和寫入相關的一些方法混個眼熟，后面了解更多再補充

Paimon Table 之 FileStoreTable

org.apache.paimon.table.Table

看到這個類圖一開始是崩潰了沒想到一個 Table 有這么多花樣
慶幸的是在 paimon sink 中只需要關注 FileStoreTable 的3個實現(xiàn)類 AbstractFileStoreTable、 AppendOnlyFileStoreTable 和 PrimaryKeyFileStoreTable，其中 AbstractFileStoreTable 是其他兩個的父類

org.apache.paimon.table.FileStoreTable

abstract class  AbstractFileStoreTable AbstractFileStoreTable implements FileStoreTable {
    protected final FileIO fileIO;
    @Override 
    public BucketMode bucketMode() { // 分桶模式很重要
        return store().bucketMode(); // store() 方法在子類實現(xiàn)
    }
  ... ...
}

在創(chuàng)建 AbstractFileStoreTable 時需要傳入一個 FileIO FileIO 是個啥先混個眼熟，看類圖是和數(shù)據(jù)存儲層交互的一個接口對不同的存儲有不同實現(xiàn)

org.apache.paimon.fs.FileIO
store() 的實現(xiàn)
- AppendOnlyFileStoreTable FileStore 為 AppendOnlyFileStore<InternalRow>
- PrimaryKeyFileStoreTable FileStore 為 KeyValueFileStore<KeyValue>
- FileStore 是數(shù)據(jù)讀寫的接口
  
  org.apache.paimon.FileStore

bucketMode 的實現(xiàn)

BucketMode 來自 FileStoreTable 的 FileStore，所以 PrimaryKeyFileStoreTable 的 BucketMode 是在 KeyValueFileStore 中定義的邏輯為：

先計算 crossPartitionUpdate

   public boolean crossPartitionUpdate() {
     if (primaryKeys.isEmpty() || partitionKeys.isEmpty()) {
        return false; //如果 primaryKeys 為空 或者  partitionKeys 返回  false
      }
     //如果 primaryKeys 包含所有的 partitionKeys 返回 false ; 如果 主鍵是 ABC, 分區(qū)字段是 A 則不支持分區(qū)變更，
     // 這個很好理解，因為如果分區(qū)變更了 那說明主鍵都變了變成新的記錄了
     // 那如果主鍵是 ABC 分區(qū)字段是 AD 呢？A 變了如何支持分區(qū)變更？？
     return !primaryKeys.containsAll(partitionKeys);
  }

crossPartitionUpdate = true 的場景就是要有主鍵要有分區(qū)并且分區(qū)鍵不全是主鍵
如果主鍵是 ABC 分區(qū)字段是 AD 呢？A 變了如何支持分區(qū)變更？？

如果 bucket =-1 默認為 -1 則看 crossPartitionUpdate 如果可以 crossPartitionUpdate 則為 GLOBAL_DYNAMIC 否則為 DYNAMIC
如果 bucket !=-1 則為 FIXED 并且此時 crossPartitionUpdate 一點要為 false 會校驗。這意味著勁量不要去設置 bucket 數(shù)目？？因為設置了如果 crossPartitionUpdate 為 true 就會報錯，為啥要這樣設計？？
```
   public BucketMode bucketMode() {
        if (options.bucket() == -1) { // 默認為 -1
          return crossPartitionUpdate ? BucketMode.GLOBAL_DYNAMIC : BucketMode.DYNAMIC;
       } else {
          checkArgument(!crossPartitionUpdate);
         return BucketMode.FIXED;
     }
}
```

對于 AppendOnlyFileStoreTable 的 BucketMode 是在 AppendOnlyFileStore 中定義的邏輯為：options.bucket() == -1 ? BucketMode.UNAWARE : BucketMode.FIXED
所以 AppendOnlyFileStoreTable 的 BucketMode 為 FIXED 或者 UNAWARE
所以 PrimaryKeyFileStoreTable 的 BucketMode 為 DYNAMIC 或者 GLOBAL_DYNAMIC 或者 FIX

BucketMode 翻譯自官網(wǎng)

Bucket 是讀寫的最小存儲單元，每個Bucket目錄中包含一棵LSM樹。

Fixed Bucket 指的是 BucketMode.FIXED

配置一個大于0的bucket，采用Fixed Bucket模式，根據(jù)Math.abs(key_hashcode % numBuckets)記錄計算bucket。重新縮放存儲桶只能通過離線流程完成，請參閱重新縮放存儲桶。桶數(shù)過多會導致小文件過多，桶數(shù)過少會導致寫入性能較差
根據(jù) bucket 鍵將數(shù)據(jù)分發(fā)到相應的 Bucket 中（默認為主鍵），對于帶有分桶鍵的查詢可以進行桶的 data skipping

Dynamic Bucket 有兩種

配置'bucket' = '-1'。以前寫入過的 key 會落入舊的 bucket，新的 key 會落入新的 bucket，bucket 和 key 的分布取決于數(shù)據(jù)到達的順序。 Paimon 維護一個索引來確定哪個鍵對應哪個桶。
分桶鍵的查詢不支持桶的 data skipping
Paimon會自動擴大桶的數(shù)量。
- Option1: 'dynamic-bucket.target-row-num'：控制一個桶的目標行數(shù)。
- Option2: 'dynamic-bucket.initial-buckets': 控制初始化bucket的數(shù)量。

動態(tài)Bucket僅支持單個寫入作業(yè)。請不要啟動多個作業(yè)來寫入同一分區(qū)（這可能會導致重復數(shù)據(jù)）。即使您啟用'write-only'并啟動專用的壓縮作業(yè)，它也不會起作用。
Dynamic bucket mode can not work with log system

Normal Dynamic Bucket Mode 指的是 BucketMode.DYNAMIC

當您的更新不跨分區(qū)（沒有分區(qū)，或者主鍵包含所有分區(qū)字段）時，BucketMode.DYNAMIC 使用 HASH 索引來維護從鍵到桶的映射，它比固定桶模式需要更多的內(nèi)存。
性能：

一般來說，沒有性能損失，但會有一些額外的內(nèi)存消耗，一個分區(qū)中的1 億個 條目多占用1 GB內(nèi)存，不再活動的分區(qū)不占用內(nèi)存。
對于更新率較低的表，建議使用此模式，以顯著提高性能。

Normal Dynamic Bucket Mode支持sort-compact以加快查詢速度。請參閱緊湊排序。

Cross Partitions Upsert Dynamic Bucket Mode 指的是 BucketMode.GLOBAL_DYNAMIC

當需要跨分區(qū)upsert（主鍵不包含所有分區(qū)字段）時，Dynamic Bucket 模式直接維護鍵到分區(qū)和桶的映射，使用本地磁盤，并在啟動流寫作業(yè)時通過讀取表中所有現(xiàn)有鍵來初始化索引。不同的合并引擎有不同的行為：
- Deduplicate：刪除舊分區(qū)中的數(shù)據(jù)，并將新數(shù)據(jù)插入到新分區(qū)中。保證數(shù)據(jù)的唯一性
- PartialUpdate & Aggregation：將新數(shù)據(jù)插入舊分區(qū)。
- FirstRow：如果有舊值，則忽略新數(shù)據(jù)。
  性能：
1. 對于數(shù)據(jù)量較大的表，性能會有明顯的損失。而且，初始化需要很長時間。
2. 如果你的upsert不依賴太舊的數(shù)據(jù)，可以考慮配置索引TTL來減少索引和初始化時間：
  2.1 'cross-partition-upsert.index-ttl'：rocksdb索引和初始化中的TTL，這樣可以避免維護太多索引而導致性能越來越差。但請注意，這也可能會導致數(shù)據(jù)重復。

FINAL

講述了 Paimon Table 的創(chuàng)建過程。

Paimon Table 的創(chuàng)建
Paimon Table 功能待補充。
介紹了 BucketMode。這個很重要不同的數(shù)據(jù)場景選擇合適的 mode, 不同的 mode 數(shù)據(jù)入 paimon 的拓撲也不一樣
FileStore 功能待補充。
Paimon sink 拓撲是在 DynamicTableSink 中定義的
下篇會介紹 Paimon 的 DynamicTableSink

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九欧美,1769亚洲,黄色成人av

paimon sink 源碼之 paimon table 創(chuàng)建

paimon sink 源碼之 paimon table 創(chuàng)建

paimon table 創(chuàng)建

Paimon 對于 flink CatalogFactory 的實現(xiàn)

Paimon 對于 flink CatalogFactory 實現(xiàn)類的不同點

Paimon 對于 Flink Catalog 的實現(xiàn)

Paimon 的 FlinkCatalog

Paimon 的 Catalog

Paimon 的 Catalog 是如何創(chuàng)建的

Paimon 的 org.apache.paimon.hive.HiveCatalog

Paimon Table 之 FileStoreTable

BucketMode 翻譯自官網(wǎng)

Fixed Bucket 指的是 BucketMode.FIXED

Dynamic Bucket 有兩種

Normal Dynamic Bucket Mode 指的是 BucketMode.DYNAMIC

Cross Partitions Upsert Dynamic Bucket Mode 指的是 BucketMode.GLOBAL_DYNAMIC

FINAL

相關閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九 欧美,1769亚洲,黄色成人av

paimon sink 源碼 之 paimon table 創(chuàng)建

paimon table 創(chuàng)建

Paimon 對于 flink CatalogFactory 的實現(xiàn)

Paimon 對于 flink CatalogFactory 實現(xiàn)類的不同點

Paimon 對于 Flink Catalog 的實現(xiàn)

Paimon 的 FlinkCatalog

Paimon 的 Catalog

Paimon 的 Catalog 是如何創(chuàng)建的

Paimon 的 org.apache.paimon.hive.HiveCatalog

Paimon Table 之 FileStoreTable

BucketMode 翻譯自官網(wǎng)

Fixed Bucket 指的是 BucketMode.FIXED

Dynamic Bucket 有兩種

Normal Dynamic Bucket Mode 指的是 BucketMode.DYNAMIC

Cross Partitions Upsert Dynamic Bucket Mode 指的是 BucketMode.GLOBAL_DYNAMIC

FINAL

相關閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九欧美,1769亚洲,黄色成人av

paimon sink 源碼之 paimon table 創(chuàng)建