文章目录

概述
- 定义
- 核心功能
- 适用场景
- 架构原理
- - 总体架构
  - 统一存储
  - 基本概念
  - 文件布局
部署
- 环境准备
- 环境部署
实战
- Catalog
- - 文件系统
  - Hive Catalog
- 创建表
- - 创建Catalog管理表
  - 查询创建表（CTAS）
  - 创建外部表
  - 创建临时表
- 修改表
- - 修改表
  - 修改列
  - 修改水印

概述

定义

Apache Paimon 官网 https://paimon.apache.org/ 最新稳定版本为0.4.0-incubating，0.5-SNAPSHOT正在开发

Apache Paimon 文档地址 https://paimon.apache.org/docs/master/

Apache Paimon 源码地址 https://github.com/apache/incubator-paimon/

Apache Paimon (incubating) 目前属于Apache 软件基金会 (ASF) 的孵化项目，其原项目为由Flink官方维护的Flink Table Store；其设计为一个开源流数据湖平台，包揽Streaming实时计算能力和LakeHouse架构优势，统一了存储，具有高速数据摄取，变更日志跟踪和高效的实时分析强大能力。

Apache Paimon 采用开放的数据格式和技术理念，不仅支持Flink SQL编写和本地查询，还可以与其他诸多业界主流计算引擎进行对接。

核心功能

统一批处理和流处理：Paimon支持批写和批读，以及流式写更改和流式读表更改日志。
数据湖：Paimon具有成本低、可靠性高、元数据可扩展等优点，具有数据湖存储的所有优势。
合并引擎：Paimon支持丰富的合并引擎。缺省情况下，保留主键的最后一项记录，可以“部分更新”或“聚合”。
变更日志生成：Paimon支持丰富的Changelog producer例如“lookup”和“full-compaction”，可以从任何数据源生成正确且完整的变更日志从而简化流管道的构建。
丰富的表类型：除了主键表，Paimon还支持append-only只追加表，自动压缩小文件，并提供有序的流读取来替换消息队列。
模式演化：支持完整的模式演化，例如可以重新命名列和重新排序。

适用场景

Apache Paimon 适用于需要在流数据上进行实时查询和分析的场景。它可以帮助用户更容易地构建流式数据湖，实现高吞吐、低延迟的数据摄入、流式订阅以及实时查询能力。例如在金融、电子商务、物联网等行业中，可以使用 Apache Paimon 来实现实时推荐、欺诈检测、异常检测等应用。

架构原理

总体架构

Paimon 创新的结合了湖存储 + LSM + 列式格式 (ORC, Parquet)，为湖存储带来大规模实时更新能力

读/写：Paimon支持多种方式来读/写数据和执行OLAP查询。
- 对于读取，它支持从历史快照(批处理模式)、从最新偏移量(流模式)中读取数据，或者以混合方式读取增量快照。
- 对于写操作，它支持从数据库变更日志(CDC)进行流同步，或者从离线数据进行批量插入/覆盖。
生态系统：除了Apache Flink, Paimon还支持其他计算引擎的读取，如Apache Hive、Apache Spark、Presto和Trino。
内部：在底层，Paimon将列文件存储在文件系统/对象存储中，并使用LSM树结构来支持大量数据更新和高性能查询。

统一存储

对于像Apache Flink这样的流媒体引擎，通常有三种类型的连接器:

消息队列，比如Apache Kafka，它在这个管道的源和中间阶段都被使用，以保证延迟保持在秒内。
OLAP系统，如ClickHouse，它以流方式接收处理过的数据，并为用户的特别查询提供服务。
批处理存储，如Apache Hive，它支持传统批处理的各种操作，包括INSERT OVERWRITE。

Paimon提供表抽象。它的使用方式与传统数据库没有什么不同:

在批处理执行模式下，它就像一个Hive表，支持批处理SQL的各种操作。查询最新快照。
在流式执行模式下，它的作用类似于消息队列。查询它的行为类似于从历史数据永远不会过期的消息队列查询流更改日志。

基本概念

