You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2020/08/19 11:09:07 UTC
[GitHub] [hudi] piyushrl opened a new issue #1985: [SUPPORT]Error while running deltastreamer on top of backfilled data using Hudi
piyushrl opened a new issue #1985:
URL: https://github.com/apache/hudi/issues/1985
We have backfilled data using df.write.format(“hudi”) from db and then in same location and hive table trying to upsert new data using deltastreamer from kafka. We are facing below error when deltastreamer tries to upsert new data once backfill is done.
`20/08/19 15:21:31 ERROR deltastreamer.HoodieDeltaStreamer: Got error running delta sync once. Shutting down
org.apache.hudi.utilities.exception.HoodieDeltaStreamerException: Unable to find previous checkpoint. Please double check if this table was indeed built via delta streamer. Last Commit :Option{val=[20200819143204__deltacommit__COMPLETED]}, Instants :[[20200819143204__deltacommit__COMPLETED]], CommitMetadata={
"partitionToWriteStats" : {
"2019-06-16" : [ {
"fileId" : "da81b7ed-296b-4c3a-b3fe-823ded3e2e5f-0",
"path" : "2019-06-16/da81b7ed-296b-4c3a-b3fe-823ded3e2e5f-0_0-20-22_20200819143204.parquet",
"prevCommit" : "null",
"numWrites" : 6,
"numDeletes" : 0,
"numUpdateWrites" : 0,
"numInserts" : 6,
"totalWriteBytes" : 461439,
"totalWriteErrors" : 0,
"tempPath" : null,
"partitionPath" : "2019-06-16",
"totalLogRecords" : 0,
"totalLogFilesCompacted" : 0,
"totalLogSizeCompacted" : 0,
"totalUpdatedRecordsCompacted" : 0,
"totalLogBlocks" : 0,
"totalCorruptLogBlock" : 0,
"totalRollbackBlocks" : 0,
"fileSizeInBytes" : 461439
} ],
"2019-06-17" : [ {
"fileId" : "423f6ccc-2391-4565-bf2e-ef9fc1354e6d-0",
"path" : "2019-06-17/423f6ccc-2391-4565-bf2e-ef9fc1354e6d-0_1-20-23_20200819143204.parquet",
"prevCommit" : "null",
"numWrites" : 2251,
"numDeletes" : 0,
"numUpdateWrites" : 0,
"numInserts" : 2251,
"totalWriteBytes" : 787516,
"totalWriteErrors" : 0,
"tempPath" : null,
"partitionPath" : "2019-06-17",
"totalLogRecords" : 0,
"totalLogFilesCompacted" : 0,
"totalLogSizeCompacted" : 0,
"totalUpdatedRecordsCompacted" : 0,
"totalLogBlocks" : 0,
"totalCorruptLogBlock" : 0,
"totalRollbackBlocks" : 0,
"fileSizeInBytes" : 787516
} ]
},
"compacted" : false,
"extraMetadata" : {
"ROLLING_STAT" : "{\n \"partitionToRollingStats\" : {\n \"2019-06-16\" : {\n \"da81b7ed-296b-4c3a-b3fe-823ded3e2e5f-0\" : {\n \"fileId\" : \"da81b7ed-296b-4c3a-b3fe-823ded3e2e5f-0\",\n \"inserts\" : 6,\n \"upserts\" : 0,\n \"deletes\" : 0,\n \"totalInputWriteBytesToDisk\" : 0,\n \"totalInputWriteBytesOnDisk\" : 461439\n }\n },\n \"2019-06-17\" : {\n \"423f6ccc-2391-4565-bf2e-ef9fc1354e6d-0\" : {\n \"fileId\" : \"423f6ccc-2391-4565-bf2e-ef9fc1354e6d-0\",\n \"inserts\" : 2251,\n \"upserts\" : 0,\n \"deletes\" : 0,\n \"totalInputWriteBytesToDisk\" : 0,\n \"totalInputWriteBytesOnDisk\" : 787516\n }\n }\n },\n \"actionType\" : \"deltacommit\"\n}",
"schema" : "<SCHEMA>"
},
"operationType" : "UPSERT",
"fileIdAndRelativePaths" : {
"da81b7ed-296b-4c3a-b3fe-823ded3e2e5f-0" : "2019-06-16/da81b7ed-296b-4c3a-b3fe-823ded3e2e5f-0_0-20-22_20200819143204.parquet",
"423f6ccc-2391-4565-bf2e-ef9fc1354e6d-0" : "2019-06-17/423f6ccc-2391-4565-bf2e-ef9fc1354e6d-0_1-20-23_20200819143204.parquet"
},
"totalRecordsDeleted" : 0,
"totalLogRecordsCompacted" : 0,
"totalScanTime" : 0,
"totalCreateTime" : 0,
"totalUpsertTime" : 0,
"totalCompactedRecordsUpdated" : 0,
"totalLogFilesCompacted" : 0,
"totalLogFilesSize" : 0
}
at org.apache.hudi.utilities.deltastreamer.DeltaSync.readFromSource(DeltaSync.java:268)
at org.apache.hudi.utilities.deltastreamer.DeltaSync.syncOnce(DeltaSync.java:226)
at org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer.sync(HoodieDeltaStreamer.java:121)
at org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer.main(HoodieDeltaStreamer.java:294)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.spark.deploy.yarn.ApplicationMaster$$anon$2.run(ApplicationMaster.scala:688)`
Since Autocommit is false by default in hudi-spark method using df.write.format(“hudi”), I tried changing the withAutoCommit(false) to withAutoCommit(true) in DataSourceUtil.java creating the HoodieWriteConfig object, but then in backfill job we got FileAlreadyExistsException for .deltacommit file. I tired this in a new directory and still got this issue.
`20/08/19 11:29:07 ERROR yarn.ApplicationMaster: User class threw exception: org.apache.hudi.exception.HoodieIOException: Failed to create file hdfs:/user/hive/warehouse/test.db/tbl_test/.hoodie/20200819112849.deltacommit
org.apache.hudi.exception.HoodieIOException: Failed to create file hdfs:/user/hive/warehouse/test.db/tbl_test/.hoodie/20200819112849.deltacommit
at org.apache.hudi.common.table.timeline.HoodieActiveTimeline.createImmutableFileInPath(HoodieActiveTimeline.java:437)
at org.apache.hudi.common.table.timeline.HoodieActiveTimeline.transitionState(HoodieActiveTimeline.java:327)
at org.apache.hudi.common.table.timeline.HoodieActiveTimeline.saveAsComplete(HoodieActiveTimeline.java:145)
at org.apache.hudi.client.AbstractHoodieWriteClient.commit(AbstractHoodieWriteClient.java:165)
at org.apache.hudi.client.AbstractHoodieWriteClient.commit(AbstractHoodieWriteClient.java:111)
at org.apache.hudi.client.AbstractHoodieWriteClient.commit(AbstractHoodieWriteClient.java:102)
at org.apache.hudi.HoodieSparkSqlWriter$.checkWriteStatus(HoodieSparkSqlWriter.scala:259)
at org.apache.hudi.HoodieSparkSqlWriter$.write(HoodieSparkSqlWriter.scala:181)
at org.apache.hudi.DefaultSource.createRelation(DefaultSource.scala:108)
at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:45)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:70)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:68)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.doExecute(commands.scala:86)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:131)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:127)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:155)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:127)
at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:80)
at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:80)
at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:668)
at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:668)
at org.apache.spark.sql.execution.SQLExecution$$anonfun$withNewExecutionId$1.apply(SQLExecution.scala:78)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:125)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:73)
at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:668)
at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:276)
at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:270)
at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:228)
at org.test.dp.connectors.HudiSink$.writeToHudi(HudiSink.scala:34)
at org.test.dp.HistoryBackfillSparkJob$.main(HistoryBackfillSparkJob.scala:47)
at org.test.dp.HistoryBackfillSparkJob.main(HistoryBackfillSparkJob.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.spark.deploy.yarn.ApplicationMaster$$anon$2.run(ApplicationMaster.scala:688)
Caused by: org.apache.hadoop.fs.FileAlreadyExistsException: /user/hive/warehouse/test.db/tbl_test/.hoodie/20200819112849.deltacommit for client <ip> already exists
at org.apache.hadoop.hdfs.server.namenode.FSDirWriteFileOp.startFile(FSDirWriteFileOp.java:388)`
Is there any way that once i backfill data using hudi-spark, in that same hdfs location and hive table, deltastreamer can read metadata or something and start daily upsert from kafka
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] bvaradar commented on issue #1985: [SUPPORT]Error while running deltastreamer on top of backfilled data using Hudi
Posted by GitBox <gi...@apache.org>.
bvaradar commented on issue #1985:
URL: https://github.com/apache/hudi/issues/1985#issuecomment-691777089
Added https://issues.apache.org/jira/browse/HUDI-1280 for smooth transition to deltastreamer.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] bvaradar closed issue #1985: [SUPPORT]Error while running deltastreamer on top of backfilled data using Hudi
Posted by GitBox <gi...@apache.org>.
bvaradar closed issue #1985:
URL: https://github.com/apache/hudi/issues/1985
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] bvaradar commented on issue #1985: [SUPPORT]Error while running deltastreamer on top of backfilled data using Hudi
Posted by GitBox <gi...@apache.org>.
bvaradar commented on issue #1985:
URL: https://github.com/apache/hudi/issues/1985#issuecomment-678044206
@piyushrl :
The strategy would be to orchestrate this bootstrap and handoff in 3 steps
1. Copy the earliest checkpoint from kafka and save it after ensuring your upstream source is up to date till the checkpoint saved.
2. Perform spark.write() to hudi.
3. Run deltastreamer in single run mode and pass the checkpoint (1) as part of --checkpoint command line argument.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] bvaradar commented on issue #1985: [SUPPORT]Error while running deltastreamer on top of backfilled data using Hudi
Posted by GitBox <gi...@apache.org>.
bvaradar commented on issue #1985:
URL: https://github.com/apache/hudi/issues/1985#issuecomment-676387943
For switching from (unmanaged)spark.write() to deltastreamer, you need to provide the checkpoint explicitly to deltastreamer either by passing --initial-checkpoint-provider or --checkpoint.
To get an example checkpoint, you can start deltastreamer on a brand new table and have it complete one commit and inspect the .commit/.deltacommit file for the checkpoint
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org