Hudi（三）集成Flink

 1、环境准备

        将编译好的jar包放到Flink的lib目录下。

cp hudi-flink1.13-bundle-0.12.0.jar /opt/module/flink-1.13.2/lib

2、sql-client方式

2.1、修改flink-conf.yaml配置

vim /opt/module/flink-1.13.2/conf/flink-conf.yaml
 
state.backend: rocksdb
execution.checkpointing.interval: 30000
state.checkpoints.dir: hdfs://hadoop1:9000/ckps
state.backend.incremental: true

2.2、yarn-session模式启动

1、启动

1、先启动hadoop集群，然后通过yarn-session启动flink：
/opt/module/flink-1.13.2/bin/yarn-session.sh -d
2、再启动sql-client
/opt/module/flink-1.13.2/bin/sql-client.sh embedded -s yarn-session

2、写入数据

表格模式（table mode）在内存中实体化结果，并将结果用规则的分页表格可视化展示出来。执行如下命令启用：
SET 'sql-client.execution.result-mode' = 'table'; --默认
 
变更日志模式（changelog mode）不会实体化和可视化结果，而是由插入（+）和撤销（-）组成的持续查询产生结果流。
SET 'sql-client.execution.result-mode' = 'changelog';
 
Tableau模式（tableau mode）更接近传统的数据库，会将执行的结果以制表的形式直接打在屏幕之上。具体显示的内容会取决于作业执行模式的不同(execution.type)：
SET 'sql-client.execution.result-mode' = 'tableau';
 
-- 创建hudi表
CREATE TABLE t1(
  uuid VARCHAR(20) PRIMARY KEY NOT ENFORCED,
  name VARCHAR(10),
  age INT,
  ts TIMESTAMP(3),
  `partition` VARCHAR(20)
)
PARTITIONED BY (`partition`)
WITH (
  'connector' = 'hudi',
  'path' = 'hdfs://hadoop1:9000/tmp/hudi_flink/t1',   --hudi表的基本路径
  'table.type' = 'MERGE_ON_READ'   --默认是COW
);
 
-- 插入数据
INSERT INTO t1 VALUES
  ('id1','Danny',23,TIMESTAMP '1970-01-01 00:00:01','par1'),
  ('id2','Stephen',33,TIMESTAMP '1970-01-01 00:00:02','par1'),
  ('id3','Julian',53,TIMESTAMP '1970-01-01 00:00:03','par2'),
  ('id4','Fabian',31,TIMESTAMP '1970-01-01 00:00:04','par2'),
  ('id5','Sophia',18,TIMESTAMP '1970-01-01 00:00:05','par3'),
  ('id6','Emma',20,TIMESTAMP '1970-01-01 00:00:06','par3'),
  ('id7','Bob',44,TIMESTAMP '1970-01-01 00:00:07','par4'),
  ('id8','Han',56,TIMESTAMP '1970-01-01 00:00:08','par4');

3、IDEA编码方式

3.1、环境准备

1、手动install依赖

mvn install:install-file -DgroupId=org.apache.hudi -DartifactId=hudi-flink_2.11 -Dversion=0.12.0 -Dpackaging=jar -Dfile=./hudi-flink1.13-bundle-0.12.0.jar

2、编写代码

import org.apache.flink.contrib.streaming.state.{EmbeddedRocksDBStateBackend, PredefinedOptions}
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.environment.{CheckpointConfig, StreamExecutionEnvironment}
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment
 
import java.util.concurrent.TimeUnit
 
 
 
object HudiExample {
  def main(args: Array[String]): Unit = {
 
    // val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration())
    val env = StreamExecutionEnvironment.getExecutionEnvironment
 
    // 设置状态后端RocksDB
    val embeddedRocksDBStateBackend = new EmbeddedRocksDBStateBackend(true)
 
    //  embeddedRocksDBStateBackend.setDbStoragePath("file:///E:/rocksdb")
    embeddedRocksDBStateBackend.setPredefinedOptions(PredefinedOptions.SPINNING_DISK_OPTIMIZED_HIGH_MEM)
    env.setStateBackend(embeddedRocksDBStateBackend)
    // checkpoint配置
    env.enableCheckpointing(TimeUnit.SECONDS.toMillis(10), CheckpointingMode.EXACTLY_ONCE)
    val checkpointConfig = env.getCheckpointConfig
    checkpointConfig.setCheckpointStorage("hdfs://hadoop1:9000/ckps")
    checkpointConfig.setMinPauseBetweenCheckpoints(TimeUnit.SECONDS.toMillis(10))
    checkpointConfig.setTolerableCheckpointFailureNumber(5)
    checkpointConfig.setCheckpointTimeout(TimeUnit.MINUTES.toMillis(1))
    checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
    val sTableEnv = StreamTableEnvironment.create(env)
    sTableEnv.executeSql("CREATE TABLE sourceT (\n" +
      "  uuid varchar(20),\n" +
      "  name varchar(10),\n" +
      "  age int,\n" + "  ts timestamp(3),\n" +
      "  `partition` varchar(20)\n" +
      ") WITH (\n" +
      "  'connector' = 'datagen',\n" +
      "  'rows-per-second' = '1'\n" +
      ")")
    sTableEnv.executeSql("create table t2(\n" +
      "  uuid varchar(20),\n" +
      "  name varchar(10),\n" +
      "  age int,\n" +
      "  ts timestamp(3),\n" +
      "  `partition` varchar(20)\n" +
      ")\n" +
      "with (\n" +
      "  'connector' = 'hudi',\n" +
      "  'path' = 'hdfs://hadoop1:9000/tmp/hudi_flink/t2',\n" +
      "  'table.type' = 'MERGE_ON_READ'\n" +
      ")")
    sTableEnv.executeSql("insert into t2 select * from sourceT")
  }
}

