Flink SQL Checkpoint 学习总结

前言

学习总结Flink SQL Checkpoint的使用，主要目的是为了验证Flink SQL流式任务挂掉后，重启时还可以继续从上次的运行状态恢复。

验证方式

Flink SQL流式增量读取Hudi表然后sink MySQL表，任务启动后处于running状态，先查看sink表有数据，然后将对应的yarn kill掉，再通过设置的checkpoint重启任务，任务重启后验证sink表的数据量。Flink SQL流式增量读取Hudi表可以参考：Flink SQL增量查询Hudi表

版本

  • Flink 1.14.3
  • Hudi 0.13.0

Checkpoint 参数

一般需要设置的常用参数

1
2
3
4
5
6
7
8
9
-- checkpoint间隔时间，单位毫秒，没有默认值，如果想开启checkpoint，需要将该参数设置一个大于0的数值
-- 如果想提升sink性能，比如写hudi，需要将该值设置大一点，因为间隔时间决定了批次大小
-- checkpoint间隔时间不能设置太短也不能设置太长，太短影响写入性能，太长影响数据及时性。
set execution.checkpointing.interval=1000;
-- 保存checkpoint文件的目录
set state.checkpoints.dir=hdfs:///flink/checkpoints/hudi2mysql;
-- 任务取消后保留checkpoint,默认值NO_EXTERNALIZED_CHECKPOINTS，
-- 可选值NO_EXTERNALIZED_CHECKPOINTS、DELETE_ON_CANCELLATION、RETAIN_ON_CANCELLATION
set execution.checkpointing.externalized-checkpoint-retention=RETAIN_ON_CANCELLATION;

从checkpoint恢复

1
set execution.savepoint.path=hdfs:///flink/checkpoints/hudi2mysql/255bdd01cee7486113feb1cbe8b45ee0/chk-1314;

其他参数

1
2
3
4
5
6
7
-- checkpoint模式，默认值EXACTLY_ONCE，可选值：EXACTLY_ONCE、AT_LEAST_ONCE
-- 要想支持EXACTLY_ONCE，需要sink端支持事务
set execution.checkpointing.mode=EXACTLY_ONCE;
-- checkpoint超时时间，默认10分钟
set execution.checkpointing.timeout=600000;
-- checkpoint文件保留数，默认1
set state.checkpoints.num-retained=3;
more >>

展开全文 >>

Spark SQL增量查询Hudi表

前言

由于项目上主要用Hive查询Hudi，所以之前总结过一篇:Hive增量查询Hudi表。最近可能会有Spark SQL增量查询Hudi表的需求，并且我发现目前用纯Spark SQL的形式还不能直接增量查询Hudi表，于是进行学习总结一下。

编程方式（DF+SQL）

先看一下官方文档上Spark SQL增量查询的方式，地址：https://hudi.apache.org/cn/docs/quick-start-guide#incremental-query 和 https://hudi.apache.org/cn/docs/querying_data#incremental-query

它是先通过spark.read中添加增量参数的形式读Hudi表为DF，然后将DF注册成临时表，最后通过Spark SQL查询临时表的形式，实现增量查询的。

more >>

展开全文 >>

Hudi Spark SQL Call Procedures学习总结（一）（查询统计表文件信息）

前言

学习总结Hudi Spark SQL Call Procedures，Call Procedures在官网被称作存储过程（Stored Procedures），它是在Hudi 0.11.0版本由腾讯的ForwardXu大佬贡献的，它除了官网提到的几个Procedures外，还支持其他许多Procedures命令。本文先学习其中的几个我觉得比较常用的命令，主要是查询统计表路径下的各种文件信息。

版本

Hudi master 0.13.0-SNAPSHOT
Spark 3.1.2 (实际上所有Hudi支持的Spark版本都支持Call Procedures)
Kyuubi 1.5.2 (使用Kyuubi是因为返回结果可以展示列名，Spark自带的spark-sql不返回列名)

参数形式

按名称传递参数，没有顺序，可以省略可选参数

1
CALL system.procedure_name(arg_name_2 => arg_2, arg_name_1 => arg_1, ... arg_name_n => arg_n)

按位置参数传递参数，有顺序，可以省略可选参数

1
CALL system.procedure_name(arg_1, arg_2, ... arg_n)

more >>

展开全文 >>

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

2016年本科毕业于合肥工业大学<br><br>信息与计算科学专业<br><br>目前在济南一家公司从事大数据的工作~~