Skip to content

Instantly share code, notes, and snippets.

@nsivabalan

nsivabalan/insert -> upsert DuplicateKey

Last active August 5, 2021 21:24
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 1 You must be signed in to fork a gist
  • Save nsivabalan/9837a90b1481c479a9c600bf16bafa57 to your computer and use it in GitHub Desktop.
Save nsivabalan/9837a90b1481c479a9c600bf16bafa57 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
insert -> upsert DuplicateKey
create table hudi_test4 (id int, name string, price double, ts long) using hudi options(primaryKey = 'id', precombineField = 'ts') partitioned by (name) location 'file:///tmp/hudi_testing/hudi_test4';
insert into hudi_test4 values(1, 100.0, 100000010, "abc");
insert into hudi_test4 values(2, 200.0, 200000010, "abc");
insert into hudi_test4 values(3, 300.0, 300000010, "abc");
// this throws DuplicateKeyException ? Even though operation w/ hudi is "upsert".
insert into hudi_test4 values(1, 600.0, 600000010, "abc");
Raw
stacktrace
> insert into hudi_test4 values(1, 600.0, 600000010, "abc");
21/08/05 17:01:41 WARN HoodieParquetRealtimeInputFormat: TEST_LOG. Operation type:: upsert
21/08/05 17:01:47 ERROR BoundedInMemoryExecutor: error consuming records
org.apache.hudi.exception.HoodieUpsertException: Failed to combine/merge new record with old value in storage, for new record {HoodieRecord{key=HoodieKey { recordKey=id:1 partitionPath=name=abc}, currentLocation='HoodieRecordLocation {instantTime=20210805164443, fileId=da4fe886-455c-491a-8aae-b1b68fc33bc8-0}', newLocation='HoodieRecordLocation {instantTime=20210805170141, fileId=da4fe886-455c-491a-8aae-b1b68fc33bc8-0}'}}, old value {{"_hoodie_commit_time": "20210805164218", "_hoodie_commit_seqno": "20210805164218_0_27", "_hoodie_record_key": "id:1", "_hoodie_partition_path": "name=abc", "_hoodie_file_name": "da4fe886-455c-491a-8aae-b1b68fc33bc8-0_0-500-15604_20210805164218.parquet", "id": 1, "price": 100.0, "ts": 100000010, "name": "abc"}}
at org.apache.hudi.io.HoodieMergeHandle.write(HoodieMergeHandle.java:321)
at org.apache.hudi.table.action.commit.AbstractMergeHelper$UpdateHandler.consumeOneRecord(AbstractMergeHelper.java:122)
at org.apache.hudi.table.action.commit.AbstractMergeHelper$UpdateHandler.consumeOneRecord(AbstractMergeHelper.java:112)
at org.apache.hudi.common.util.queue.BoundedInMemoryQueueConsumer.consume(BoundedInMemoryQueueConsumer.java:37)
at org.apache.hudi.common.util.queue.BoundedInMemoryExecutor.lambda$null$2(BoundedInMemoryExecutor.java:121)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.hudi.exception.HoodieDuplicateKeyException: Duplicate key found for insert statement, key is: id:1
at org.apache.spark.sql.hudi.command.ValidateDuplicateKeyPayload.combineAndGetUpdateValue(InsertIntoHoodieTableCommand.scala:285)
at org.apache.hudi.io.HoodieMergeHandle.write(HoodieMergeHandle.java:301)
... 8 more
21/08/05 17:01:47 ERROR BaseSparkCommitActionExecutor: Error upserting bucketType UPDATE for partition :0
org.apache.hudi.exception.HoodieException: org.apache.hudi.exception.HoodieException: java.util.concurrent.ExecutionException: org.apache.hudi.exception.HoodieUpsertException: Failed to combine/merge new record with old value in storage, for new record {HoodieRecord{key=HoodieKey { recordKey=id:1 partitionPath=name=abc}, currentLocation='HoodieRecordLocation {instantTime=20210805164443, fileId=da4fe886-455c-491a-8aae-b1b68fc33bc8-0}', newLocation='HoodieRecordLocation {instantTime=20210805170141, fileId=da4fe886-455c-491a-8aae-b1b68fc33bc8-0}'}}, old value {{"_hoodie_commit_time": "20210805164218", "_hoodie_commit_seqno": "20210805164218_0_27", "_hoodie_record_key": "id:1", "_hoodie_partition_path": "name=abc", "_hoodie_file_name": "da4fe886-455c-491a-8aae-b1b68fc33bc8-0_0-500-15604_20210805164218.parquet", "id": 1, "price": 100.0, "ts": 100000010, "name": "abc"}}
at org.apache.hudi.table.action.commit.SparkMergeHelper.runMerge(SparkMergeHelper.java:102)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpdateInternal(BaseSparkCommitActionExecutor.java:334)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpdate(BaseSparkCommitActionExecutor.java:325)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpsertPartition(BaseSparkCommitActionExecutor.java:298)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.lambda$execute$ecf5068c$1(BaseSparkCommitActionExecutor.java:156)
at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsWithIndex$1(JavaRDDLike.scala:102)
at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsWithIndex$1$adapted(JavaRDDLike.scala:102)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2(RDD.scala:889)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2$adapted(RDD.scala:889)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:349)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:313)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:349)
at org.apache.spark.rdd.RDD.$anonfun$getOrCompute$1(RDD.scala:362)
at org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1388)
at org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1298)
at org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1362)
at org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1186)
at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:360)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:311)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:349)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:313)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:127)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:446)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:449)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.hudi.exception.HoodieException: java.util.concurrent.ExecutionException: org.apache.hudi.exception.HoodieUpsertException: Failed to combine/merge new record with old value in storage, for new record {HoodieRecord{key=HoodieKey { recordKey=id:1 partitionPath=name=abc}, currentLocation='HoodieRecordLocation {instantTime=20210805164443, fileId=da4fe886-455c-491a-8aae-b1b68fc33bc8-0}', newLocation='HoodieRecordLocation {instantTime=20210805170141, fileId=da4fe886-455c-491a-8aae-b1b68fc33bc8-0}'}}, old value {{"_hoodie_commit_time": "20210805164218", "_hoodie_commit_seqno": "20210805164218_0_27", "_hoodie_record_key": "id:1", "_hoodie_partition_path": "name=abc", "_hoodie_file_name": "da4fe886-455c-491a-8aae-b1b68fc33bc8-0_0-500-15604_20210805164218.parquet", "id": 1, "price": 100.0, "ts": 100000010, "name": "abc"}}
at org.apache.hudi.common.util.queue.BoundedInMemoryExecutor.execute(BoundedInMemoryExecutor.java:147)
at org.apache.hudi.table.action.commit.SparkMergeHelper.runMerge(SparkMergeHelper.java:100)
... 31 more
Caused by: java.util.concurrent.ExecutionException: org.apache.hudi.exception.HoodieUpsertException: Failed to combine/merge new record with old value in storage, for new record {HoodieRecord{key=HoodieKey { recordKey=id:1 partitionPath=name=abc}, currentLocation='HoodieRecordLocation {instantTime=20210805164443, fileId=da4fe886-455c-491a-8aae-b1b68fc33bc8-0}', newLocation='HoodieRecordLocation {instantTime=20210805170141, fileId=da4fe886-455c-491a-8aae-b1b68fc33bc8-0}'}}, old value {{"_hoodie_commit_time": "20210805164218", "_hoodie_commit_seqno": "20210805164218_0_27", "_hoodie_record_key": "id:1", "_hoodie_partition_path": "name=abc", "_hoodie_file_name": "da4fe886-455c-491a-8aae-b1b68fc33bc8-0_0-500-15604_20210805164218.parquet", "id": 1, "price": 100.0, "ts": 100000010, "name": "abc"}}
at java.util.concurrent.FutureTask.report(FutureTask.java:122)
at java.util.concurrent.FutureTask.get(FutureTask.java:192)
at org.apache.hudi.common.util.queue.BoundedInMemoryExecutor.execute(BoundedInMemoryExecutor.java:141)
... 32 more
Caused by: org.apache.hudi.exception.HoodieUpsertException: Failed to combine/merge new record with old value in storage, for new record {HoodieRecord{key=HoodieKey { recordKey=id:1 partitionPath=name=abc}, currentLocation='HoodieRecordLocation {instantTime=20210805164443, fileId=da4fe886-455c-491a-8aae-b1b68fc33bc8-0}', newLocation='HoodieRecordLocation {instantTime=20210805170141, fileId=da4fe886-455c-491a-8aae-b1b68fc33bc8-0}'}}, old value {{"_hoodie_commit_time": "20210805164218", "_hoodie_commit_seqno": "20210805164218_0_27", "_hoodie_record_key": "id:1", "_hoodie_partition_path": "name=abc", "_hoodie_file_name": "da4fe886-455c-491a-8aae-b1b68fc33bc8-0_0-500-15604_20210805164218.parquet", "id": 1, "price": 100.0, "ts": 100000010, "name": "abc"}}
at org.apache.hudi.io.HoodieMergeHandle.write(HoodieMergeHandle.java:321)
at org.apache.hudi.table.action.commit.AbstractMergeHelper$UpdateHandler.consumeOneRecord(AbstractMergeHelper.java:122)
at org.apache.hudi.table.action.commit.AbstractMergeHelper$UpdateHandler.consumeOneRecord(AbstractMergeHelper.java:112)
at org.apache.hudi.common.util.queue.BoundedInMemoryQueueConsumer.consume(BoundedInMemoryQueueConsumer.java:37)
at org.apache.hudi.common.util.queue.BoundedInMemoryExecutor.lambda$null$2(BoundedInMemoryExecutor.java:121)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
... 3 more
Caused by: org.apache.hudi.exception.HoodieDuplicateKeyException: Duplicate key found for insert statement, key is: id:1
at org.apache.spark.sql.hudi.command.ValidateDuplicateKeyPayload.combineAndGetUpdateValue(InsertIntoHoodieTableCommand.scala:285)
at org.apache.hudi.io.HoodieMergeHandle.write(HoodieMergeHandle.java:301)
... 8 more
21/08/05 17:01:47 WARN BlockManager: Putting block rdd_68_0 failed due to exception org.apache.hudi.exception.HoodieUpsertException: Error upserting bucketType UPDATE for partition :0.
21/08/05 17:01:47 WARN BlockManager: Block rdd_68_0 could not be removed as it was not found on disk or in memory
21/08/05 17:01:47 ERROR Executor: Exception in task 0.0 in stage 29.0 (TID 1820)
org.apache.hudi.exception.HoodieUpsertException: Error upserting bucketType UPDATE for partition :0
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpsertPartition(BaseSparkCommitActionExecutor.java:305)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.lambda$execute$ecf5068c$1(BaseSparkCommitActionExecutor.java:156)
at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsWithIndex$1(JavaRDDLike.scala:102)
at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsWithIndex$1$adapted(JavaRDDLike.scala:102)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2(RDD.scala:889)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2$adapted(RDD.scala:889)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:349)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:313)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:349)
at org.apache.spark.rdd.RDD.$anonfun$getOrCompute$1(RDD.scala:362)
at org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1388)
at org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1298)
at org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1362)
at org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1186)
at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:360)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:311)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:349)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:313)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:127)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:446)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:449)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.hudi.exception.HoodieException: org.apache.hudi.exception.HoodieException: java.util.concurrent.ExecutionException: org.apache.hudi.exception.HoodieUpsertException: Failed to combine/merge new record with old value in storage, for new record {HoodieRecord{key=HoodieKey { recordKey=id:1 partitionPath=name=abc}, currentLocation='HoodieRecordLocation {instantTime=20210805164443, fileId=da4fe886-455c-491a-8aae-b1b68fc33bc8-0}', newLocation='HoodieRecordLocation {instantTime=20210805170141, fileId=da4fe886-455c-491a-8aae-b1b68fc33bc8-0}'}}, old value {{"_hoodie_commit_time": "20210805164218", "_hoodie_commit_seqno": "20210805164218_0_27", "_hoodie_record_key": "id:1", "_hoodie_partition_path": "name=abc", "_hoodie_file_name": "da4fe886-455c-491a-8aae-b1b68fc33bc8-0_0-500-15604_20210805164218.parquet", "id": 1, "price": 100.0, "ts": 100000010, "name": "abc"}}
at org.apache.hudi.table.action.commit.SparkMergeHelper.runMerge(SparkMergeHelper.java:102)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpdateInternal(BaseSparkCommitActionExecutor.java:334)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpdate(BaseSparkCommitActionExecutor.java:325)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpsertPartition(BaseSparkCommitActionExecutor.java:298)
... 28 more
Caused by: org.apache.hudi.exception.HoodieException: java.util.concurrent.ExecutionException: org.apache.hudi.exception.HoodieUpsertException: Failed to combine/merge new record with old value in storage, for new record {HoodieRecord{key=HoodieKey { recordKey=id:1 partitionPath=name=abc}, currentLocation='HoodieRecordLocation {instantTime=20210805164443, fileId=da4fe886-455c-491a-8aae-b1b68fc33bc8-0}', newLocation='HoodieRecordLocation {instantTime=20210805170141, fileId=da4fe886-455c-491a-8aae-b1b68fc33bc8-0}'}}, old value {{"_hoodie_commit_time": "20210805164218", "_hoodie_commit_seqno": "20210805164218_0_27", "_hoodie_record_key": "id:1", "_hoodie_partition_path": "name=abc", "_hoodie_file_name": "da4fe886-455c-491a-8aae-b1b68fc33bc8-0_0-500-15604_20210805164218.parquet", "id": 1, "price": 100.0, "ts": 100000010, "name": "abc"}}
at org.apache.hudi.common.util.queue.BoundedInMemoryExecutor.execute(BoundedInMemoryExecutor.java:147)
at org.apache.hudi.table.action.commit.SparkMergeHelper.runMerge(SparkMergeHelper.java:100)
... 31 more
Caused by: java.util.concurrent.ExecutionException: org.apache.hudi.exception.HoodieUpsertException: Failed to combine/merge new record with old value in storage, for new record {HoodieRecord{key=HoodieKey { recordKey=id:1 partitionPath=name=abc}, currentLocation='HoodieRecordLocation {instantTime=20210805164443, fileId=da4fe886-455c-491a-8aae-b1b68fc33bc8-0}', newLocation='HoodieRecordLocation {instantTime=20210805170141, fileId=da4fe886-455c-491a-8aae-b1b68fc33bc8-0}'}}, old value {{"_hoodie_commit_time": "20210805164218", "_hoodie_commit_seqno": "20210805164218_0_27", "_hoodie_record_key": "id:1", "_hoodie_partition_path": "name=abc", "_hoodie_file_name": "da4fe886-455c-491a-8aae-b1b68fc33bc8-0_0-500-15604_20210805164218.parquet", "id": 1, "price": 100.0, "ts": 100000010, "name": "abc"}}
at java.util.concurrent.FutureTask.report(FutureTask.java:122)
at java.util.concurrent.FutureTask.get(FutureTask.java:192)
at org.apache.hudi.common.util.queue.BoundedInMemoryExecutor.execute(BoundedInMemoryExecutor.java:141)
... 32 more
Caused by: org.apache.hudi.exception.HoodieUpsertException: Failed to combine/merge new record with old value in storage, for new record {HoodieRecord{key=HoodieKey { recordKey=id:1 partitionPath=name=abc}, currentLocation='HoodieRecordLocation {instantTime=20210805164443, fileId=da4fe886-455c-491a-8aae-b1b68fc33bc8-0}', newLocation='HoodieRecordLocation {instantTime=20210805170141, fileId=da4fe886-455c-491a-8aae-b1b68fc33bc8-0}'}}, old value {{"_hoodie_commit_time": "20210805164218", "_hoodie_commit_seqno": "20210805164218_0_27", "_hoodie_record_key": "id:1", "_hoodie_partition_path": "name=abc", "_hoodie_file_name": "da4fe886-455c-491a-8aae-b1b68fc33bc8-0_0-500-15604_20210805164218.parquet", "id": 1, "price": 100.0, "ts": 100000010, "name": "abc"}}
at org.apache.hudi.io.HoodieMergeHandle.write(HoodieMergeHandle.java:321)
at org.apache.hudi.table.action.commit.AbstractMergeHelper$UpdateHandler.consumeOneRecord(AbstractMergeHelper.java:122)
at org.apache.hudi.table.action.commit.AbstractMergeHelper$UpdateHandler.consumeOneRecord(AbstractMergeHelper.java:112)
at org.apache.hudi.common.util.queue.BoundedInMemoryQueueConsumer.consume(BoundedInMemoryQueueConsumer.java:37)
at org.apache.hudi.common.util.queue.BoundedInMemoryExecutor.lambda$null$2(BoundedInMemoryExecutor.java:121)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
... 3 more
Caused by: org.apache.hudi.exception.HoodieDuplicateKeyException: Duplicate key found for insert statement, key is: id:1
at org.apache.spark.sql.hudi.command.ValidateDuplicateKeyPayload.combineAndGetUpdateValue(InsertIntoHoodieTableCommand.scala:285)
at org.apache.hudi.io.HoodieMergeHandle.write(HoodieMergeHandle.java:301)
... 8 more
21/08/05 17:01:47 WARN TaskSetManager: Lost task 0.0 in stage 29.0 (TID 1820, 10.0.0.202, executor driver): org.apache.hudi.exception.HoodieUpsertException: Error upserting bucketType UPDATE for partition :0
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpsertPartition(BaseSparkCommitActionExecutor.java:305)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.lambda$execute$ecf5068c$1(BaseSparkCommitActionExecutor.java:156)
at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsWithIndex$1(JavaRDDLike.scala:102)
at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsWithIndex$1$adapted(JavaRDDLike.scala:102)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2(RDD.scala:889)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2$adapted(RDD.scala:889)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:349)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:313)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:349)
at org.apache.spark.rdd.RDD.$anonfun$getOrCompute$1(RDD.scala:362)
at org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1388)
at org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1298)
at org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1362)
at org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1186)
at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:360)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:311)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:349)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:313)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:127)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:446)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:449)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.hudi.exception.HoodieException: org.apache.hudi.exception.HoodieException: java.util.concurrent.ExecutionException: org.apache.hudi.exception.HoodieUpsertException: Failed to combine/merge new record with old value in storage, for new record {HoodieRecord{key=HoodieKey { recordKey=id:1 partitionPath=name=abc}, currentLocation='HoodieRecordLocation {instantTime=20210805164443, fileId=da4fe886-455c-491a-8aae-b1b68fc33bc8-0}', newLocation='HoodieRecordLocation {instantTime=20210805170141, fileId=da4fe886-455c-491a-8aae-b1b68fc33bc8-0}'}}, old value {{"_hoodie_commit_time": "20210805164218", "_hoodie_commit_seqno": "20210805164218_0_27", "_hoodie_record_key": "id:1", "_hoodie_partition_path": "name=abc", "_hoodie_file_name": "da4fe886-455c-491a-8aae-b1b68fc33bc8-0_0-500-15604_20210805164218.parquet", "id": 1, "price": 100.0, "ts": 100000010, "name": "abc"}}
at org.apache.hudi.table.action.commit.SparkMergeHelper.runMerge(SparkMergeHelper.java:102)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment