Seatunnel 检查点用法看这篇文章就够了

文摘 2024-11-19 00:00 重庆

检查点是一种容错恢复机制。这种机制确保程序在运行时，即使突然遇到异常，也能自行恢复。

检查点存储概念

SeaTunnel Engine支持以下检查点存储类型:

HDFS (OSS,COS,S3,HDFS,LocalFile)
LocalFile (本地)，(已弃用: 使用HDFS(LocalFile)替代).

我们使用微内核设计模式将检查点存储模块从引擎中分离出来。这允许用户实现他们自己的检查点存储模块。

checkpoint-storage-api是检查点存储模块API，它定义了检查点存储模块的接口。

如果你想实现你自己的检查点存储模块，你需要实现CheckpointStorage并提供相应的CheckpointStorageFactory实现。

这或许是一个对你有用的开源项目，data-warehouse-learning 项目是一套基于 MySQL + Kafka + Hadoop + Hive + Dolphinscheduler + Doris + Seatunnel + Paimon + Hudi + Iceberg + Flink + Dinky + DataRT + SuperSet 实现的实时离线数仓（数据湖）系统，以大家最熟悉的电商业务为切入点，详细讲述并实现了数据产生、同步、数据建模、数仓（数据湖）建设、数据服务、BI报表展示等数据全链路处理流程。
https://gitee.com/wzylzjtn/data-warehouse-learning
https://github.com/Mrkuhuo/data-warehouse-learning
https://bigdatacircle.top/
项目演示：

检查点存储配置

seatunnel-server模块的配置在seatunnel.yaml文件中。


seatunnel:    engine:        checkpoint:            storage:                type: hdfs #检查点存储的插件名称，支持hdfs(S3, local, hdfs), 默认为localfile (本地文件), 但这种方式已弃用              # 插件配置                plugin-config:                   namespace: #检查点存储父路径，默认值为/seatunnel/checkpoint/                  K1: V1 # 插件其它配置                  K2: V2 # 插件其它配置

注意: namespace必须以"/"结尾。

OSS

阿里云OSS是基于hdfs-file，所以你可以参考Hadoop OSS文档来配置oss.

OSS buckets交互外，oss客户端需要与buckets交互所需的凭据。客户端支持多种身份验证机制，并且可以配置使用哪种机制及其使用顺序。也可以使用of org.apache.hadoop.fs.aliyun.oss.AliyunCredentialsProvider的自定义实现。如果您使用AliyunCredentialsProvider(可以从阿里云访问密钥管理中获得)，它们包括一个access key和一个secret key。你可以这样配置:

seatunnel:  engine:    checkpoint:      interval: 6000      timeout: 7000      storage:        type: hdfs        max-retained: 3        plugin-config:          storage.type: oss          oss.bucket: your-bucket          fs.oss.accessKeyId: your-access-key          fs.oss.accessKeySecret: your-secret-key          fs.oss.endpoint: endpoint address

阿里云OSS凭证提供程序实现见: （https://github.com/aliyun/aliyun-oss-java-sdk/tree/master/src/main/java/com/aliyun/oss/common/auth）

COS

腾讯云COS基于hdfs-file，所以你可以参考（https://hadoop.apache.org/docs/stable/hadoop-cos/cloud-storage/）来配置COS.

除了与公共COS buckets交互之外，COS客户端需要与buckets交互所需的凭据。客户端支持多种身份验证机制，并且可以配置使用哪种机制及其使用顺序。也可以使用com.qcloud.cos.auth.COSCredentialsProvider的自定义实现。如果您使用SimpleCredentialsProvider(可以从腾讯云API密钥管理中获得)，它们包括一个secretId和一个secretKey。您可以这样配置:

seatunnel:  engine:    checkpoint:      interval: 6000      timeout: 7000      storage:        type: hdfs        max-retained: 3        plugin-config:          storage.type: cos          cos.bucket: cosn://your-bucket          fs.cosn.credentials.provider: org.apache.hadoop.fs.cosn.auth.SimpleCredentialsProvider          fs.cosn.userinfo.secretId: your-secretId          fs.cosn.userinfo.secretKey: your-secretKey          fs.cosn.bucket.region: your-region

腾讯云COS相关配置可参考：（https://doc.fincloud.tencent.cn/tcloud/Storage/COS/846365/hadoop）

使用前请将如下jar添加到lib目录下：

