分区不存在导致的写入kudu出错

ゞ浴缸里的玫瑰 2024-03-30 11:24 23阅读 0赞

错误信息如下：

2022-12-22 13:34:40
    java.io.IOException: Error while sending value. 
     Row error for primary key="\x80\x00\x00\x00\x00\x07\xE3\xB7ypsx_scm_purchase_3\x00\x00purchase_order_detail\x00\x00\x80\x05\xC3\xE0a\xF3\x9F\xC0", tablet=null, server=null, status=Not found: ([0x000000018005A955F9188000, 0x000000018005C604852C6000))
    Row error for primary key="\x80\x00\x00\x00\x00\x07\xE8\xF6ypsx_scm_purchase_3\x00\x00purchase_order_detail\x00\x00\x80\x05\xC3\xF1\x93(\x9B\xC0", tablet=null, server=null, status=Not found: ([0x000000028005A955F9188000, 0x000000028005C604852C6000))
    Row error for primary key="\x80\x00\x00\x00\x00\x07\x88\xE9ypsx_scm_purchase_3\x00\x00purchase_order_detail\x00\x00\x80\x05\xC3=\xCF\xE7\x0B\x00", tablet=null, server=null, status=Not found: ([0x000000038005A955F9188000, 0x000000038005C604852C6000))
    Row error for primary key="\x80\x00\x00\x00\x00\x02\x0FJypsx_scm_purchase_2\x00\x00purchase_order_detail\x00\x00\x80\x05\xB6\x1F\xA1\xA8\xF8\x00", tablet=null, server=null, status=Not found: ([0x000000018005A955F9188000, 0x000000018005C604852C6000))
    Row error for primary key="\x80\x00\x00\x00\x00\x05\xE4\xF8ypsx_scm_purchase_1\x00\x00purchase_order_detail\x00\x00\x80\x05\xBF\xF8\x0F\xE4\xC9\xC0", tablet=null, server=null, status=Not found: ([0x000000038005A955F9188000, 0x000000038005C604852C6000))
    Row error for primary key="\x80\x00\x00\x00\x00\x01\xB4~ypsx_scm_purchase_1\x00\x00purchase_order_detail\x00\x00\x80\x05\xB4\xE0\x04\xD0~\x80", tablet=null, server=null, status=Not found: ([0x000000038005A955F9188000, 0x000000038005C604852C6000))
    Row error for primary key="\x80\x00\x00\x00\x00\x05\xE4\xF7ypsx_scm_purchase_1\x00\x00purchase_order_detail\x00\x00\x80\x05\xBF\xF8\x0F\xE4\xC9\xC0", tablet=null, server=null, status=Not found: ([0x000000038005A955F9188000, 0x000000038005C604852C6000))
    Row error for primary key="\x80\x00\x00\x00\x00\x05\xE4\xF6ypsx_scm_purchase_1\x00\x00purchase_order_detail\x00\x00\x80\x05\xBF\xF8\x0F\xE4\xC9\xC0", tablet=null, server=null, status=Not found: ([0x000000038005A955F9188000, 0x000000038005C604852C6000))
    
    	at org.apache.flink.connectors.kudu.connector.failure.DefaultKuduFailureHandler.onFailure(DefaultKuduFailureHandler.java:37)
    	at org.apache.flink.connectors.kudu.connector.writer.KuduWriter.checkAsyncErrors(KuduWriter.java:152)
    	at org.apache.flink.connectors.kudu.connector.writer.KuduWriter.write(KuduWriter.java:95)
    	at org.apache.flink.connectors.kudu.streaming.KuduSink.invoke(KuduSink.java:94)
    	at org.apache.flink.streaming.api.functions.sink.SinkFunction.invoke(SinkFunction.java:50)
    	at org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:54)
    	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:82)
    	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:57)
    	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
    	at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:56)
    	at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:29)
    	at SinkConversion$80.processElement(Unknown Source)
    	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:82)
    	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:57)
    	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
    	at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:56)
    	at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:29)
    	at StreamExecCalc$77.processElement_split5(Unknown Source)
    	at StreamExecCalc$77.processElement(Unknown Source)
    	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:82)
    	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:57)
    	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
    	at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:56)
    	at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:29)
    	at org.apache.flink.streaming.api.operators.StreamFilter.processElement(StreamFilter.java:39)
    	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:82)
    	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:57)
    	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
    	at org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask$AsyncDataOutputToOutput.emitRecord(SourceOperatorStreamTask.java:313)
    	at org.apache.flink.streaming.api.operators.source.SourceOutputWithWatermarks.collect(SourceOutputWithWatermarks.java:110)
    	at org.apache.flink.connector.pulsar.source.reader.emitter.PulsarRecordEmitter$SourceOutputWrapper.collect(PulsarRecordEmitter.java:65)
    	at org.apache.flink.connector.pulsar.table.source.PulsarRowDataConverter.emitRow(PulsarRowDataConverter.java:128)
    	at org.apache.flink.connector.pulsar.table.source.PulsarRowDataConverter.lambda$projectToProducedRowAndCollect$0(PulsarRowDataConverter.java:72)
    	at java.util.ArrayList.forEach(ArrayList.java:1259)
    	at org.apache.flink.connector.pulsar.table.source.PulsarRowDataConverter.projectToProducedRowAndCollect(PulsarRowDataConverter.java:71)
    	at org.apache.flink.connector.pulsar.table.source.PulsarTableDeserializationSchema.deserialize(PulsarTableDeserializationSchema.java:107)
    	at org.apache.flink.connector.pulsar.source.reader.emitter.PulsarRecordEmitter.emitRecord(PulsarRecordEmitter.java:54)
    	at org.apache.flink.connector.pulsar.source.reader.emitter.PulsarRecordEmitter.emitRecord(PulsarRecordEmitter.java:36)
    	at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:143)
    	at org.apache.flink.connector.pulsar.source.reader.source.PulsarOrderedSourceReader.pollNext(PulsarOrderedSourceReader.java:109)
    	at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:385)
    	at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68)
    	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
    	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:519)
    	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:203)
    	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:804)
    	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:753)
    	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:948)
    	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:927)
    	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:741)
    	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:563)
    	at java.lang.Thread.run(Thread.java:750)

