---
url: /paimon/quickstart-flink.md
description: 本文为大家讲解如何使用Flink完成Paimon官方的入门案例，建议大家收藏（对英文文档有恐惧感）。
---

# Flink尝鲜Apache Paimon入门案例

本文为大家讲解如何使用Flink完成Paimon官方的入门案例，建议大家收藏（对英文文档有恐惧感）。

本文会用到Flink环境，还不清楚如何部署Flink的同学可以查看：

《基于Flink CDC实现Mysql实时同步到Doris系列教程一：Flink环境的部署》

本文演示使用的Ubuntu环境。

## 下载并解压Flink

下载最新版本的Flink

```sh
wget https://dlcdn.apache.org/flink/flink-1.19.1/flink-1.19.1-bin-scala_2.12.tgz
```

![](/img/0e6ed421-9a61-4b55-9b0d-b4f40525dfbe.png)

解压后会得到`flink-1.19.1`文件夹，并cd进入该目录。后续所有的操作都是在该目录下执行。

```sh
tar -xzf flink-1.19.1-bin-scala_2.12.tgz

cd flink-1.19.1
```

![](/img/d7e5c476-769f-41cd-b3a7-aa5487e41569.png)

## 下载Paimon的jar包

使用Paimon需要用到`paimon-flink.jar`和`flink-shaded-hadoop-2-uber-2.8.3-10.0.jar`jar包，下载与Flink相对应的版本，并复制到`flink-1.19.1/lib`目录下。

```sh
wget https://repository.apache.org/content/groups/snapshots/org/apache/paimon/paimon-flink-1.19/0.9-SNAPSHOT/paimon-flink-1.19-0.9-20240628.002224-23.jar
 
mv paimon-flink-1.19-0.9-20240628.002224-23.jar lib/

wget https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.8.3-10.0/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar

mv flink-shaded-hadoop-2-uber-2.8.3-10.0.jar lib/
```

![](/img/47b911a7-ba34-4ced-ba13-9485e9f8e92f.png)

![](/img/e98b4b03-adee-4424-a127-f3e1c39a6be5.png)

## 启动Flink

在启动Flink之前，一定要注意，需要修改一个配置，不然后续会报错（非常打击士气）。

在`flink-1.19.1/conf/conf.yaml`中找到`taskmanager`，修改`numberOfTaskSlots`配置为2，numberOfTaskSlots表示可以同时运行的任务数，也可以修改的更大一点。

![](/img/d000afd3-6428-481e-9418-f989d7c41c42.png)

![](/img/a33a585c-89c7-45e9-9c9e-9ee2dcfe6372.png)

启动Flink环境

```sh
./bin/start-cluster.sh
```

![](/img/61a9fd4a-24b9-4277-a1d2-ba61bc4a6d48.png)

访问Flink Web UI`http://localhost:8081`。

启动Flink Sql客户端

```sh
./bin/sql-client.sh
```

![](/img/32741f0b-b775-4c40-a97f-82fe83a14889.png)

![](/img/d637bcf5-b6a4-4e33-ba18-375c8f2116b9.png)

出现上图则代表成功了。

## 创建Catalog和表

```sql
create catalog my_catalog with (
 'type'='paimon',
 'warehouse'='file:/home/liuyq/paimon'
);
```

注意，此处`file:/home/liuyq/paimon`为本地的一个绝对路径，会自动创建`paimon`文件夹。

![](/img/c30ad4e4-fc8c-42a9-9143-2f9bb0915343.png)

可以看到，paimon文件夹下会有`default.db`，代表我们创建的my\_catalog下有一个默认的default数据库。

切换到我们要使用的my\_catalog下

```sql
use catalog my_catalog;
```

![](/img/97cb4ce1-7d5d-4f68-904d-35d1e6661c56.png)

创建`word_count`表

```sql
create table word_count (
 word string primary key not enforced,
 cnt bigint
);
```

![](/img/9cc8f535-6e07-48a3-89c8-8f0e0786b728.png)