hadoop-cos-3.4.1.jar（https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-cos/3.4.1）
cos_api-bundle-5.6.69.jar（https://mvnrepository.com/artifact/com.qcloud/cos_api-bundle/5.6.69）
hadoop-shaded-guava-1.1.1.jar（https://mvnrepository.com/artifact/org.apache.hadoop.thirdparty/hadoop-shaded-guava/1.1.1）

S3基于hdfs-file，所以你可以参考（https://hadoop.apache.org/docs/stable/hadoop-aws/tools/hadoop-aws/index.html）来配置s3。

除了与公共S3 buckets交互之外，S3A客户端需要与buckets交互所需的凭据。客户端支持多种身份验证机制，并且可以配置使用哪种机制及其使用顺序。也可以使用com.amazonaws.auth.AWSCredentialsProvider的自定义实现。如果您使用SimpleAWSCredentialsProvider(可以从Amazon Security Token服务中获得)，它们包括一个access key和一个secret key。您可以这样配置:

seatunnel:    engine:        checkpoint:            interval: 6000            timeout: 7000            storage:                type: hdfs                max-retained: 3                plugin-config:                    storage.type: s3                    s3.bucket: your-bucket                    fs.s3a.access.key: your-access-key                    fs.s3a.secret.key: your-secret-key                    fs.s3a.aws.credentials.provider: org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider

如果您使用InstanceProfileCredentialsProvider，它支持在EC2 VM中运行时使用实例配置文件凭据，您可以检查（https://docs.aws.amazon.com/zh_cn/AWSEC2/latest/UserGuide/iam-roles-for-amazon-ec2.html）. 您可以这样配置:


seatunnel:  engine:    checkpoint:      interval: 6000      timeout: 7000      storage:        type: hdfs        max-retained: 3        plugin-config:          storage.type: s3          s3.bucket: your-bucket          fs.s3a.endpoint: your-endpoint          fs.s3a.aws.credentials.provider: org.apache.hadoop.fs.s3a.InstanceProfileCredentialsProvider

HDFS

如果您使用HDFS，您可以这样配置:

seatunnel:  engine:    checkpoint:      storage:        type: hdfs        max-retained: 3        plugin-config:          storage.type: hdfs          fs.defaultFS: hdfs://localhost:9000          // 如果您使用kerberos，您可以这样配置:          kerberosPrincipal: your-kerberos-principal          kerberosKeytabFilePath: your-kerberos-keytab

如果HDFS是HA模式，您可以这样配置:

seatunnel:  engine:    checkpoint:      storage:        type: hdfs        max-retained: 3        plugin-config:          storage.type: hdfs          fs.defaultFS: hdfs://usdp-bing          seatunnel.hadoop.dfs.nameservices: usdp-bing          seatunnel.hadoop.dfs.ha.namenodes.usdp-bing: nn1,nn2          seatunnel.hadoop.dfs.namenode.rpc-address.usdp-bing.nn1: usdp-bing-nn1:8020          seatunnel.hadoop.dfs.namenode.rpc-address.usdp-bing.nn2: usdp-bing-nn2:8020          seatunnel.hadoop.dfs.client.failover.proxy.provider.usdp-bing: org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider

如果HDFS在hdfs-site.xml或core-site.xml中有其他配置，只需使用seatunnel.hadoop.前缀设置HDFS配置即可。

本地文件

seatunnel:  engine:    checkpoint:      interval: 6000      timeout: 7000      storage:        type: hdfs        max-retained: 3        plugin-config:          storage.type: hdfs          fs.defaultFS: file:/// # 请确保该目录具有写权限

开启高速缓存

当storage:type为hdfs时，默认关闭cache。如果您想启用它，请设置为disable.cache: false。

seatunnel:  engine:    checkpoint:      interval: 6000      timeout: 7000      storage:        type: hdfs        max-retained: 3        plugin-config:          storage.type: hdfs          disable.cache: false          fs.defaultFS: hdfs:/// # Ensure that the directory has written permission

或者

seatunnel:  engine:    checkpoint:      interval: 6000      timeout: 7000      storage:        type: hdfs        max-retained: 3        plugin-config:          storage.type: hdfs          disable.cache: false          fs.defaultFS: file:///

进交流群群添加作者

Seatunnel 检查点用法看这篇文章就够了

推荐阅读系列文章