主要原因是因为数据在kudu中所属的分区不存在，导致数据无法写入，因此增加相应的分区就好了，

没增加前：

range_partitions = '[{"lower":null,"upper":"2020-07-01T00:00:00.000Z"},
    
    {"lower":"2021-07-01T00:00:00.000Z","upper":"2022-01-01T00:00:00.000Z"},
    
    {"lower":"2022-01-01T00:00:00.000Z","upper":"2022-07-01T00:00:00.000Z"},
    
    {"lower":"2022-07-01T00:00:00.000Z","upper":"2023-01-01T00:00:00.000Z"}]'

因为缺失了 2020-07-01-2021-07-01 这个分区

增加后：

range_partitions = '[{"lower":null,"upper":"2020-07-01T00:00:00.000Z"},
    
    {"lower":"2020-07-01T00:00:00.000Z","upper":"2021-07-01T00:00:00.000Z"},
    
    {"lower":"2021-07-01T00:00:00.000Z","upper":"2022-01-01T00:00:00.000Z"},
    
    {"lower":"2022-01-01T00:00:00.000Z","upper":"2022-07-01T00:00:00.000Z"},
    
    {"lower":"2022-07-01T00:00:00.000Z","upper":"2023-01-01T00:00:00.000Z"}]'

相同错误的博客：[flinksql 写入kudu出现问题的排查\_wppwpp1的博客-CSDN博客\_error while sending value. row error for primary k][flinksql _kudu_wppwpp1_-CSDN_error while sending value. row error for primary k]

我这里使用的是presto，查看建表语句中的分区列表sql如下：

show create table kudu.表schema.表名

添加分区的语句如下

CALL kudu.system.add_range_partition('表schema', '表名', '{"lower": "2020-07-01", "upper": "2021-07-01"}');

更多语法：[什么是Kudu连接器\_开源大数据平台 E-MapReduce-阿里云帮助中心][Kudu_ E-MapReduce-]

[flinksql _kudu_wppwpp1_-CSDN_error while sending value. row error for primary k]: https://blog.csdn.net/wppwpp1/article/details/122438266
[Kudu_ E-MapReduce-]: https://help.aliyun.com/document_detail/444574.htm