3、提交运行

bin/flink run -t yarn-per-job -c com.my.example.HudiExample ./myjars/HudiExample-1.0-SNAPSHOT-jar-with-dependencies.jar

4、核心参数设置

        Flink可配参数：https://hudi.apache.org/docs/configurations#FLINK_SQL

4.1、去重参数

        通过如下语法设置主键：

-- 设置单个主键
create table hoodie_table (
  f0 int primary key not enforced,
  f1 varchar(20),
  ...
) with (
  'connector' = 'hudi',
  ...
)
 
-- 设置联合主键
create table hoodie_table (
  f0 int,
  f1 varchar(20),
  ...
  primary key(f0, f1) not enforced
) with (
  'connector' = 'hudi',
  ...
)

4.2、并发参数

案例演示

可以flink建表时在with中指定，或Hints临时指定参数的方式：在需要调整的表名后面加上 /*+ OPTIONS() */

CREATE TABLE sourceT (
  uuid varchar(20),
  name varchar(10),
  age int,
  ts timestamp(3),
  `partition` varchar(20)
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '1'
);
 
create table t2(
  uuid varchar(20),
  name varchar(10),
  age int,
  ts timestamp(3),
  `partition` varchar(20)
)
with (
  'connector' = 'hudi',
  'path' = '/tmp/hudi_flink/t2',
  'table.type' = 'MERGE_ON_READ'
);
 
insert into t2 /*+ OPTIONS('write.tasks'='2','write.bucket_assign.tasks'='3','compaction.tasks'='4') */ 
select * from sourceT;

执行如下图所示：

4.3、压缩参数

1、参数说明

        在线压缩的参数，通过设置 compaction.async.enabled =false关闭在线压缩执行，但是调度compaction.schedule.enabled 仍然建议开启，之后通过离线压缩直接执行 在线压缩任务 阶段性调度的压缩 plan。

2、案例演示

CREATE TABLE t3(
  uuid VARCHAR(20) PRIMARY KEY NOT ENFORCED,
  name VARCHAR(10),
  age INT,
  ts TIMESTAMP(3),
  `partition` VARCHAR(20)
)
WITH (
  'connector' = 'hudi',
  'path' = '/tmp/hudi_flink/t3',
  'compaction.async.enabled' = 'true',
  'compaction.tasks' = '1',
  'compaction.schedule.enabled' = 'true',
  'compaction.trigger.strategy' = 'num_commits',
  'compaction.delta_commits' = '2',
  'table.type' = 'MERGE_ON_READ'
);
 
set table.dynamic-table-options.enabled=true;
insert into t3
select * from sourceT/*+ OPTIONS('rows-per-second' = '5')*/;

4.4、文件大小

1、参数说明

        Hudi会自管理文件大小，避免向查询引擎暴露小文件，其中自动处理文件大小起很大作用。在进行insert/upsert操作时，Hudi可以将文件大小维护在一个指定文件大小。

        目前只有log文件的写入大小可以做到精确控制，parquet 文件大小按照估算值。

5、内存优化 

5.1、内存参数

5.2、MOR

（1）state backend 换成 rocksdb (默认的 in-memory state-backend 非常吃内存)

（2）内存够的话，compaction.max_memory 调大些 (默认是 100MB 可以调到 1GB)

（3）关注 TM 分配给每个 write task 的内存，保证每个 write task 能够分配到 write.task.max.size 所配置的大小，比如 TM 的内存是 4GB 跑了 2 个 StreamWriteFunction 那每个 write function 能分到 2GB，尽量预留一些 buffer，因为网络 buffer，TM 上其他类型 task (比如 BucketAssignFunction 也会吃些内存)

（4）需要关注 compaction 的内存变化，compaction.max_memory 控制了每个 compaction task 读 log 时可以利用的内存大小，compaction.tasks 控制了 compaction task 的并发

        注意: write.task.max.size - compaction.max_memory 是预留给每个 write task 的内存 buffer

5.3、COW

（1）state backend 换成 rocksdb（默认的 in-memory state-backend 非常吃内存）。

（2）write.task.max.size 和 write.merge.max_memory 同时调大（默认是 1GB 和 100MB 可以调到 2GB 和 1GB）。

（3）关注 TM 分配给每个 write task 的内存，保证每个 write task 能够分配到 write.task.max.size 所配置的大小，比如 TM 的内存是 4GB 跑了 2 个 StreamWriteFunction 那每个 write function 能分到 2GB，尽量预留一些 buffer，因为网络 buffer，TM 上其他类型 task（比如 BucketAssignFunction 也会吃些内存）。

        注意：write.task.max.size - write.merge.max_memory 是预留给每个 write task 的内存 buffer。