快照：快照捕获表在某个时间点的状态。用户可以通过最新的快照访问表的最新数据。通过时间旅行，用户还可以通过较早的快照访问表的先前状态。
分区
- Paimon采用与Apache Hive相同的分区概念来分离数据。
- 分区是一种可选的方法，可以根据特定列(如日期、城市和部门)的值将表划分为相关部分。每个表可以有一个或多个分区键来标识一个特定的分区。
- 通过分区，用户可以有效地操作表中的记录切片。如果定义了主键，分区键必须是主键的子集。
桶
- 未分区的表或分区表中的分区被细分为桶，为数据提供额外的结构，可用于更有效的查询。
- bucket的范围由记录中一个或多个列的哈希值决定。用户可以通过提供bucket-key选项来指定bucket列。如果没有指定桶键选项，则使用主键(如果定义了)或完整记录作为桶键。
- bucket是用于读写的最小存储单元，因此bucket的数量限制了最大的处理并行性。但是，这个数字不应该太大，因为它将导致大量小文件和低读取性能。一般情况下，建议每个bucket中的数据大小为1GB左右。
一致性保证
- Paimon编写器使用两阶段提交协议自动将一批记录提交到表中。每次提交在提交时最多产生两个快照。
- 对于任何两个同时修改一个表的写入器，只要他们没有修改同一个桶，他们的提交就是可序列化的。如果修改的是同一个桶，则只保证快照隔离。也就是说，最终的表状态可能是两次提交的混合状态，但不会丢失任何更改。

文件布局

表的所有文件都存储在一个基本目录下。Paimon文件以分层的方式组织。下图说明了文件布局。从快照文件开始，Paimon读取器可以递归地访问表中的所有记录。

快照文件：所有快照文件都保存在快照目录下。快照文件是一个JSON文件，其中包含有关该快照的信息，包括架构文件使用包含此快照的所有更改的清单列表。
Manifest文件：所有清单列表和清单文件都存储在manifest目录中。清单列表是清单文件名的列表，清单文件是包含有关LSM数据文件和更改日志文件的更改的文件。例如在相应的快照中创建了哪个LSM数据文件，删除了哪个文件。
数据文件：数据文件按分区和桶分组。每个桶目录包含一个LSM树及其变更日志文件。目前，Paimon支持使用orc(默认)、parquet和avro作为数据文件格式。
LSM树：Paimon采用LSM树(日志结构的合并树)作为文件存储的数据结构。数据文件中的记录按其主键排序；在Sorted Runs中，数据文件的主键范围从不重叠。不同Sorted Runs可能有重叠的主键范围，甚至可能包含相同的主键。在查询LSM树时，必须将所有Sorted Runs组合起来，并且必须根据用户指定的合并引擎和每条记录的时间戳合并具有相同主键的所有记录。写入LSM树的新记录将首先在内存中进行缓冲。当内存缓冲区已满时，将对内存中的所有记录进行排序并刷新到磁盘。得益于 LSM 数据结构的追加写能力，Paimon 在大规模的更新数据输入的场景中提供了出色的性能。

合并：当越来越多的记录写入LSM树时，Sorted Runs次数将会增加。因为查询LSM树需要将所有Sorted Runs组合在一起，太多的Sorted Runs将导致查询性能差，甚至导致内存不足。为了限制排序运行的次数，必须偶尔将几个Sorted Runs合并为一个大的Sorted Runs，这个过程称为合并或压缩，合并是一个资源密集型过程，它会消耗一定的CPU时间和磁盘IO，因此过于频繁的压缩可能会导致写速度变慢。这是查询性能和写性能之间的权衡。目前，Paimon采用了一种与Rocksdb的通用压实类似的压实策略。默认情况下，当Paimon将记录追加到LSM树时，它还将根据需要执行压缩，还可以选择在专用压缩作业中执行所有压缩。

部署

环境准备

官方提供对应引擎的版本支持如下

环境部署

# 这里选择下载最新版本Flink1.17.1
wget https://dlcdn.apache.org/flink/flink-1.17.1/flink-1.17.1-bin-scala_2.12.tgz
# 解压文件
tar -xvf flink-1.17.1-bin-scala_2.12.tgz
# 进入flink目录
cd flink-1.17.1

配置环境变量，修改flink-conf.yaml配置文件

