# การ Save Data ด้วย pyspark หลังจาก drop duplicate

**URL:** <https://discuss.dataengineercafe.io/t/save-data-pyspark-drop-duplicate/328>\
**Category:** Spark\
**Tags:** python, data-pipeline, pyspark, databricks, spark\
**Created:** [July 8, 2022, 3:34pm UTC](https://discuss.dataengineercafe.io/t/save-data-pyspark-drop-duplicate/328 "2022-07-08T15:34:01Z")\
**Posts on this page:** 9\
**Page:** 1

<div class="post-metadata">

**Author:** ![premsurawut](https://avatars.discourse-cdn.com/v4/letter/p/df788c/32.png) [@premsurawut](https://discuss.dataengineercafe.io/u/premsurawut)\
**Post date:** [July 8, 2022, 3:34pm UTC](https://discuss.dataengineercafe.io/t/save-data-pyspark-drop-duplicate/328/1 "2022-07-08T15:34:01Z")

</div>

สวัสดีครับ อยากจะขอถาม use case นิดนึงครับ ผมเพิ่งจะเริ่มมาจัดการ data จริงจังหลังจากย้ายมาทำ Data engineer และเพิ่งเริ่มใช้ Spark and Databricks ครับ

ในกรณีที่ อ่านข้อมูลที่อยู่ใน parquet format ซึ่งมี data duplicates อยู่ 2 rows(อ่านข้อมูลจาก blob และ filter ตาม condition) หลังจากจัดการ drop duplicates แล้ว ก็จะเหลือแค่ row เดียว ทีนี้เราจะเซฟข้อมูลกลับไปที่เดิมยังไงครับ ที่ไปอ่านใน document ของ databricks/pyspark ก็จะมีการ append / overwrite ซึ่งผมเข้าใจว่าถ้า overwrite ด้วยข้อมูล row เดียวมันจะไปเอา content นี้ไปแทนที่ข้อมูลทั้งหมดใน blob ใช่หรือป่าวครับ ส่วน append จะเพิ่ม row นั้นเข้าไปใน table ที่หรือ dir ที่มีอยู่แล้ว ไม่แน่ใจว่ามันเป็นการเพิ่ม duplicate เข้าไปอีกทีไหมครับ

ช่วยแชร์ทีครับว่าครรทำแบบไหน หรือมีวิธีอื่นเพิ่มเติมไหมครับ ผมกังวลเรื่อง data loss เลยพยายามทำความเข้าใจตรงนี้อยู่ครับ

---

<div class="post-metadata">

**Author:** ![zkan](https://yyz1.discourse-cdn.com/flex035/user_avatar/discuss.dataengineercafe.io/zkan/32/2_2.png) [@zkan](https://discuss.dataengineercafe.io/u/zkan)\
**Post date:** [July 9, 2022, 1:49am UTC](https://discuss.dataengineercafe.io/t/save-data-pyspark-drop-duplicate/328/2 "2022-07-09T01:49:09Z")

</div>

เข้าใจถูกแล้วครับ overwrite คือทับลงไปที่ไฟล์เดิม ซึ่งเราจะเสียความเป็น origin ไป แล้วก็ append เป็นการเพิ่ม row เข้าไปครับ ซึ่งก็จะเพิ่ม duplicate เข้าไปอีกที

ตอนนี้ผมใช้วิธีแยก 2 folders (หรือ 2 buckets) ครับ โดยที่ folder แรกเก็บ data ที่มาจาก source จริง ๆ เลย ออกแนวมาแบบไหนก็เก็บประมาณนั้น ก็จะมี duplicates อยู่ในนั้น ถ้าเป็น parquet อยู่แล้วก็ดีเลย แล้วก็ transform + dedup เอามาเก็บอีก folder หนึ่งครับผม

ประมาณนี้

![raw-to-staging](https://canada1.discourse-cdn.com/flex035/uploads/dataengineercafe/original/1X/1be7848fd3ee581641e7fc4b0bed54f7b698e3b1.png)

เดี๋ยวรอเพื่อน ๆ ท่านอื่นมาเสริมเพิ่มเติม 😊

ปล. pts-ark นี่คือ pyspark ใช่เปล่านะ o\_O

---

<div class="post-metadata">

**Author:** ![premsurawut](https://avatars.discourse-cdn.com/v4/letter/p/df788c/32.png) [@premsurawut](https://discuss.dataengineercafe.io/u/premsurawut)\
**Post date:** [July 9, 2022, 3:39am UTC](https://discuss.dataengineercafe.io/t/save-data-pyspark-drop-duplicate/328/3 "2022-07-09T03:39:22Z")

</div>

ขอบคุณมากครับ ช่วยคลายความสงสัยไปเลยครับ แล้วก็ขอบคุณที่ช่วยแชร์ use case ให้นะครับ กำลังจะต้องเริ่มทำ pipeline ด้วยเลยครับ

ปล. แก้หัวข้อแล้วครับ แหะๆ เมื่อคืนน่าจะเบลอครับ อ่านแต่ docs ทั้งวันเลยครับ

---

<div class="post-metadata">

**Author:** ![kahnwong](https://yyz1.discourse-cdn.com/flex035/user_avatar/discuss.dataengineercafe.io/kahnwong/32/295_2.png) [@kahnwong](https://discuss.dataengineercafe.io/u/kahnwong)\
**Post date:** [July 9, 2022, 6:46am UTC](https://discuss.dataengineercafe.io/t/save-data-pyspark-drop-duplicate/328/4 "2022-07-09T06:46:00Z")

</div>

ควรจะเขียนลงที่ใหม่ ไม่ควรเขียนทับที่เดิม

อ้อ แล้วก็ระวังเรื่อง partition size ด้วย  
uneven spark partitions มีผลกับ performance  
แล้วก็ partition size ควรจะมีขนาดใหญ่พอประมาณนึง ไม่งั้นมันจะไม่คุ้มกะ spark overhead (อย่างต่ำขอประมาณ 10-20mb per partition)

---

<div class="post-metadata">

**Author:** ![premsurawut](https://avatars.discourse-cdn.com/v4/letter/p/df788c/32.png) [@premsurawut](https://discuss.dataengineercafe.io/u/premsurawut)\
**Post date:** [July 9, 2022, 6:51am UTC](https://discuss.dataengineercafe.io/t/save-data-pyspark-drop-duplicate/328/5 "2022-07-09T06:51:04Z")

</div>

ขอถามเพิ่มเติมอีกนิดครับ กรณีที่ถ้าอยากจะเซฟไว้ที่เดิม

1. ถ้าผมอ่านเป็น delta table ขึ้น มาจากนั้น
2. อ่านข้อมูลอีกชุดที่อ่านมาจาก source เดียวกันแต่อ่านเป็น DataFrame แล้ว filter ตาม condition เพื่อหา duplicates แล้วเซฟเก็บไว้ในตัวแปรแค่ row เดียว
3. เลือก drop ทั้ง 2 rows จาก delta table (1)
4. แล้วทำการ update เข้าไปใหม่ด้วยการใช้ merge แล้ว set condition ตาม column ที่กำหนดให้ตรงกัน จากนั้นใช้ whenMatchedUpdatedAll() แบบนี้สามารถทำได้ไหมครับ

---

<div class="post-metadata">

**Author:** ![premsurawut](https://avatars.discourse-cdn.com/v4/letter/p/df788c/32.png) [@premsurawut](https://discuss.dataengineercafe.io/u/premsurawut)\
**Post date:** [July 9, 2022, 6:52am UTC](https://discuss.dataengineercafe.io/t/save-data-pyspark-drop-duplicate/328/6 "2022-07-09T06:52:11Z")

</div>

ขอบคุณมากครับ ตอนนี้ยังไม่มีความรู้เรื่อง performance เท่าไหร่ ตรงนี้ พอจะมี resource ให้ตามไปศึกษาเพิ่มเติมไหมครับ

---

<div class="post-metadata">

**Author:** ![kahnwong](https://yyz1.discourse-cdn.com/flex035/user_avatar/discuss.dataengineercafe.io/kahnwong/32/295_2.png) [@kahnwong](https://discuss.dataengineercafe.io/u/kahnwong)\
**Post date:** [July 9, 2022, 6:55am UTC](https://discuss.dataengineercafe.io/t/save-data-pyspark-drop-duplicate/328/7 "2022-07-09T06:55:12Z")

</div>

> **[Learning Spark](https://www.goodreads.com/book/show/24808098-learning-spark?ac=1&from_search=true&qid=r9P1agHTfA&rank=1)**
>
> Data in all domains is getting bigger. How can you work with it efficiently? Recently updated for Spark 1.3, this book introduces Apac...

เล่มนี้โอเคอยู่

delta table ไม่เคยใช้ แต่ตามหลักการแล้วไม่ควรมีการเขียนทับ source table  
ลองนึกสภาพดูว่า ถ้า source table มี data เข้ามาเรื่อยๆ  
สมมติ ตอนนี้ 2:00PM, มีการ ingest data มาทำกับ de-duplication  
แล้ว process นี้ ใช้เวลา 5 นาที  
ระหว่างนี้ มี 50 rows เพิ่มเข้ามาใน source table  
ถ้าเขียนทับไป จบเลยนะ ข้อมูลหาย

---

<div class="post-metadata">

**Author:** ![premsurawut](https://avatars.discourse-cdn.com/v4/letter/p/df788c/32.png) [@premsurawut](https://discuss.dataengineercafe.io/u/premsurawut)\
**Post date:** [July 9, 2022, 7:13am UTC](https://discuss.dataengineercafe.io/t/save-data-pyspark-drop-duplicate/328/8 "2022-07-09T07:13:04Z")

</div>

ขอบคุณมากเลยครับ พอจะเข้าใจละครับ

---

<div class="post-metadata">

**Author:** ![zkan](https://yyz1.discourse-cdn.com/flex035/user_avatar/discuss.dataengineercafe.io/zkan/32/2_2.png) [@zkan](https://discuss.dataengineercafe.io/u/zkan)\
**Post date:** [July 9, 2022, 8:49am UTC](https://discuss.dataengineercafe.io/t/save-data-pyspark-drop-duplicate/328/9 "2022-07-09T08:49:54Z")

</div>

> [@kahnwong](#):
>
> ตามหลักการแล้วไม่ควรมีการเขียนทับ source table  
> ลองนึกสภาพดูว่า ถ้า source table มี data เข้ามาเรื่อยๆ  
> สมมติ ตอนนี้ 2:00PM, มีการ ingest data มาทำกับ de-duplication  
> แล้ว process นี้ ใช้เวลา 5 นาที  
> ระหว่างนี้ มี 50 rows เพิ่มเข้ามาใน source table  
> ถ้าเขียนทับไป จบเลยนะ ข้อมูลหาย

ชอบตัวอย่างนี้มากครัช แจ่มเลย 👍