当前位置： 首页 > news >正文

自己建的网站也要注册域名吗电商网站平台

news 2026/4/7 19:32:49
自己建的网站也要注册域名吗,电商网站平台,网站展示,网站建设深圳亿联时代前言 如题，记录几个Hudi Flink使用问题，学习和使用Hudi Flink有一段时间，虽然目前用的还不够深入，但是目前也遇到了几个问题，现在将遇到的这几个问题以及解决方式记录一下 版本 Flink 1.15.4Hudi 0.13.0 流写 流写…

前言

如题，记录几个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

public 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表发现数据量变多，也就是有重复数据。
主要参数:

set 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

'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 to BUCKET. 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索引主要参数：

'index.type' =  'BUCKET', -- flink只支持两种index，默认FLINK_STATE index,FLINK_STATE index对于数据量比较大的情况会因为tm内存不足导致GC OOM
'hoodie.bucket.index.num.buckets' = '16', -- 桶数

注意，index.type是flink客户端独有的，和公共的不一样（使用公共参数不生效），没有前缀hoodie.,而桶数配置项是hudi公共参数，对于flink客户端哪些用公共参数哪些用flink独有的参数，官方文档并没有提供，需要自己在类org.apache.hudi.configuration.FlinkOptions查看，该类中的参数为flink重写的独有参数，没有的话则需要使用公共参数

insert转upsert问题

对于BUCKET如果先insert一部分历史数据，再upsert增量数据时，默认参数配置会抛出如下异常：
(复现此问题只需要批写一条数据即可)

Caused by: java.lang.NumberFormatException: For input string: "4ff32a41"at java.lang.NumberFormatException.forInputString(NumberFormatException.java:65)at java.lang.Integer.parseInt(Integer.java:580)at java.lang.Integer.parseInt(Integer.java:615)at org.apache.hudi.index.bucket.BucketIdentifier.bucketIdFromFileId(BucketIdentifier.java:79)at org.apache.hudi.sink.bucket.BucketStreamWriteFunction.lambda$bootstrapIndexIfNeed$1(BucketStreamWriteFunction.java:162)at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184)at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1382)at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:151)at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:174)at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:418)at org.apache.hudi.sink.bucket.BucketStreamWriteFunction.bootstrapIndexIfNeed(BucketStreamWriteFunction.java:160)at org.apache.hudi.sink.bucket.BucketStreamWriteFunction.processElement(BucketStreamWriteFunction.java:112)at org.apache.flink.streaming.api.operators.ProcessOperator.processElement(ProcessOperator.java:66)at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:233)

原因是：默认参数下，insert时没有按照bucket索引的逻辑写文件，而upsert是按照bucket逻辑写文件的，bucket索引写的文件名前缀都带有桶号，不是bucket索引写的文件名没有桶号，所以upsert时会尝试解析insert写的历史文件的桶号，导致解析失败。

非bucket索引逻辑写的文件

/tmp/cdc/hudi_sink_insert/4ff32a41-4232-4f47-855a-6364eb1d6ce8-0_0-1-0_20230820210751280.parquet

bucket索引逻辑写的文件

/tmp/cdc/hudi_sink_insert/00000000-82f4-48a5-85e9-2c4bb9679360_0-1-0_20230820211542006.parquet

解决方法

对于实际应用场景是有这种先insert在upsert的需求的，解决方法就是尝试通过配置参数使insert也按照bucket索引的逻辑写数据
主要参数：'write.insert.cluster'='true'
相关参数：

'write.operation'='insert', 
'table.type'='COPY_ON_WRITE',
'write.insert.cluster'='true',
'index.type' =  'BUCKET',

我是通过阅读源码发现这个参数可以使insert按照bucket逻辑写数据的
对应的源码在HoodieTableSink.getSinkRuntimeProvider,我在上篇文章Hudi Flink SQL源码调试学习（一）中分析了写hudi时是如何调用到这个方法的，感兴趣得可以看一下。

