前些天发现了一个巨牛的人工智能学习网站,通俗易懂,风趣幽默,忍不住给大家分享一下。点击跳转到网站:https://www.captainai.net/dongkelun
前言
如题,记录几个Hudi Flink使用问题,学习和使用Hudi Flink有一段时间,虽然目前用的还不够深入,但是目前也遇到了几个问题,现在将遇到的这几个问题以及解决方式记录一下
版本
- Flink 1.15.4
- Hudi 0.13.0
流写
流写Hudi,必须要开启Checkpoint,这个我在之前的文章:Flink SQL Checkpoint 学习总结提到过。
如果不设置Checkpoint,不会生成commit,感觉像是卡住一样,具体表现为只生成.commit.requested和.inflight,然后不写文件、不生成.commit也不报错,对于新手来说很费劲,很难找到解决方法。
索引
hudi-flink 仅支持两种索引:FLINK_STATE
和BUCKET
,默认FLINK_STATE
。
最开始使用hudi是用的spark,hudi-spark支持
BLOOM
索引,hudi java client也支持BLOOM
索引,所以认为hudi-flink也支持BLOOM
索引,但其实不支持,而且官网并没有相关的文档说明,可以从下面这段代码中看出来
Pipelines.hoodieStreamWrite
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30public static DataStream<Object> hoodieStreamWrite(Configuration conf, DataStream<HoodieRecord> dataStream) {
// 如果是`BUCKET`索引
if (OptionsResolver.isBucketIndexType(conf)) {
WriteOperatorFactory<HoodieRecord> operatorFactory = BucketStreamWriteOperator.getFactory(conf);
int bucketNum = conf.getInteger(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS);
String indexKeyFields = conf.getString(FlinkOptions.INDEX_KEY_FIELD);
BucketIndexPartitioner<HoodieKey> partitioner = new BucketIndexPartitioner<>(bucketNum, indexKeyFields);
return dataStream.partitionCustom(partitioner, HoodieRecord::getKey)
.transform(opName("bucket_write", conf), TypeInformation.of(Object.class), operatorFactory)
.uid(opUID("bucket_write", conf))
.setParallelism(conf.getInteger(FlinkOptions.WRITE_TASKS));
} else {
// 否则按`FLINK_STATE`索引的逻辑
WriteOperatorFactory<HoodieRecord> operatorFactory = StreamWriteOperator.getFactory(conf);
return dataStream
// Key-by record key, to avoid multiple subtasks write to a bucket at the same time
.keyBy(HoodieRecord::getRecordKey)
.transform(
"bucket_assigner",
TypeInformation.of(HoodieRecord.class),
new KeyedProcessOperator<>(new BucketAssignFunction<>(conf)))
.uid(opUID("bucket_assigner", conf))
.setParallelism(conf.getInteger(FlinkOptions.BUCKET_ASSIGN_TASKS))
// shuffle by fileId(bucket id)
.keyBy(record -> record.getCurrentLocation().getFileId())
.transform(opName("stream_write", conf), TypeInformation.of(Object.class), operatorFactory)
.uid(opUID("stream_write", conf))
.setParallelism(conf.getInteger(FlinkOptions.WRITE_TASKS));
}
}
FLINK_STATE 重复问题
如果使用默认的FLINK_STATE
索引,在upsert时可能会有重复问题。(之前使用BLOOM
索引时不会有这个问题)
问题复现
先写一部分数据作为历史数据到Hudi表,然后再写相同的数据到这个表,最后count表发现数据量变多,也就是有重复数据。
主要参数:1
2
3
4
5
6set parallelism.default=12;
set taskmanager.numberOfTaskSlots=2;
'write.operation'='upsert',
'write.tasks'='11',
'table.type'='COPY_ON_WRITE',
场景为kafka2hudi,kafka数据量200w,没有重复,设置并发的主要目的是为了将数据打散分布在不同的文件里,这样更容易复现问题。(因为如果只有一个历史文件时,很难复现)
第一次任务跑完表数据量为200w,第二次跑完表数据量大于200w,证明数据重复。
重复原因
index state
:保存在state中的主键和文件ID的对应关系
重复的原因为FLINK_STATE
将主键和文件ID的对应关系保存在state中,当新启动一个任务时,index state
需要重新建立,而默认情况下不会包含历史文件的index state
,只会建立新数据的index state
,所以对于没有历史文件的新表是不会有重复问题的。(对于有历史文件的表,如果从checkpoint恢复也不会有重复问题,因为从checkpoint恢复时,也恢复了之前历史文件的index state
)
解决方法
通过参数index.bootstrap.enabled
解决,默认为false,当为true时,写hudi任务启动时会先引导(加载)历史文件的index state
1
'index.bootstrap.enabled'='true'
除了重复问题,
FLINK_STATE
因为将index保存在state中,所以随着数据量的增加,state越来越大。这样对于数据量特别大的表,对内存的要求也会很高,所以会遇到内存不足OOM的问题。 所以建议对于大表,还是选择使用BUCKET
索引。
增量数据,’index.bootstrap.enabled’=’false’时的checkpoint记录,checkpoint大小开始很小,然后逐渐增加
增量数据,’index.bootstrap.enabled’=’true’时的checkpoint记录,checkpoint大小开始和结束差不多大
BUCKET INDEX
BUCKET
索引需要根据表数据量大小设定好桶数(hoodie.bucket.index.num.buckets
),但是默认情况下不能动态调整bucket
数量。
另外可以通过参数hoodie.index.bucket.engine
将其值设为CONSISTENT_HASHING
,通过一致性哈希实现动态调整bucket
数量,但是仅支持MOR
表,我还没有试过这个功能,大家可以通过官网:https://hudi.apache.org/docs/configurations/了解相关参数自行测试。
hoodie.index.bucket.engine | SIMPLE (Optional) | org.apache.hudi.index.HoodieIndex$BucketIndexEngineType: Determines the type of bucketing or hashing to use when
hoodie.index.type
is set toBUCKET
. SIMPLE(default): Uses a fixed number of buckets for file groups which cannot shrink or expand. This works for both COW and MOR tables. CONSISTENT_HASHING: Supports dynamic number of buckets with bucket resizing to properly size each bucket. This solves potential data skew problem where one bucket can be significantly larger than others in SIMPLE engine type. This only works with MOR tables.Config Param: BUCKET_INDEX_ENGINE_TYPE
Since Version: 0.11.0
BUCKET
索引主要参数:
1 | 'index.type' = 'BUCKET', -- flink只支持两种index,默认FLINK_STATE index,FLINK_STATE index对于数据量比较大的情况会因为tm内存不足导致GC OOM |
注意,
index.type
是flink客户端独有的,和公共的不一样(使用公共参数不生效),没有前缀hoodie.
,而桶数配置项是hudi公共参数,对于flink客户端哪些用公共参数哪些用flink独有的参数,官方文档并没有提供,需要自己在类org.apache.hudi.configuration.FlinkOptions
查看,该类中的参数为flink重写的独有参数,没有的话则需要使用公共参数
insert转upsert问题
对于BUCKET
如果先insert一部分历史数据,再upsert增量数据时,默认参数配置会抛出如下异常:
(复现此问题只需要批写一条数据即可)
1 | Caused by: java.lang.NumberFormatException: For input string: "4ff32a41" |
原因是:默认参数下,insert时没有按照bucket
索引的逻辑写文件,而upsert
是按照bucket
逻辑写文件的,bucket
索引写的文件名前缀都带有桶号,不是bucket
索引写的文件名没有桶号,所以upsert时会尝试解析insert写的历史文件的桶号,导致解析失败。
非bucket
索引逻辑写的文件1
/tmp/cdc/hudi_sink_insert/4ff32a41-4232-4f47-855a-6364eb1d6ce8-0_0-1-0_20230820210751280.parquet
bucket
索引逻辑写的文件1
/tmp/cdc/hudi_sink_insert/00000000-82f4-48a5-85e9-2c4bb9679360_0-1-0_20230820211542006.parquet
解决方法
对于实际应用场景是有这种先insert在upsert的需求的,解决方法就是尝试通过配置参数使insert也按照bucket
索引的逻辑写数据
主要参数:'write.insert.cluster'='true'
相关参数:1
2
3
4'write.operation'='insert',
'table.type'='COPY_ON_WRITE',
'write.insert.cluster'='true',
'index.type' = 'BUCKET',
我是通过阅读源码发现这个参数可以使insert按照
bucket
逻辑写数据的
对应的源码在HoodieTableSink.getSinkRuntimeProvider
,我在上篇文章Hudi Flink SQL源码调试学习(一)中分析了写hudi时是如何调用到这个方法的,感兴趣得可以看一下。
1 | public SinkRuntimeProvider getSinkRuntimeProvider(Context context) { |
我们在上面的代码中可以发现,当是append模式时会走单独的写逻辑,不是append模式时,才会走下面的Pipelines.hoodieStreamWrite
,那么就需要看一下append模式的判断逻辑
OptionsResolver.isAppendMode(conf)
1 | public static boolean isAppendMode(Configuration conf) { |
对于cow表insert时,默认参数的情况needsScheduleClustering(conf)
返回false,而!conf.getBoolean(FlinkOptions.INSERT_CLUSTER)
返回true,所以只需要让!conf.getBoolean(FlinkOptions.INSERT_CLUSTER)
返回false就可以跳过append模式的逻辑了,也就是上面的 'write.insert.cluster'='true'
。(每个版本的源码不太一样,所以对于其他版本,可能这个参数并不能解决该问题)
Hive查询异常
记录一个Hive SQL查询Hudi表的异常
异常信息
1 | Caused by: java.lang.ClassCastException: org.apache.hadoop.io.ArrayWritable cannot be cast to org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch |
异常复现
找一个hudi mor
表的rt
表,执行count语句(有人反馈聚合函数也会出现此异常)
解决方法
1 | set hive.vectorized.execution.enabled=false; (我验证的这一个参数就可以了) |