ad1happy2go commented on issue #6591:
URL: https://github.com/apache/hudi/issues/6591#issuecomment-1525255395
Issue still exists in master.
Reproducible script -
```
//action1: spark-dataframe write
import org.apache.spark.sql.SaveMode._
import org.apache.hudi.DataSourceReadOptions._
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import scala.collection.mutable
val tableName = "issue_6591_try6"
val basePath = "file:///tmp/issue_6591_try6"
val spark = SparkSession.builder.enableHiveSupport.getOrCreate
import spark.implicits._
// spark-shell
val df = Seq(
("1", "10001", "2022-08-30","2022-08-30 12:00:00.000","2022-08-30"),
("2", "10002", "2022-08-31","2022-08-30 12:00:00.000","2022-08-30"),
("3", "10003", "2022-08-31","2022-08-30 12:00:00.000","2022-08-30"),
("4", "10004", "2022-08-31","2022-08-30 12:00:00.000","2022-08-30"),
("5", "10005", "2022-08-31","2022-08-30 12:00:00.000","2022-08-30"),
("6", "10006", "2022-08-31","2022-08-30 12:00:00.000","2022-08-30")
).toDF("game_schedule_id", "game_id", "game_date_cn", "insert_date",
"dt")
df.createOrReplaceTempView("f_schedule_test")
// df.show()
val hudiOptions = mutable.Map(
"hoodie.table.name" -> tableName,
"hoodie.datasource.write.table.type" -> "MERGE_ON_READ",
"hoodie.datasource.write.operation" -> "upsert",
"hoodie.datasource.write.recordkey.field" -> "game_schedule_id",
"hoodie.datasource.write.precombine.field" -> "insert_date",
"hoodie.datasource.write.partitionpath.field" -> "dt",
"hoodie.index.type" -> "GLOBAL_BLOOM",
"hoodie.compact.inline" -> "true",
"hoodie.datasource.write.keygenerator.class" ->
"org.apache.hudi.keygen.ComplexKeyGenerator"
)
//step1: insert --no issue
df.write.format("hudi").
options(hudiOptions).
mode(Append).
save(basePath)
spark.read.format("hudi").load(basePath).show(false)
spark.read.format("hudi").load(basePath).createOrReplaceTempView("f_schedule_test")
//step2: move part data to another partition --no issue
val df1 = spark.sql("select * from f_schedule_test where dt =
'2022-08-30'").withColumn("dt",lit("2022-08-31")).limit(3)
df1.write.format("hudi").
options(hudiOptions).
mode(Append).
save(basePath)
spark.read.format("hudi").load(basePath).show(false)
//step3: move back --duplicate occurs
//Updating an existing set of rows will result in either a) a companion
log/delta file for an existing base parquet file generated from a previous
compaction or b) an update written to a log/delta file in case no compaction
ever happened for it.
spark.read.format("hudi").load(basePath).createOrReplaceTempView("f_schedule_test")
val df2 = spark.sql("select * from f_schedule_test where dt =
'2022-08-31'").withColumn("dt",lit("2022-08-30")).limit(3)
df2.write.format("hudi").
options(hudiOptions).
mode(Append).
save(basePath)
spark.read.format("hudi").load(basePath).show(false)
```
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