# 如果是Flink17版本以下env.java.opts.all则需改为env.java.opts
env.java.opts.all: "-Dfile.encoding=UTF-8"
classloader.check-leaked-classloader: false
taskmanager.numberOfTaskSlots: 4
execution.checkpointing.interval: 10s
# 用于存储和检查点状态
state.backend: rocksdb
# 存储检查点的数据文件和元数据的默认目录
state.checkpoints.dir: hdfs://myns/flink/myns
# savepoints 的默认目标目录(可选)
state.savepoints.dir: hdfs://myns/flink/savepoints
# 用于启用/禁用增量 checkpoints 的标志
state.backend.incremental: true

使用paimon非常简单，和其他数据湖产品一样，都是将jar包放在引擎的目录下

# 解决依赖问题，将hadoop的hadoop-mapreduce-client-core-3.3.4.jar拷贝到flink
cp /opt/module/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-client-core-3.3.4.jar ./lib/
# 下载最新版的paimon，目前0.5属于快照版本，,可以先进入lib目录，然后下载到当前lib，也可以通过其他地方下载然后上传拷贝到flink的lib目录下
cd lib/
wget https://repository.apache.org/content/groups/snapshots/org/apache/paimon/paimon-flink-1.17/0.5-SNAPSHOT/paimon-flink-1.17-0.5-20230802.034234-105.jar
# 由于后续会使用到其他连接器，这里先下载安装好，后面直接使用即可，先下载flink-sql-connector-hive连接器
wget https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-hive-3.1.3_2.12/1.17.1/flink-sql-connector-hive-3.1.3_2.12-1.17.1.jar
# 下载flink-sql-connector-mysql-cdc连接器
wget https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mysql-cdc/2.4.1/flink-sql-connector-mysql-cdc-2.4.1.jar
# 下载flink-sql-connector-kafka连接器
wget https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka/1.17.1/flink-sql-connector-kafka-1.17.1.jar
# 重新回到flink的根目录
cd ../
# 先保证hadoop环境，通过yarn启动flink集群
./bin/yarn-session.sh -d
# 以提交yarn-session方式启动sql客户端
./bin/sql-client.sh -s yarn-session

测试环境是否可用

set 'sql-client.execution.result-mode' = 'tableau'; 
show databases;
show tables;
select 1;

通过yarn管理页面可以看到有 Flink session cluster运行job，点击该记录的ApplicationMaster跳转到flink管理页面，也可以看到刚才job已经完成，环境准备完毕。

实战

Catalog

Paimon Catalog可以持久化元数据，当前支持两种类型的metastore

文件系统（默认）：将元数据和表文件存储在文件系统中。
hive：在hive metastore存储元数据，用户可以直接从hive访问表。

文件系统

下面的Flink SQL注册并使用一个名为fs_catalog的Paimon编目。元数据和表文件存放在hdfs://myns/paimon/fs下。

CREATE CATALOG fs_catalog WITH (
'type' = 'paimon',
'warehouse' = 'hdfs://myns/paimon/fs'
);

show catalogs;

Hive Catalog

使用Hive Catalog前需要先启动hive元数据服务

nohup hive --service metastore &

通过使用Paimon Hive catalog，对catalog的更改将直接影响到相应的Hive metastore。使用Hive catalog，数据库名、表名和字段名应该是小写的。

CREATE CATALOG hive_catalog WITH (
    'type' = 'paimon',
    'metastore' = 'hive',
    'uri' = 'thrift://hadoop2:9083',
    'hive-conf-dir' = '/home/commons/apache-hive-3.1.3-bin/conf/', 
    'warehouse' = 'hdfs://myns/paimon/hive'
);
show catalogs;

USE CATALOG hive_catalog;
CREATE TABLE test1 (
  id BIGINT,
  a INT,
  b STRING,
  dt STRING COMMENT 'timestamp string in format yyyyMMdd',
  PRIMARY KEY(id, dt) NOT ENFORCED
) PARTITIONED BY (dt);

在指定的Catalog中创建表

关闭重新进入sql-client后，只剩下默认的default_catalog，因此可以在启动客户端时执行指定创建catalog语句，vim conf/sql-client-init.sql

