rangareddy commented on issue #10823:
URL: https://github.com/apache/hudi/issues/10823#issuecomment-4852510993
Hi @Priyanka128
Below is a working example of the code.
`vi /tmp/reproduce_hudi_10823.py`
```python
import sys
from datetime import date
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
def main():
# 1. Initialize Spark Session with Hudi configurations
spark = SparkSession.builder \
.appName("ReproduceHudiIssue10823") \
.config("spark.serializer",
"org.apache.spark.serializer.KryoSerializer") \
.config("spark.sql.extensions",
"org.apache.spark.sql.hudi.HoodieSparkSessionExtension") \
.config("spark.sql.catalog.spark_catalog",
"org.apache.spark.sql.hudi.catalog.HoodieCatalog") \
.getOrCreate()
tableName = "reproduce_10823_hudi_issue"
tablePath = f"s3://warehouse/10823/{tableName}"
columns = ["ts", "uuid", "rider", "driver", "fare", "dt"]
data = [
(1695159649087, "334e26e9-8355-45cc-97c6-c31daf0df330", "rider-A",
"driver-K", 19.10, "2012-01-01"),
(1695091554788, "e96c4396-3fad-413a-a942-4cb36106d721", "rider-B",
"driver-L", 27.70, "2012-01-01"),
(1695046462179, "9909a8b1-2d15-4d3d-8ec9-efc48c536a00", "rider-C",
"driver-M", 33.90, "2012-01-01"),
(1695516137016, "e3cf430c-889d-4015-bc98-59bdce1e530c", "rider-C",
"driver-N", 34.15, "2012-01-01")
]
insertsDF = spark.createDataFrame(data).toDF(*columns)
hudi_options = {
'hoodie.table.name': tableName,
'hoodie.datasource.write.recordkey.field': 'uuid',
'hoodie.datasource.write.precombine.field': 'ts',
'hoodie.datasource.write.partitionpath.field': 'dt',
'hoodie.datasource.write.keygenerator.consistent.logical.timestamp.enabled':
'true',
'hoodie.datasource.write.keygenerator.class':
'org.apache.hudi.keygen.TimestampBasedKeyGenerator',
'hoodie.keygen.timebased.timestamp.type': 'DATE_STRING',
'hoodie.keygen.timebased.input.dateformat': 'yyyy-MM-dd',
'hoodie.keygen.timebased.output.dateformat': 'yyyy-MM-dd',
'hoodie.datasource.write.hive_style_partitioning': 'true'
}
# ------- 2. Upsert (Overwrite Mode) -------
print(">>> Performing Upsert...")
insertsDF.write.format("hudi").options(**hudi_options) \
.option("hoodie.datasource.write.operation", "upsert") \
.mode("overwrite").save(tablePath)
# ------- 3. Read and Show Active Dataset -------
print(">>> Active Dataset Layout:")
deleteDF = spark.read.format("hudi").load(tablePath).select("ts",
"uuid", "rider", "driver", "fare", "dt")
deleteDF.show()
# ------- 4. Delete Operation -------
print(">>> Executing Records Deletion...")
deleteDF.write.format("hudi").options(**hudi_options) \
.option("hoodie.datasource.write.operation", "delete") \
.mode("append").save(tablePath)
# ------- 5. Verify Results Post-Deletion -------
print(">>> Verifying Table Records After Deletion Block:")
spark.read.format("hudi").load(tablePath).show()
# Stop Spark Session elegantly
spark.stop()
if __name__ == "__main__":
main()
```
**Using Hudi 0.14.0**
```sh
spark-submit \
--master local[*] \
--name "ReproduceHudiIssue10823" \
--packages org.apache.hudi:hudi-spark3.4-bundle_2.12:0.14.0 \
--conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
--conf
spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension \
--conf
spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog
\
/tmp/reproduce_hudi_10823.py
```
**Using Hudi 1.1.1**
```sh
spark-submit \
--master local[*] \
--name "ReproduceHudiIssue10823" \
--packages org.apache.hudi:hudi-spark3.4-bundle_2.12:1.1.1 \
--conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
--conf
spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension \
--conf
spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog
\
/tmp/reproduce_hudi_10823.py
```
--
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]
