在学习 paimon sink 的过程中本来只想快速梳理下 paimon 的 sink 时对 DataStream 操作的拓扑, 但是过程中发现 paimon 会有很多概念，并且这些概念都做了很好的抽象，一口吃不了大胖子，慢慢的边啃边理解吧。这篇记录在学习过程中的 paimon table，包含 paimon table 的创建和 table 本身。
paimon table 创建
在 sql 中 table 的创建少不了 catalog, catalog 的创建又离不开 flink 的 CatalogFactory。
Paimon 对于 flink CatalogFactory 的实现
org.apache.flink.table.factories.CatalogFactory
Paimon 对于 flink CatalogFactory 实现类的不同点
- 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 的实现不一样
- FlinkGenericCatalogFactory#createCatalog
- 创建 flink hive connector 的 HiveCatalogFactory
- 然后创建 flink hive connector 的 HiveCatalog
- new paimon 的 FlinkCatalog
- 返回 new FlinkGenericCatalog (paimonCatalog, hiveCatalog)
- FlinkCatalogFactory#createCatalog 直接 new paimon 的 FlinkCatalog
- 所以 FlinkGenericCatalogFactory 创建出来的是 FlinkGenericCatalog
FlinkCatalogFactory 创建出来的是 FlinkCatalog
FlinkGenericCatalog 和 FlinkCatalog 都实现了 flink Catalog, 那么他们的区别是什么呢？
Paimon 对于 Flink Catalog 的实现
org.apache.flink.table.catalog.Catalog
从刚刚 createCatalog 方法中可以看到他们区别是 FlinkGenericCatalog 不仅仅含有 paimon 的 FlinkCatalog 还包含 flink hive connector 的 HiveCatalog，从 FlinkGenericCatalog 的实现来看，很多操作都会同时操作两个 catalog, 其中 HiveCatalog 是对 hive HMS 进行请求操作，FlinkCatalog 是对 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 表，优先用 paimon catalog 去尝试。
- 这个 catalog 可以兼容普通的 hive 表和 paimon 表
对于 listTables 等一些操作为什么只要用 HiveCatalog 就行？
既然都包含，那么 paimon 的 FlinkCatalog 是怎么把表同步到 HMS 的？
Paimon 的 FlinkCatalog
上面得知 FlinkCatalog 是 flink Catalog 的一个实现
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 的什么什么就是想要做区分，在 Paimon 中好多类名看起来很容易误以为是 flink 的类，其实不是。。。。 - 所以要看 Paimon 的 org.apache.paimon.flink.FlinkCatalog 还得看 paimon 的 org.apache.paimon.catalog.Catalog
Paimon 的 Catalog
org.apache.paimon.catalog.Catalog
Paimon 的 Catalog 是如何创建的
- 直接说结论吧还是通过 flink SPI 机制根据配置、IDENTIFIER 和 org.apache.paimon.catalog.CatalogFactory 实现类去先找到 factory 然后在 factory 进行创建 paimon org.apache.paimon.catalog.Catalog 的具体实现
-
看一眼 org.apache.paimon.catalog.CatalogFactory 类图
org.apache.paimon.catalog.CatalogFactory
-
Paimon Catalog 创建示例
// 创建 org.apache.paimon.catalog.FileSystemCatalog CREATE CATALOG my_catalog WITH ( 'type' = 'paimon', 'warehouse' = 'hdfs:///path/to/warehouse' ); // 创建 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 ); // 创建 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 所以来看看这个的实现，刚好也可以解答 paimon 的 FlinkGenericCatalog 是怎么把表同步到 HMS 的，在 创建 FlinkGenericCatalog 时会创建 Paimon 的 org.apache.paimon.catalog.Catalog 会注入核心参数 '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 的创建前揍， 从 Flink 的 CatalogFactory 到 Flink 的 Catalog, 再到 Paimon 基于 Flink 的 Catalog, Paimon 基于 Flink 的 Catalog 是一个壳子实际是用的 Paimon 的 Catalog, Paimon 的 Catalog 又是通过 Paimon 的 CatalogFactory 创建而来。接下来看看主角 Paimon 的 Catalog 创建的 Paimon Table, Table 的创建是 org.apache.paimon.catalog.AbstractCatalog 实现的
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 方法中根据是否有主键会创建 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 就已经创建完成了，Table 提供了表的读取写入和表操作的一些抽象，涉及面较多
简单看看 Table 和写入相关的一些 方法 混个眼熟，后面了解更多再补充
Paimon Table 之 FileStoreTable
org.apache.paimon.table.Table
- 看到这个类图一开始是崩溃了没想到一个 Table 有这么多花样
-
庆幸的是在 paimon sink 中只需要关注 FileStoreTable 的3个实现类 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() 方法在子类实现
}
... ...
}
-
在创建 AbstractFileStoreTable 时需要传入一个 FileIO FileIO 是个啥先混个眼熟，看类图是和 数据存储层交互的一个接口对不同的存储有不同实现
org.apache.paimon.fs.FileIO -
store() 的实现
- AppendOnlyFileStoreTable FileStore 为 AppendOnlyFileStore<InternalRow>
- PrimaryKeyFileStoreTable FileStore 为 KeyValueFileStore<KeyValue>
-
FileStore 是数据读写的接口
org.apache.paimon.FileStore
-
bucketMode 的实现
- BucketMode 来自 FileStoreTable 的 FileStore，所以 PrimaryKeyFileStoreTable 的 BucketMode 是在 KeyValueFileStore 中定义的逻辑为：
- 先计算 crossPartitionUpdate
public boolean crossPartitionUpdate() { if (primaryKeys.isEmpty() || partitionKeys.isEmpty()) { return false; //如果 primaryKeys 为空 或者 partitionKeys 返回 false } //如果 primaryKeys 包含所有的 partitionKeys 返回 false ; 如果 主键是 ABC, 分区字段是 A 则不支持分区变更， // 这个很好理解，因为如果分区变更了 那说明主键都变了变成新的记录了 // 那如果主键是 ABC 分区字段是 AD 呢？A 变了如何支持分区变更？？ return !primaryKeys.containsAll(partitionKeys); }- crossPartitionUpdate = true 的场景就是要有主键要有分区并且分区键不全是主键
- 如果主键是 ABC 分区字段是 AD 呢？A 变了如何支持分区变更？？
- 如果 bucket =-1 默认为 -1 则看 crossPartitionUpdate 如果可以 crossPartitionUpdate 则为 GLOBAL_DYNAMIC 否则为 DYNAMIC
如果 bucket !=-1 则为 FIXED 并且此时 crossPartitionUpdate 一点要为 false 会校验。 这意味着劲量不要去设置 bucket 数目？？ 因为设置了 如果 crossPartitionUpdate 为 true 就会报错，为啥要这样设计？？public BucketMode bucketMode() { if (options.bucket() == -1) { // 默认为 -1 return crossPartitionUpdate ? BucketMode.GLOBAL_DYNAMIC : BucketMode.DYNAMIC; } else { checkArgument(!crossPartitionUpdate); return BucketMode.FIXED; } }
- 先计算 crossPartitionUpdate
- BucketMode 来自 FileStoreTable 的 FileStore，所以 PrimaryKeyFileStoreTable 的 BucketMode 是在 KeyValueFileStore 中定义的逻辑为：
对于 AppendOnlyFileStoreTable 的 BucketMode 是在 AppendOnlyFileStore 中定义的逻辑为：options.bucket() == -1 ? BucketMode.UNAWARE : BucketMode.FIXED
所以 AppendOnlyFileStoreTable 的 BucketMode 为 FIXED 或者 UNAWARE
所以 PrimaryKeyFileStoreTable 的 BucketMode 为 DYNAMIC 或者 GLOBAL_DYNAMIC 或者 FIX
BucketMode 翻译自官网
- Bucket 是读写的最小存储单元，每个Bucket目录中包含一棵LSM树。
Fixed Bucket 指的是 BucketMode.FIXED
- 配置一个大于0的bucket，采用Fixed Bucket模式，根据
Math.abs(key_hashcode % numBuckets)记录计算bucket。重新缩放存储桶只能通过离线流程完成，请参阅重新缩放存储桶。桶数过多会导致小文件过多，桶数过少会导致写入性能较差 - 根据 bucket 键将数据分发到相应的 Bucket 中（默认为主键），对于带有分桶键的查询可以进行桶的 data skipping
Dynamic Bucket 有两种
- 配置'bucket' = '-1'。以前写入过的 key 会落入旧的 bucket，新的 key 会落入新的 bucket，bucket 和 key 的分布取决于数据到达的顺序。 Paimon 维护一个索引来确定哪个键对应哪个桶。
- 分桶键的查询不支持桶的 data skipping
- Paimon会自动扩大桶的数量。
- Option1: 'dynamic-bucket.target-row-num'：控制一个桶的目标行数。
- Option2: 'dynamic-bucket.initial-buckets': 控制初始化bucket的数量。
动态Bucket仅支持单个写入作业。请不要启动多个作业来写入同一分区（这可能会导致重复数据）。即使您启用'write-only'并启动专用的压缩作业，它也不会起作用。
Dynamic bucket mode can not work with log system
Normal Dynamic Bucket Mode 指的是 BucketMode.DYNAMIC
- 当您的更新不跨分区（没有分区，或者主键包含所有分区字段）时，BucketMode.DYNAMIC 使用 HASH 索引来维护从键到桶的映射，它比固定桶模式需要更多的内存。
性能：
- 一般来说，没有性能损失，但会有一些额外的内存消耗，一个分区中的1 亿个 条目多占用1 GB内存，不再活动的分区不占用内存。
- 对于更新率较低的表，建议使用此模式，以显着提高性能。
Normal Dynamic Bucket Mode支持sort-compact以加快查询速度。请参阅紧凑排序。
Cross Partitions Upsert Dynamic Bucket Mode 指的是 BucketMode.GLOBAL_DYNAMIC
- 当需要跨分区upsert（主键不包含所有分区字段）时，Dynamic Bucket 模式直接维护键到分区和桶的映射，使用本地磁盘，并在启动流写作业时通过读取表中所有现有键来初始化索引。不同的合并引擎有不同的行为：
- Deduplicate：删除旧分区中的数据，并将新数据插入到新分区中。保证数据的唯一性
- PartialUpdate & Aggregation：将新数据插入旧分区。
- FirstRow：如果有旧值，则忽略新数据。
性能：
- 对于数据量较大的表，性能会有明显的损失。而且，初始化需要很长时间。
- 如果你的upsert不依赖太旧的数据，可以考虑配置索引TTL来减少索引和初始化时间：
2.1 'cross-partition-upsert.index-ttl'：rocksdb索引和初始化中的TTL，这样可以避免维护太多索引而导致性能越来越差。但请注意，这也可能会导致数据重复。
FINAL
-
讲述了 Paimon Table 的创建过程。
Paimon Table 的创建 - Paimon Table 功能待补充。
- 介绍了 BucketMode。 这个很重要不同的数据场景选择合适的 mode, 不同的 mode 数据入 paimon 的 拓扑也不一样
- FileStore 功能待补充。
- Paimon sink 拓扑 是在 DynamicTableSink 中定义的
- 下篇会介绍 Paimon 的 DynamicTableSink