set 'sql-client.execution.result-mode' = 'tableau'; 

CREATE CATALOG fs_catalog WITH (
'type' = 'paimon',
'warehouse' = 'hdfs://myns/paimon/fs'
);

CREATE CATALOG hive_catalog WITH (
    'type' = 'paimon',
    'metastore' = 'hive',
    'uri' = 'thrift://hadoop2:9083',
    'hive-conf-dir' = '/home/commons/apache-hive-3.1.3-bin/conf/', 
    'warehouse' = 'hdfs://myns/paimon/hive'
);

USE CATALOG hive_catalog;

通过-i参数启动执行sql文件，启动后就可以看到hive_catalog之前已创建的表了

./bin/sql-client.sh -s yarn-session -i conf/sql-client-init.sql

创建表

创建Catalog管理表

在Paimon Catalog中创建的表由Catalog管理也就是管理表。当表从目录中删除时，它的表文件也将被删除，与hive内部表相似。

带主键表

CREATE TABLE user_behavior1 (
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING,
    dt STRING,
    hh STRING,
    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
);

在删除表之前，应该停止在表上插入作业，否则不能完全删除表文件。

分区表

CREATE TABLE user_behavior2 (
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING,
    dt STRING,
    hh STRING,
    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
) PARTITIONED BY (dt, hh);

指定统计模式：Paimon将自动收集数据文件的统计信息，以加快查询过程。支持四种模式:
- Full：收集完整指标:null_count, min, Max。
- Truncate (length)：长度可以是任何正数，默认模式是Truncate(16)，这意味着收集null计数，min/max值，截断长度为16。这主要是为了避免太大的列，将扩大清单文件。
- 计数：只收集空计数。
- None：关闭元数据采集。
字段默认值：Paimon表目前支持为表属性中的字段设置默认值，注意不能指定分区字段和主键字段。

CREATE TABLE user_behavior2 (
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING,
    dt STRING,
    hh STRING,
    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
) PARTITIONED BY (dt, hh)
with(
    'fields.item_id.deafult-value'='0'
);

查询创建表（CTAS）

表可以通过查询结果创建或填充，简单创建表。

CREATE TABLE items1 (
    user_id BIGINT,
    item_id BIGINT
);
CREATE TABLE items2 AS SELECT * FROM items1;

/* 分区表 */
CREATE TABLE user_behavior_p1 (
     user_id BIGINT,
     item_id BIGINT,
     behavior STRING,
     dt STRING,
     hh STRING
) PARTITIONED BY (dt, hh);
CREATE TABLE user_behavior_p2 WITH ('partition' = 'dt') AS SELECT * FROM user_behavior_p1;
    
/* change options */
CREATE TABLE user_behavior_3 (
       user_id BIGINT,
       item_id BIGINT
) WITH ('file.format' = 'orc');
CREATE TABLE user_behavior_4 WITH ('file.format' = 'parquet') AS SELECT * FROM user_behavior_3;

/* 主键 */
CREATE TABLE user_behavior_5 (
      user_id BIGINT,
      item_id BIGINT,
      behavior STRING,
      dt STRING,
      hh STRING,
      PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
) ;
CREATE TABLE user_behavior_6 WITH ('primary-key' = 'dt,hh') AS SELECT * FROM user_behavior_5;

/* 主键 + 分区 */
CREATE TABLE user_behavior_all (
      user_id BIGINT,
      item_id BIGINT,
      behavior STRING,
      dt STRING,
      hh STRING,
      PRIMARY KEY (dt, hh, user_id) NOT ENFORCED 
) PARTITIONED BY (dt, hh);
CREATE TABLE user_behavior_all_as WITH ('primary-key' = 'dt,hh', 'partition' = 'dt') AS SELECT * FROM user_behavior_all;

CREATE TABLE LIKE

CREATE TABLE user_behavior (
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING,
    dt STRING,
    hh STRING,
    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
) ;

CREATE TABLE user_behavior_like LIKE user_behavior;

表属性

CREATE TABLE user_behavior (
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING,
    dt STRING,
    hh STRING,
    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
) PARTITIONED BY (dt, hh) WITH (
    'bucket' = '2',
    'bucket-key' = 'user_id'
);