public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {return (DataStreamSinkProviderAdapter) dataStream -> {// setup configurationlong ckpTimeout = dataStream.getExecutionEnvironment().getCheckpointConfig().getCheckpointTimeout();conf.setLong(FlinkOptions.WRITE_COMMIT_ACK_TIMEOUT, ckpTimeout);// set up default parallelismOptionsInference.setupSinkTasks(conf, dataStream.getExecutionConfig().getParallelism());RowType rowType = (RowType) schema.toSinkRowDataType().notNull().getLogicalType();// bulk_insert modefinal String writeOperation = this.conf.get(FlinkOptions.OPERATION);if (WriteOperationType.fromValue(writeOperation) == WriteOperationType.BULK_INSERT) {return Pipelines.bulkInsert(conf, rowType, dataStream);}// Append modeif (OptionsResolver.isAppendMode(conf)) {DataStream<Object> pipeline = Pipelines.append(conf, rowType, dataStream, context.isBounded());if (OptionsResolver.needsAsyncClustering(conf)) {return Pipelines.cluster(conf, rowType, pipeline);} else {return Pipelines.dummySink(pipeline);}}DataStream<Object> pipeline;// bootstrapfinal DataStream<HoodieRecord> hoodieRecordDataStream =Pipelines.bootstrap(conf, rowType, dataStream, context.isBounded(), overwrite);// write pipelinepipeline = Pipelines.hoodieStreamWrite(conf, hoodieRecordDataStream);// compactionif (OptionsResolver.needsAsyncCompaction(conf)) {// use synchronous compaction for bounded source.if (context.isBounded()) {conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);}return Pipelines.compact(conf, pipeline);} else {return Pipelines.clean(conf, pipeline);}};}

我们在上面的代码中可以发现，当是append模式时会走单独的写逻辑，不是append模式时，才会走下面的Pipelines.hoodieStreamWrite,那么就需要看一下append模式的判断逻辑

OptionsResolver.isAppendMode(conf)

  public static boolean isAppendMode(Configuration conf) {// 1. inline clustering is supported for COW table;// 2. async clustering is supported for both COW and MOR tablereturn isCowTable(conf) && isInsertOperation(conf) && !conf.getBoolean(FlinkOptions.INSERT_CLUSTER)|| needsScheduleClustering(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表的异常

异常信息

Caused by: java.lang.ClassCastException: org.apache.hadoop.io.ArrayWritable cannot be cast to org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatchat org.apache.hadoop.hive.ql.exec.vector.VectorMapOperator.deliverVectorizedRowBatch(VectorMapOperator.java:803)at org.apache.hadoop.hive.ql.exec.vector.VectorMapOperator.process(VectorMapOperator.java:845)... 20 morejava.lang.ClassCastException: org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch cannot be cast to org.apache.hadoop.io.ArrayWritablejava.lang.ClassCastException: org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch cannot be cast to org.apache.hadoop.io.ArrayWritable

异常复现

找一个hudi mor表的rt表，执行count语句(有人反馈聚合函数也会出现此异常)

解决方法

set hive.vectorized.execution.enabled=false; (我验证的这一个参数就可以了)
set hive.vectorized.execution.reduce.enabled=false;（不确定此参数是否必须）

相关阅读

  • Flink SQL Checkpoint 学习总结
  • Hudi Flink SQL源码调试学习（一）
  • Flink SQL操作Hudi并同步Hive使用总结
http://www.hkea.cn/news/752085/

相关文章：

  • 哪里做网站比较快2345网址导航下载桌面
  • 广州建设委员会官方网站凡科建站下载
  • 全球做网站的公司排名百度一下你就知道官网
  • 小企业网站价格免费发链接的网站
  • 买了空间和域名 怎么做网站哪家公司网站做得好
  • 网站备案是否关闭衡阳网站建设公司
  • 遂昌建设局网站个人怎么做网站
  • 软件开发和网站建设网络营销的未来6个发展趋势
  • 做网站一年多少钱免费seo网站推广
  • 智通人才网东莞最新招聘信息官网seo是如何做优化的
  • 个人做跨境电商网站百度地图导航手机版免费下载
  • 阿里云注册网站之后怎么做网站百度联盟是什么
  • 动画制作视频河南网站排名优化
  • 网站关键词怎么做排名掌门一对一辅导官网
  • 现在什么网站做推广比较好网页设计需要学什么
  • 个人购物网站 怎么建网络营销包括
  • 有没有做鸭的网站工作室招聘广州网站优化工具
  • 深圳营销外深圳网络营销公司seo和sem的联系
  • 专业的网站制作公司哪家好竞价专员是做什么的
  • 海南省建设厅网站百度seo霸屏软件
  • 淄博张店做网站的公司爱站小工具圣经
  • wordpress w3seo优化自学
  • 临沂手机建站模板微信seo排名优化软件
  • 网站管理员怎么做板块建设艺人百度指数排行榜
  • 如何创建企业网站网络舆情处置的五个步骤
  • 做站长工具网站周口seo公司
  • 泉州自助建站系统地推
  • 美国 做网站免费网站建设哪家好
  • 如何做响应式布局网站seo搜索引擎优化期末及答案
  • 电脑系统优化软件十大排名北京网优化seo公司