此时，我们观察一下文件夹，发现多出来word\_count文件夹，对应的正好就是我们刚刚创建好的表。

我们再看一下文件夹内的文件

![](/img/c128dabc-1769-4902-8206-ddf0d8591e1c.png)

## 写入数据

创建一个临时表，用于模拟生成单词

```sql
create temporary table word_table (
 word string
 ) with (
 'connector' = 'datagen',
 'fields.word.length' = '1'
);
```

`datagen`连接器用于模拟生成数据，该表中只有一个字符串字段`word`，并且使用`fields.word.length`指定生成的数据长度为1，注意此处1要带引号，不然会报错。生成的值如：`a、b、c、1、2`等。

设置`checkpoint`时长并写入数据

```sql
set 'execution.checkpointing.interval' = '10 s';

insert into word_count select word, count(*) from word_table group by word;
```

![](/img/20ba188e-d78c-403e-b14d-eb126f93a6f2.png)

`insert into`操作相当于给flink中提交了一个实时计算的任务。该任务会把10s内`word_table`新产生的word数据聚合计算个数后把结果存入到`word_count`表中。

![](/img/8ae18f40-f994-48a1-a284-dea16232cf8e.png)

## OLAP查询

设置查询结果的样式为`tableau`

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

![](/img/9d614c88-dd7c-4c88-824d-a1ccd7c6539a.png)

从流处理模式切换到批处理模式

```sql
reset 'execution.checkpointing.interval';

set 'execution.runtime-mode' = 'batch';
```

![](/img/978f91b4-47a4-44eb-a945-712fb871466b.png)

查询`word_count`表中的数据

```sql
select * from word_count;
```

![](/img/f691c538-2a72-4ae1-afd2-6f30fa8f48da.png)

如何你卡在这里没有任何反应，说明你忘记修改`numberOfTaskSlots`配置了，只需要退出后修改，再重复上面的操作即可。

![](/img/e0c4c456-8788-435c-9d1a-350a8b93b32e.png)

批处理模式下的一次查询，也相当于给Flink中提交了一个任务，只不过该任务执行一次完成后就结束了。

## 流式查询

切换到流模式下

```sql
set 'execution.runtime-mode' = 'streaming';
```

![](/img/d4b6e108-49bd-41a5-b692-248206b556e8.png)

实时查询

```sql
select `interval`, count(*) as interval_cnt from (select cnt / 10000 as `interval` from word_count) group by `interval`;
```

![](/img/1bb887ea-3721-438c-9eef-c2f16ef06c33.png)

![](/img/ede7ab05-8e27-43ed-a2ec-010ec02a22c4.png)

拉到最下面，仔细观察，你会发现数据每个大约10s会有变动。

此时，也相当于在Flink中提交了一个任务

![](/img/d63dc010-aefa-4915-b239-1de3a9dd8f38.png)

在界面上取消任务，就会看到刚才的实时查询也中断了。

![](/img/3a53efd3-55e5-4453-bb3a-a34580ffe0f0.png)

## 退出

退出SQL客户端

```sql
exit;
```

![](/img/fdb56e87-f828-4954-8588-40a89a62662c.png)

停止Flink环境

```sh
./bin/stop-cluster.sh
```

![](/img/ab580f74-e549-4bd6-8cf4-24e5cac8597e.png)

至此，一个我们完整体验了在Flink中使用Paimon。

## 进阶

现在，如果问你Paimon是什么？你心中会有怎样的答案呢。我们再看一下`/paimon/default.db/word_count`下面有什么

![](/img/a065562f-96ea-4396-92d7-51c8adbc173a.png)

这时我们是不是可以说Paimon是具有严格结构的一组文件夹和文件的组合，这些文件会描述出数据库、数据表、数据等，用这种方式表达数据库也正是数据湖。

有一点点绕，希望大家可以体会到。想要学习更多Paimon相关的内容，可以关注：遇码，回复**Paimon**获取官方文档。