创建外部表

外部表由Catalog记录，但不由Catalog管理。如果删除外部表，则不会删除其表文件。可以在任何目录中使用Paimon外部表。如果不想创建Paimon目录，而只想读/写表，那么可以考虑使用外部表。Flink SQL支持读写外部表；外部Paimon表是通过指定连接器和路径表属性创建的。

use catalog default_catalog;
CREATE TABLE user_behavior_external (
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING,
    dt STRING,
    hh STRING,
    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
) WITH (
    'connector' = 'paimon',
    'path' = 'hdfs://myns/paimon/external',
    'auto-create' = 'true' -- 如果表路径不存在，此table属性将为空表创建表文件目前仅支持Flink
);

创建临时表

临时表仅由Flink支持。与外部表一样，临时表只是记录，而不是由当前Flink SQL会话管理。如果临时表被删除，它的资源不会被删除。当Flink SQL会话关闭时，也会删除临时表。如果希望将Paimon Catalog与其他表一起使用，但又不希望将它们存储在其他Catalog中，则可以创建一个临时表。

USE CATALOG hive_catalog;
CREATE TEMPORARY TABLE temp_table (
    k INT,
    v STRING
) WITH (
    'connector' = 'filesystem',
    'path' = 'hdfs://myns/paimon/temp/temp_table.csv',
    'format' = 'csv'
);

# 可以使用临时表和其他表进行关联查询
SELECT my_table.k, my_table.v, temp_table.v FROM my_table JOIN temp_table ON my_table.k = temp_table.k;

修改表

CREATE TABLE my_table (
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING,
    dt STRING,
    hh STRING,
    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
);
# 修改表属性
ALTER TABLE my_table SET (
    'write-buffer-size' = '256 MB'
);
# 修改表名
ALTER TABLE my_table RENAME TO my_table_new;
# 删除表的属性
ALTER TABLE my_table RESET ('write-buffer-size');

修改列

# 填写列
ALTER TABLE my_table ADD (c1 INT, c2 STRING);
# 重命名列
ALTER TABLE my_table RENAME c0 TO c1;
# 删除列
ALTER TABLE my_table DROP (c1, c2);
# 更改列的空属性
CREATE TABLE my_table (id INT PRIMARY KEY NOT ENFORCED, coupon_info FLOAT NOT NULL);
-- 将列' coupon_info '从NOT NULL更改为可空
ALTER TABLE my_table MODIFY coupon_info FLOAT;
-- 将列' coupon_info '从可空改为NOT NULL如果已经有NULL值，设置如下表选项，在修改表之前静默删除这些记录。
SET 'table.exec.sink.not-null-enforcer' = 'DROP';
ALTER TABLE my_table MODIFY coupon_info FLOAT NOT NULL;
# 更改列注释
ALTER TABLE my_table MODIFY buy_count BIGINT COMMENT 'buy count'
# 添加列位置
ALTER TABLE my_table ADD c INT FIRST;
ALTER TABLE my_table ADD c INT AFTER b;
# 改变列位置
ALTER TABLE my_table MODIFY col_a DOUBLE FIRST;
ALTER TABLE my_table MODIFY col_a DOUBLE AFTER col_b;
# 修改列类型
ALTER TABLE my_table MODIFY col_a DOUBLE;

修改水印

# 添加WATERMARK，下面的SQL从现有的列log_ts中添加一个计算列ts，并在列ts上添加一个策略为ts - INTERVAL '1' HOUR的水印，该水印被标记为表my_table的事件时间属性。
CREATE TABLE my_test_wm (
    id BIGINT,
    name STRING,
    log_ts BIGINT
);
ALTER TABLE my_test_wm ADD (
    ts AS TO_TIMESTAMP_LTZ(log_ts,3),
    WATERMARK FOR ts AS ts - INTERVAL '1' HOUR
);
# 修改WATERMARK
ALTER TABLE my_test_wm MODIFY WATERMARK FOR ts AS ts - INTERVAL '2' HOUR
# 删除WATERMARK
ALTER TABLE my_test_wm DROP WATERMARK