ไปป์ไลน์การตรวจจับการประพฤติมิชอบด้วย Data Agent Kit และ Antigravity IDE

1. บทนำ

สมมติว่าคุณเป็นนักวิทยาศาสตร์ข้อมูลที่ Cymbal Financial ซึ่งเป็นผู้ประมวลผลการชำระเงินที่มีปริมาณธุรกรรมสูง การชำระเงินล่าช้าเกิดขึ้นเป็นระลอก และทีมการปฏิบัติตามข้อกำหนดสงสัยว่ามีการฉ้อโกงที่ประสานงานกัน คุณต้องสร้างไปป์ไลน์เพื่อส่งบันทึกธุรกรรมของ Clearinghouse ดิบ ล้างข้อมูล ฝึกโมเดลแมชชีนเลิร์นนิง เรียกใช้การอนุมานแบบกลุ่ม และส่งธุรกรรมที่มีความเสี่ยงสูงไปยังคิวตรวจสอบของ Cloud Spanner เพื่อตรวจสอบด้วยตนเอง

โดยปกติแล้ว คุณจะต้องใช้เวลาหลายวันในการเขียนโค้ดการตั้งค่าที่ซ้ำกัน (สมุดบันทึก Spark, การกำหนดค่า dbt, สคริปต์การฝึก, DAG ของ Airflow) และสลับบริบทระหว่างอินเทอร์เฟซคอนโซลกับเอดิเตอร์อยู่ตลอดเวลา

ใน Codelab นี้ คุณจะได้เขียนโปรแกรมแบบคู่กับ Agent โดยใช้ Google Cloud Data Agent Kit (DAK) ภายใน Antigravity IDE Agent จะช่วยคุณสร้างสมุดบันทึก Spark, คอมไพล์โปรเจ็กต์ dbt, สร้างลูปการอนุมาน และจัดระเบียบเวิร์กโฟลว์โดยใช้Managed Service สำหรับ Apache Airflow โดยใช้ภาษาธรรมชาติที่ใช้ในการสนทนา

สิ่งที่คุณต้องทำ

  • นำเข้าบันทึกของ Clearinghouse จาก Cloud Storage โดยใช้ Managed Service สำหรับ Apache Spark (Spark แบบ Serverless) ลงในตาราง BigQuery
  • ขจัดข้อมูลที่ซ้ำกันและทำให้ธุรกรรมเป็นมาตรฐานโดยใช้ dbt เพื่อสร้างเลเยอร์ข้อมูลที่สะอาด (ดิบ การจัดเตรียม ข้อมูลเสริม)
  • ฝึกโมเดลการแยกประเภท Random Forest แบบกระจาย (RandomForestClassifier) ใน Spark Serverless
  • เรียกใช้การอนุมานแบบเป็นกลุ่มในธุรกรรมใหม่และเขียนการแจ้งเตือนที่มีความเสี่ยงสูงไปยัง Cloud Spanner โดยตรง
  • จัดระเบียบ กำหนดค่าด้วยภาพ และติดตั้งใช้งานไปป์ไลน์ทั้งหมดโดยใช้บริการที่มีการจัดการสำหรับ Apache Airflow และการตรวจสอบ DAG แบบอินเทอร์แอกทีฟภายใน IDE

สิ่งที่คุณต้องมี

  • เว็บเบราว์เซอร์ เช่น Chrome
  • โปรเจ็กต์ Google Cloud ที่เปิดใช้การเรียกเก็บเงิน (เราขอแนะนำให้ใช้โปรเจ็กต์ใหม่เฉพาะสำหรับแล็บภาคปฏิบัติ)
  • มีความคุ้นเคยพื้นฐานกับ SQL, Python และ PySpark
  • Antigravity IDE พร้อมการสมัครใช้บริการ Google AI Pro (แนะนำ)

ทรัพยากรที่สร้างในโค้ดแล็บนี้ควรมีค่าใช้จ่ายน้อยกว่า $5 อย่าลืมทำตามวิธีการล้างข้อมูลที่ท้ายแล็บเพื่อลบทรัพยากรที่จัดสรร

2. การตั้งค่าสภาพแวดล้อม

หากต้องการเริ่มแล็บ คุณจะต้องเรียกใช้สคริปต์การเริ่มต้น สคริปต์นี้จะเปิดใช้ GCP API ที่จำเป็นโดยอัตโนมัติ สร้างที่เก็บข้อมูล Cloud Storage สำหรับการนำเข้า สร้างชุดข้อมูลธุรกรรมและไดเรกทอรีจำลอง โหลดไดเรกทอรีอ้างอิงลงใน BigQuery และเริ่มการจัดสรร Cloud Spanner และ Managed Service สำหรับ Apache Airflow (เดิมชื่อ Cloud Composer) ในเบื้องหลัง

เลือกหรือสร้างโปรเจ็กต์

เลือกโปรเจ็กต์ที่มีอยู่หรือสร้างโปรเจ็กต์ใหม่ในคอนโซล Google Cloud

ยืนยันการเรียกเก็บเงิน

ตรวจสอบว่าได้เปิดใช้การเรียกเก็บเงินสำหรับโปรเจ็กต์ Google Cloud แล้ว ดูข้อมูลเพิ่มเติมเกี่ยวกับวิธีดำเนินการนี้ได้โดยทำตามคำแนะนำนี้

เรียกใช้สคริปต์การตั้งค่า

คุณจะใช้ Google Cloud Shell (หรือเชลล์ในเครื่องที่กำหนดค่าด้วย Google Cloud CLI) เพื่อเปิดใช้การตั้งค่าสภาพแวดล้อม

  1. เปิด คอนโซล Google Cloud
  2. คลิกเปิดใช้งาน Cloud Shell ในแถบเครื่องมือด้านขวาบน

เปิด Cloud Shell

  1. กำหนดค่าโปรเจ็กต์ที่ใช้งานอยู่ในเทอร์มินัล Cloud Shell โดยทำดังนี้
gcloud config set project <<YOUR_PROJECT_ID>>
export PROJECT_ID=$(gcloud config get-value project)
  1. โคลนที่เก็บ Codelab และไปที่โฟลเดอร์สคริปต์
cd ~/
git clone --filter=blob:none --no-checkout https://github.com/GoogleCloudPlatform/devrel-demos.git
cd ~/devrel-demos
git sparse-checkout init --cone
git sparse-checkout set codelabs/agentic-data-labs/data-science
git checkout main
cd codelabs/agentic-data-labs/data-science/scripts
  1. เรียกใช้สคริปต์การตั้งค่าการเริ่มต้นเพื่อทำให้ทรัพยากรทั้งหมดใช้งานได้กับ us-central1
chmod +x setup.sh setup_spanner.sh setup_composer.sh
export REGION=us-central1
./setup.sh
  1. เมื่อสคริปต์ทํางานเสร็จแล้ว คุณจะเห็นเอาต์พุตสรุปที่ระบุว่าชุดข้อมูล BigQuery และ Bucket ของ Cloud Storage พร้อมใช้งานแล้ว ในเบื้องหลัง Cloud Spanner (ใช้เวลาประมาณ 2 นาที) และ Managed Airflow (ใช้เวลาประมาณ 20 นาที) จะยังคงจัดสรรต่อไป คุณตรวจสอบความคืบหน้าได้ทุกเมื่อโดยเรียกใช้คำสั่งต่อไปนี้
tail -f /tmp/spanner_setup.log
tail -f /tmp/composer_setup.log

เปิด Antigravity IDE

  1. ดาวน์โหลดและติดตั้ง Antigravity IDE จากหน้าดาวน์โหลด Google Antigravity
  2. เปิดใช้ Antigravity IDE
  3. สร้างโฟลเดอร์ใหม่ที่ว่างเปล่าในเครื่อง (เช่น ชื่อ agentic-data-labs) แล้วเปิดใน IDE โดยเลือกเปิดโฟลเดอร์ ซึ่งจะทำหน้าที่เป็นพื้นที่ทำงานในเครื่องสำหรับ Codelab

กำหนดค่าโฟลเดอร์โปรเจ็กต์ Antigravity IDE

ติดตั้งส่วนขยาย Data Agent Kit

ส่วนขยายชุดเครื่องมือตัวแทนข้อมูล Google Cloud ช่วยให้ผสานรวมกับบริการข้อมูลของ Google Cloud ได้อย่างลึกซึ้งภายในเอดิเตอร์โดยตรง ซึ่งช่วยให้คุณโต้ตอบกับ BigQuery, Cloud SQL, Cloud Storage และอื่นๆ ได้โดยไม่ต้องเปลี่ยนบริบท

  1. ใน Antigravity IDE ให้คลิกไอคอนส่วนขยายในแถบกิจกรรมทางด้านซ้ายสุดของหน้าจอ (มีลักษณะเป็นสี่เหลี่ยม 4 รูป)
  2. พิมพ์ Google Cloud Data Agent Kit ในแถบค้นหาที่ด้านบนของแผงส่วนขยาย
  3. ค้นหาส่วนขยายชื่อ Google Cloud Data Agent Kit ที่เผยแพร่โดย googlecloudtools
  4. คลิกปุ่มติดตั้ง
  5. ข้อความแจ้งอาจปรากฏขึ้นเพื่อถามว่า "คุณเชื่อถือผู้เผยแพร่ ‘googlecloudtools' และส่วนขยายของผู้เผยแพร่รายนี้ไหม" คลิกเชื่อถือผู้เผยแพร่โฆษณาและติดตั้งเพื่อดำเนินการต่อ

ติดตั้งส่วนขยาย Data Agent Kit

เมื่อติดตั้งแล้ว คุณจะเห็นไอคอน Google Cloud Data Agent Kit ใหม่ปรากฏในแถบกิจกรรมทางด้านซ้ายสุดของ Antigravity IDE

  1. ระบบควรเปิดหน้าเริ่มต้นใช้งานชื่อ "ยินดีต้อนรับสู่ชุดเครื่องมือตัวแทนข้อมูลของ Google Cloud" โดยอัตโนมัติ หากไม่ได้ลงชื่อเข้าใช้บัญชีระบบคลาวด์ ให้ทำตามข้อความแจ้งเพื่ออนุญาตการเข้าถึง
  2. ในส่วนสรุปการกำหนดค่า ให้ค้นหาช่องโปรเจ็กต์ คลิกเมนูแบบเลื่อนลงแล้วเลือกโปรเจ็กต์ที่อยู่ในระบบคลาวด์ของ Google ตั้งค่าภูมิภาคเป็น us-central1 จากนั้นเลือกกำหนดค่าเซิร์ฟเวอร์ MCP

การกำหนดค่าเริ่มต้นของส่วนขยาย Data Agent Kit

  1. เลือก Configure MCP Servers ในบานหน้าต่างการกำหนดค่า MCP ให้ตรวจสอบว่าคุณได้เปิดใช้เซิร์ฟเวอร์ MCP ระยะไกลต่อไปนี้
    • BigQuery
    • Spanner
    • สมุดบันทึก

จากนั้นคลิกเริ่มต้นใช้งาน

กำหนดค่าเซิร์ฟเวอร์ MCP

ดูตัวเลือกการกำหนดค่า

เมื่อตั้งค่าเสร็จแล้ว คุณจะไปที่หน้า "เริ่มต้นใช้งานชุดเครื่องมือ Data Agent ของ Google Cloud"

  1. คลิกเริ่มต้นใช้งานในส่วน "การตั้งค่าและการกำหนดค่า"
  2. ซึ่งจะเปิดแผงการกำหนดค่าชุดเครื่องมือตัวแทนข้อมูล สำรวจแท็บต่างๆ ดังนี้
    • โปรเจ็กต์และภูมิภาค: ยืนยันรหัสโปรเจ็กต์ที่เลือกและตรวจสอบว่าสคริปต์การตั้งค่าได้เปิดใช้ API ที่จำเป็นทั้งหมด (Compute Engine, Cloud Storage, BigQuery, Spanner ฯลฯ)
    • BigQuery: กำหนดค่าตำแหน่งเริ่มต้นสำหรับการค้นหา BigQuery ใช้ภูมิภาค us-central1
    • กำหนดค่าเซิร์ฟเวอร์ MCP: ดูเซิร์ฟเวอร์ MCP ที่เปิดใช้ (BigQuery, Notebook, Spanner ฯลฯ) ซึ่งอนุญาตให้เอเจนต์ AI โต้ตอบกับข้อมูลของคุณได้อย่างปลอดภัย
    • ทักษะ: สำรวจทักษะที่สร้างไว้ล่วงหน้าซึ่งมอบความสามารถเฉพาะทางให้กับเอเจนต์สำหรับงานข้อมูลที่ซับซ้อน

แผงการตั้งค่าชุดเครื่องมือ Data Agent

สรุปส่วน: คุณเรียกใช้สคริปต์การเริ่มต้นเพื่อสร้างชิ้นงาน GCS และ BigQuery ขณะที่ Spanner และ Airflow สร้างในเบื้องหลัง จากนั้นคุณได้เปิดโปรเจ็กต์ใน Antigravity IDE และเปิดใช้งานส่วนขยาย Google Cloud Data Agent Kit ตอนนี้คุณพร้อมที่จะเขียน Notebook แรกแล้ว

3. นำเข้าบันทึกดิบโดยใช้ Spark Serverless

ในส่วนนี้ คุณจะนำเข้าบันทึกธุรกรรม JSON ดิบไปยังที่เก็บข้อมูลขนาดใหญ่ Managed Service สำหรับ Apache Spark (Spark Serverless) จะเชื่อมต่อกับที่เก็บข้อมูลดั้งเดิมของ BigQuery โดยตรง คุณจะใช้เครื่องมือเชื่อมต่อ BigQuery มาตรฐานเพื่อจัดการข้อมูลตาราง และเปิดใช้การค้นหาและการวิเคราะห์โดยตรง

สำรวจรันไทม์ Spark Serverless ที่กำหนดค่าไว้ล่วงหน้า

ก่อนที่จะเรียกใช้โค้ด Spark ให้ตรวจสอบเทมเพลตรันไทม์แบบ Serverless ที่สคริปต์การตั้งค่ากำหนดค่าไว้ล่วงหน้า เทมเพลตนี้กำหนดแบ็กเอนด์สภาพแวดล้อมการดำเนินการเป้าหมายและรวมทรัพยากร Dependency ของตัวเชื่อมต่อที่จำเป็น

  1. เปิดแผง Google Cloud Data Agent Kit ในแถบกิจกรรม IDE
  2. ขยายเมนูแบบเลื่อนลง Apache Spark แล้วขยาย Serverless
  3. คลิกขวาที่ fraud-pipeline-runtime แล้วเลือกโปรไฟล์เพื่อเปิดมุมมองการกำหนดค่าในเอดิเตอร์
  4. ในแท็บโปรไฟล์ ให้เลื่อนลงและขยายพร็อพเพอร์ตี้เพื่อตรวจสอบการอ้างอิงที่กำหนดเองซึ่งแนบมากับสภาพแวดล้อม
    • spark.jars: มี gs://spark-lib/spanner/spark-3.5-spanner-1.4.0.jar ซึ่งใช้ตัวเชื่อมต่อ Spark Spanner เพื่อให้งาน Spark เขียนผลลัพธ์การอนุมานไปยัง Cloud Spanner ได้โดยตรงในภายหลังในแล็บ (หมายเหตุ: Dataproc Serverless มีเครื่องมือเชื่อมต่อ Spark BigQuery ของ Google Cloud โดยค่าเริ่มต้น จึงไม่จำเป็นต้องกำหนดค่า JAR เพิ่มเติมเพื่ออ่านและเขียนตาราง BigQuery)

สำรวจพร็อพเพอร์ตี้ของรันไทม์แบบไร้เซิร์ฟเวอร์ของ Spark

  1. สังเกตแท็บเซสชันแบบอินเทอร์แอกทีฟทางด้านซ้าย ขณะนี้ไม่มีอะไรเนื่องจากคุณยังไม่ได้เรียกใช้โค้ด ทันทีที่คุณเรียกใช้ Notebook ในขั้นตอนถัดไป เซสชันการประมวลผลแบบไร้เซิร์ฟเวอร์ที่ใช้งานจริงจะจัดสรรแบบไดนามิกและปรากฏที่นี่

นำเข้าข้อมูลโดยใช้ชุดเครื่องมือ Data Agent

คุณจะจับคู่โปรแกรมกับตัวแทนโดยใช้ Data Agent Kit แทนการกำหนดค่า Spark Session ด้วยตนเองหรือเขียนสคริปต์การโหลด PySpark ตั้งแต่ต้น

  1. เปิดแผงแชทกับตัวแทนโดยคลิกไอคอนสลับตัวแทนในแถบเครื่องมือด้านขวาบน
  2. วางพรอมต์ต่อไปนี้ลงในแชท (อย่าลืมแทนที่ ${PROJECT_ID} ด้วยรหัสโปรเจ็กต์ Google Cloud จริงของคุณ)
Create a PySpark notebook (01_ingestion.ipynb) to ingest JSON transaction logs
from gs://${PROJECT_ID}-fin-clearing-raw/ into a BigQuery table
`${PROJECT_ID}.transactions_dataset_evals.raw_transactions`
using the Spark BigQuery connector (`format("bigquery")`) with overwrite mode.
  1. หากตัวแทนขอสิทธิ์ในการเรียกใช้คำสั่งการยืนยันในเบื้องหลัง (เช่น "อนุญาตให้เรียกใช้คำสั่งนี้ไหม") ให้ตรวจสอบคำสั่งที่เสนอและเลือกใช่ อนุญาตครั้งนี้ (หรือใช่ และอนุญาตเสมอ)
  2. เมื่อตัวแทนสร้างไฟล์เสร็จแล้ว ให้คลิกปุ่มยอมรับทั้งหมดสีน้ำเงิน (หรือไอคอนเครื่องหมายถูก) ที่ด้านล่างของแผงแชทเพื่อบันทึก notebooks/01_ingestion.ipynb ลงในพื้นที่ทำงาน

ตัวแทนที่สร้าง Notebook การส่งผ่านข้อมูล

ตรวจสอบและเรียกใช้ Notebook

  1. เปิด notebooks/01_ingestion.ipynb ที่สร้างขึ้นใหม่ใน IDE
  2. ตรวจสอบโค้ด PySpark สำหรับตรรกะการเขียนของเครื่องมือเชื่อมต่อ BigQuery
  3. คลิกเรียกใช้ทั้งหมดในแถบเครื่องมือของสมุดบันทึก IDE
  4. หากเรียกใช้สมุดบันทึก Spark ระยะไกลเป็นครั้งแรก IDE อาจแจ้งให้คุณติดตั้งการอ้างอิงในเครื่อง หากได้รับข้อความแจ้ง ให้คลิกติดตั้งการอ้างอิงสำหรับเคอร์เนล Spark ระยะไกล และยืนยันกล่องโต้ตอบการติดตั้ง จากนั้นคลิกเรียกใช้ทั้งหมดอีกครั้ง
  5. ในเมนูแบบเลื่อนลงเลือกเคอร์เนล ให้เลือกเคอร์เนล Spark ระยะไกล -> fraud-pipeline-runtime ใน Serverless Spark (เคล็ดลับ: หากไม่เห็นเทมเพลตเวลาเรียกใช้ที่กำหนดค่าไว้ล่วงหน้า ให้คลิกไอคอนรีเฟรชที่ด้านขวาบนของเมนูแบบเลื่อนลงของตัวเลือกเคอร์เนลเพื่อโหลดเคอร์เนลระยะไกลที่พร้อมใช้งานอีกครั้ง)
  6. ดูแถบสถานะที่ด้านซ้ายล่างของโปรแกรมแก้ไข คุณจะเห็น Connecting to kernel: fraud-pipeline-runtime on Serverless Spark... เนื่องจากเป็นการเปิดตัวครั้งแรกของแบ็กเอนด์เคอร์เนลรันไทม์แบบไร้เซิร์ฟเวอร์ของ Spark ระบบจึงใช้เวลาสักครู่ในการจัดสรรและบูต
  7. เมื่อเคอร์เนลเชื่อมต่อเสร็จแล้ว Notebook จะเริ่มดำเนินการกับเซลล์ทั้งหมดตามลำดับโดยอัตโนมัติเพื่อประมวลผลบันทึกธุรกรรมดิบเป็นชุดข้อมูล BigQuery

การยืนยัน

เมื่อการดำเนินการเสร็จสมบูรณ์แล้ว ให้ตรวจสอบแคตตาล็อก Data Agent Kit เพื่อยืนยันการสร้างตาราง

ยืนยันตารางดิบใน Catalog Explorer

  1. เปิดแผง Google Cloud Data Agent Kit ในแถบกิจกรรม IDE
  2. ขยายส่วนแคตตาล็อก
  3. ขยายรหัสโปรเจ็กต์
  4. ขยาย BigQuery
  5. ขยายชุดข้อมูล transactions_dataset_evals
  6. คลิกตาราง raw_transactions เพื่อเปิดมุมมองรายละเอียดในเครื่องมือแก้ไขหลัก
  7. ในการนำทางด้านซ้าย ให้สำรวจแท็บข้อมูล สคีมา และรายละเอียดเพื่อตรวจสอบระเบียนและข้อมูลเมตาที่ส่งผ่านข้อมูล

สรุปส่วน: คุณใช้ภาษาธรรมชาติใน Agent Chat เพื่อสร้างภาระงาน Spark Serverless ที่สมบูรณ์ จากนั้นคุณก็เรียกใช้เพื่อประมวลผลบันทึก JSON ที่ไม่มีโครงสร้างเป็นตาราง BigQuery (ดิบ)

4. กรองข้อมูลที่ซ้ำกันออกและจัดรูปแบบให้เป็นมาตรฐานด้วย dbt

ก่อนฝึกโมเดล ML คุณจะบังคับใช้คุณภาพของข้อมูลโดยการนำบันทึกการสตรีมที่ซ้ำกันออก แยกเรคคอร์ดที่ไม่ดี (เช่น รหัสธุรกรรมที่ว่างเปล่า) และรวมข้อมูลมิติ (ผู้ชำระเงินและผู้รับเงิน) กระบวนการนี้ต้องมีการเปลี่ยนรูปแบบ SQL ที่เชื่อถือได้และไม่เปลี่ยนค่าเดิม ซึ่งทำให้ dbt (เครื่องมือสร้างข้อมูล) เหมาะสมอย่างยิ่ง

สร้างโครงสร้างไปป์ไลน์ dbt

ใช้เอเจนต์เพื่อสร้างโปรเจ็กต์ dbt ในชุดข้อมูล BigQuery

  1. กลับไปที่แผงแชทกับตัวแทน
  2. ระบุวิธีการต่อไปนี้เพื่อสร้างโปรเจ็กต์ dbt
Scaffold a us-central1 dbt project in dbt_project/ that maps raw_transactions
through to an enriched_transactions model in dataset transactions_dataset_evals.

Deduplicate by transaction_id in staging. Quarantine null IDs to an invalid_transactions model.
Join the valid staging records with dim_payers and dim_payees for the enriched_transactions model,
preserving the historical `is_fraud` label column, and finally add a transaction uniqueness test.

Create an implementation plan first.
  1. Agent จะแสดงอาร์ติแฟกต์แผนการติดตั้งใช้งานในบานหน้าต่างเอดิเตอร์หลัก ตรวจสอบโครงสร้างไฟล์และตรรกะ SQL ที่เสนอ
  2. คลิกดำเนินการต่อ (และยอมรับทั้งหมด) เพื่ออนุญาตให้เอเจนต์สร้างไฟล์ในพื้นที่ทำงาน

แผนการติดตั้งใช้งานพร้อมปุ่มดำเนินการต่อ

  1. เมื่อสร้างเสร็จแล้ว เอเจนต์จะแสดงคำแนะนำแบบทีละขั้นตอนที่สรุปคอมโพเนนต์ใหม่ ยอมรับการเปลี่ยนแปลงทั้งหมดหากได้รับข้อความแจ้ง

ยอมรับไฟล์ที่สร้างขึ้นทั้งหมดในแผงแชท

สร้างและทดสอบ

แม้ว่าเอเจนต์จะเรียกใช้ dbt compile โดยอัตโนมัติเพื่อให้แน่ใจว่า SQL ที่สร้างขึ้นนั้นถูกต้องตามไวยากรณ์ แต่ตอนนี้คุณจะต้องสร้างมุมมองและตารางเหล่านี้ใน BigQuery และเรียกใช้การทดสอบคุณภาพของข้อมูลเพื่อการยืนยันในเครื่อง (หมายเหตุ: ในภายหลังใน Lab คุณจะทำให้ขั้นตอน dbt นี้เป็นอัตโนมัติซึ่งเป็นส่วนหนึ่งของ DAG ของ Airflow แบบครบวงจร)

  1. ในแถบกิจกรรมทางด้านซ้ายสุด ให้คลิกไอคอนสำรวจ (หรือกด Cmd/Ctrl+Shift+E)
  2. ขยาย dbt_project -> models เพื่อตรวจสอบโมเดล SQL ที่สร้างขึ้น คลิก enriched_transactions.sql เพื่อเปิดและตรวจสอบตรรกะของฟีเจอร์การแปลงและการฉ้อโกงในเครื่องมือแก้ไข
  3. ใน File Explorer ให้คลิกขวาที่โฟลเดอร์ dbt_project แล้วเลือกเปิดในเทอร์มินัลแบบผสานรวม ซึ่งจะเปิดแผงเทอร์มินัลที่ตั้งค่าโดยตรงไปยังdbt_projectไดเรกทอรีการทำงานที่จำเป็นโดยอัตโนมัติ
  4. หากยังไม่ได้ติดตั้ง dbt ให้สร้างสภาพแวดล้อมเสมือนภายนอก dbt_project/ (ที่รูทของบ้านหรือพื้นที่ทํางาน) แล้วติดตั้งอะแดปเตอร์ BigQuery ดังนี้
python3 -m venv ~/.venv/dbt
source ~/.venv/dbt/bin/activate
pip install dbt-bigquery
  1. เรียกใช้โมเดล dbt และการทดสอบคุณภาพของข้อมูลที่เกี่ยวข้อง
dbt build
  1. ดูเอาต์พุตของเทอร์มินัล dbt จะคอมไพล์ SQL สร้างตารางการจัดเตรียมและตารางที่เพิ่มคุณค่าใน BigQuery และเรียกใช้การทดสอบข้อมูล

สร้างและทดสอบโปรเจ็กต์ dbt ในเทอร์มินัลแบบผสานรวม

  1. เมื่อการสร้างเสร็จสิ้น ให้ปิดแผงเทอร์มินัลเพื่อเพิ่มพื้นที่หน้าจอสำหรับขั้นตอนที่เหลือ

สรุปส่วน: คุณสร้างโปรเจ็กต์ dbt ด้วยเอเจนต์ เรียกใช้การทดสอบคุณภาพของข้อมูล และแปลงระเบียนดิบเป็นตาราง BigQuery ที่จัดเตรียมและเสริม

5. ฝึกโมเดลการตรวจจับการฉ้อโกงแบบกระจายด้วย Random Forest

เมื่อธุรกรรมที่เพิ่มคุณค่าแล้วปรากฏใน BigQuery คุณจะสร้างโมเดลแมชชีนเลิร์นนิงเพื่อจัดประเภทเหตุการณ์ที่เป็นการประพฤติมิชอบ Random Forest เป็นวิธีการเรียนรู้แบบรวมที่เหมาะกับข้อมูลการจัดประเภทแบบตาราง การเรียกใช้ RandomForestClassifier ใน Spark Serverless จะกระจายการฝึกโมเดลไปยังโหนด Worker โดยไม่ต้องให้คุณจัดการโครงสร้างพื้นฐาน

ในขั้นตอนนี้ คุณจะใช้ Agent เพื่อสร้างไปป์ไลน์การฝึกโมเดล Spark ML

สร้าง Notebook การฝึก ML

  1. เปิดแผงแชทกับตัวแทน
  2. ระบุพรอมต์ต่อไปนี้เพื่อออกแบบลําดับการฝึกโมเดล (อย่าลืมแทนที่ ${PROJECT_ID} ด้วยรหัสโปรเจ็กต์ที่ใช้งานอยู่)
Create a PySpark notebook (02_training.ipynb) to train a distributed Random Forest
(RandomForestClassifier) model on the BigQuery table
`transactions_dataset_evals`.`enriched_transactions`, predicting the `is_fraud` label.
Train only on historically labeled records where `is_fraud` is not null.

One-hot encode categorical strings, scale amounts, cache the dataset in memory,
evaluate AUC, and save the evaluated model to gs://${PROJECT_ID}-models/fraud_model.
  1. ตรวจสอบแพ็กเกจของเอเจนต์หรือโค้ดที่สร้างขึ้น แล้วคลิกดำเนินการต่อ / ยอมรับทั้งหมดเพื่อบันทึก notebooks/02_training.ipynb ลงในพื้นที่ทำงาน

ตัวแทนสร้าง Notebook การฝึก

ตรวจสอบและเรียกใช้ Notebook

  1. เปิด notebooks/02_training.ipynb ในเครื่องมือแก้ไข
  2. ตรวจสอบขั้นตอนไปป์ไลน์ ML ของ PySpark สำหรับการเข้ารหัสฟีเจอร์ การประกอบเวกเตอร์ และตรรกะการจัดประเภท Random Forest
  3. คลิกเรียกใช้ทั้งหมดในแถบเครื่องมือของสมุดบันทึก IDE
  4. เมื่อตัวเลือกแบบเลื่อนลงเลือกเคอร์เนลเปิดขึ้น ให้เลือก fraud-pipeline-runtime ใน Serverless Spark

การเลือกเคอร์เนล Spark แบบ Serverless สำหรับ Notebook การฝึก

การยืนยัน

เมื่อการดำเนินการเสร็จสมบูรณ์แล้ว ให้ยืนยันว่าโมเดลได้รับการฝึกและส่งออกอย่างถูกต้อง

  1. ตรวจสอบเอาต์พุตของเซลล์การประเมินที่ด้านล่างของ Notebook เพื่อยืนยันคะแนนพื้นที่ใต้ ROC (AUC) ที่รายงาน
  2. ขยายแผงสำรวจพื้นที่เก็บข้อมูลในแถบด้านข้างของ Data Agent Kit เพื่อให้แน่ใจว่าระบบบันทึกอาร์ติแฟกต์ของโมเดลไปยัง GCS เรียบร้อยแล้ว
  3. ค้นหาที่เก็บข้อมูลที่ลงท้ายด้วย -models (เชื่อมโยงกับรหัสโปรเจ็กต์ที่ใช้งานอยู่) ขยายที่เก็บข้อมูลดังกล่าว และเจาะลึกลงไปเพื่อยืนยันว่ามีไดเรกทอรี fraud_model และขั้นตอนของไปป์ไลน์

ยืนยันว่าโมเดลบันทึกใน GCS แล้ว

สรุปส่วน: คุณใช้ Agent เพื่อสร้างไปป์ไลน์การฝึก ML ของ PySpark ฝึกโมเดล Random Forest ในตาราง BigQuery ที่เพิ่มคุณค่าแล้ว และส่งออกโมเดลไปยัง Cloud Storage

6. การอนุมานแบบเป็นชุดและการเขียน Cloud Spanner

เมื่อมีโมเดลการคาดการณ์ที่ฝึกแล้วซึ่งจัดเก็บไว้ใน Cloud Storage คุณจะเรียกใช้การอนุมานแบบกลุ่มกับธุรกรรมใหม่ที่ไหลผ่าน BigQuery ได้ ระบบต้องกำหนดเส้นทางการทำธุรกรรมที่มีความเสี่ยงสูงไปยังระบบปฏิบัติการเพื่อให้ทีมการปฏิบัติตามข้อกำหนดตรวจสอบธุรกรรมเหล่านั้นได้ Cloud Spanner มีฐานข้อมูลธุรกรรมที่ปรับขนาดได้สำหรับคิวการตรวจสอบนี้

สร้าง Notebook การอนุมานแบบกลุ่ม

ใช้ Agent เพื่อสร้าง Notebook การอนุมานที่เชื่อมต่อ BigQuery, Cloud Storage และ Cloud Spanner โดยทำดังนี้

  1. เปิดแผงแชทกับตัวแทน
  2. ระบุพรอมต์ต่อไปนี้
Create an inference notebook (03_inference.ipynb) that loads the RandomForestClassifier
model to score unlabeled records (where `is_fraud` is null) from the BigQuery table
`transactions_dataset_evals`.`enriched_transactions`.

Filter for high-risk transactions with a 50%+ fraud probability score (probability >= 0.50)
and write them to the Cloud Spanner table SparkEvalFraudReviewQueue in the cymbal-fraud instance
under fraud-db.
  1. ยอมรับ Notebook ที่สร้างขึ้นเพื่อบันทึก notebooks/03_inference.ipynb ลงในพื้นที่ทำงาน

ตัวแทนที่สร้าง Notebook การอนุมาน

ตรวจสอบและเรียกใช้ Notebook

  1. เปิด notebooks/03_inference.ipynb ที่สร้างขึ้นใหม่ในเครื่องมือแก้ไข
  2. ตรวจสอบลำดับการอนุมานของ PySpark ดังนี้
    • ทรัพยากร Dependency: เทมเพลตรันไทม์แบบ Serverless มี cloud-spanner ทรัพยากร Dependency ของ JAR ที่จำเป็นสำหรับการดำเนินการ Spark
    • การจัดรูปแบบข้อมูล: สคริปต์จะทิ้งคอลัมน์เวกเตอร์ Spark ML ที่ซับซ้อน (เช่น ฟีเจอร์ดิบและความน่าจะเป็น) ก่อนที่จะเขียนเพื่อให้ตรงกับสคีมาตาราง Spanner
    • Spanner Connector: เขียนแถวที่แจ้งโดยใช้ .format("cloud-spanner") เพื่อต่อท้ายคิวรีวิวโดยตรง
  3. คลิกเรียกใช้ทั้งหมดในแถบเครื่องมือของสมุดบันทึก IDE
  4. เมื่อระบบแจ้งให้เลือกเคอร์เนล ให้เลือก fraud-pipeline-runtime on Serverless Spark

การยืนยัน

เมื่อ Notebook การอนุมานประมวลผลเสร็จแล้ว คุณจะค้นหาฐานข้อมูล Spanner ที่ใช้งานได้โดยตรงภายใน IDE ได้

  1. เปิดแผง Google Cloud Data Agent Kit ในแถบกิจกรรม IDE
  2. ขยายส่วนแคตตาล็อก
  3. ขยายรหัสโปรเจ็กต์ แล้วขยาย Spanner
  4. ไปที่ cymbal-fraud -> fraud-db -> Tables -> SparkEvalFraudReviewQueue
  5. คลิกขวาที่ตาราง แล้วเลือกตารางการสืบค้น จากนั้นเรียกใช้การสืบค้นข้อมูล
SELECT *
FROM `SparkEvalFraudReviewQueue`
LIMIT 100;
  1. ในแผงผลการค้นหาด้านล่าง คุณควรเห็นแถวที่เพิ่งแทรกซึ่งแสดงถึงธุรกรรมที่มีความเสี่ยงสูงซึ่งมีการแจ้งว่าไม่เหมาะสมเพื่อให้ตรวจสอบด้วยตนเอง

ยืนยันแถวใน Cloud Spanner

สรุปส่วน: คุณใช้ Agent เพื่อสร้าง Notebook การอนุมานแบบกลุ่ม ให้คะแนนระเบียน BigQuery ที่ไม่มีป้ายกำกับด้วยโมเดลที่ฝึกแล้ว และเขียนธุรกรรมที่มีความเสี่ยงสูงไปยัง Cloud Spanner โดยตรง

7. สร้างโครงร่างและจัดระเบียบด้วย Airflow ที่มีการจัดการ

ปัจจุบันไปป์ไลน์ของคุณประกอบด้วยขั้นตอนที่แยกกัน ได้แก่ Notebook การนำเข้า โปรเจ็กต์การเปลี่ยนรูปแบบ dbt และ Notebook การอนุมานแบบกลุ่ม หากต้องการทำให้พร้อมใช้งานจริง คุณจะต้องเชื่อมโยงเข้าด้วยกันเป็นกราฟทรัพยากร Dependency ที่กำหนดเวลาไว้

Managed Service for Apache Airflow (เดิมชื่อ Cloud Composer) มีเครื่องมือจัดระเบียบที่มีการจัดการสำหรับเวิร์กโฟลว์นี้ ชุดเครื่องมือ Data Agent มีฟีเจอร์ไปป์ไลน์การจัดระเบียบที่แปลงคำจำกัดความไปป์ไลน์ YAML แบบประกาศเป็น DAG ของ Airflow โดยตรง

กำหนดไปป์ไลน์

ใช้เอเจนต์เพื่อสร้างการกำหนดค่าไปป์ไลน์การจัดระเบียบ

  1. ใน Agent Chat ให้ป้อนพรอมต์ต่อไปนี้ (อย่าลืมแทนที่ ${PROJECT_ID})
Initialize and define an orchestration pipeline (fraud_analysis_pipeline) triggering
the ingestion notebook, dbt project, and inference notebook in sequential order.
For Dataproc Serverless engine configs in us-central1, use resourceProfile.inline
(defining properties with spark.jars: "gs://spark-lib/spanner/spark-3.5-spanner-1.4.0.jar"
for inference) rather than resourceProfile.path or overrides.

Set the schedule interval to run daily at midnight, and use
gs://${PROJECT_ID}-airflow-artifacts for artifact storage.

ตรวจสอบการกำหนดค่า DAG

Orchestrator ของ Data Agent Kit ใช้การกำหนดค่า YAML แบบประกาศเพื่อกำหนดและติดตั้งใช้งานไปป์ไลน์กับ Apache Airflow ซึ่งช่วยให้สามารถควบคุมเวอร์ชันและติดตั้งใช้งานคำจำกัดความผ่าน CI/CD ได้

ในแผง IDE Explorer ให้ตรวจสอบไฟล์ไปป์ไลน์ 2 ไฟล์ที่เอเจนต์สร้างขึ้นที่รูทของพื้นที่ทำงาน

  1. deployment.yaml: เปิดไฟล์นี้ ซึ่งจะทำหน้าที่เป็นรีจิสทรีสภาพแวดล้อมของคุณ โดยจะเชื่อมโยงไปป์ไลน์เชิงตรรกะ dev กับสภาพแวดล้อม cymbal-airflow ตั้งค่าภูมิภาคการดำเนินการ (us-central1) และกำหนดที่เก็บข้อมูล artifact_storage ที่จัดเก็บ DAG ที่คอมไพล์แล้วและการอ้างอิง
  2. fraud_analysis_pipeline.yaml: เปิดไฟล์นี้ ซึ่งจะเป็นตัวกำหนดกราฟการดำเนินการ โดยจะระบุตารางเวลาทริกเกอร์ (interval: '0 0 * * *') และจัดลําดับ 3 ขั้นตอนภายใต้บล็อก actions ดังนี้
    • notebook การดำเนินการนำเข้าสำหรับ 01_ingestion.ipynb ที่ทำงานใน Dataproc Serverless
    • pipeline การดำเนินการเปลี่ยนรูปแบบที่กำหนดเป้าหมายไปยังไดเรกทอรี dbt_project โดยมีdependsOnทรัพยากร Dependency ที่ชี้ไปยังขั้นตอนการส่งผ่านข้อมูล
    • notebook การดำเนินการอนุมานสำหรับ 03_inference.ipynb ที่มีdependsOnทรัพยากร Dependency ที่ชี้ไปยังขั้นตอน dbt ซึ่งรวมพร็อพเพอร์ตี้ JAR ของ Spanner
  3. นอกจากนี้ เอเจนต์ยังจะสรุปอาร์ติแฟกต์ที่สร้างขึ้นเหล่านี้ไว้ในแท็บคำแนะนำแบบทีละขั้นตอนในแผงเอดิเตอร์ โดยจะระบุการกำหนดค่าและการตรวจสอบที่ดำเนินการ

การกำหนดค่า DAG แบบโต้ตอบ

Data Agent Kit จะแสดงการกำหนดค่าไปป์ไลน์เป็นกราฟแบบอินเทอร์แอกทีฟเพื่อตรวจสอบและแก้ไขพร็อพเพอร์ตี้ DAG ของ Airflow

  1. เปิดแผง Google Cloud Data Agent Kit ในแถบกิจกรรม IDE
  2. ในส่วนDATA ENGINEERING ให้ขยายOrchestration Pipelines
  3. คลิก fraud_analysis_pipeline.yaml เพื่อเปิด Canvas ของ DAG แบบภาพในเครื่องมือแก้ไขหลัก

Canvas ภาพของ DAG การจัดการเป็นกลุ่ม

  1. คลิกโหนด Schedule trigger ที่ด้านบน ฟลายเอาต์การกำหนดค่าจะเปิดขึ้นทางด้านขวา โดยจะแสดงสตริง Cron ที่แยกวิเคราะห์แล้ว (0 0 * * *) และให้คุณปรับพารามิเตอร์ต่างๆ เช่น การป้อนข้อมูลย้อนหลังและการติดตาม
  2. คลิกโหนดงาน Notebook (เช่น ขั้นตอนการนำเข้าหรือการอนุมาน) ฟลายเอาต์จะอัปเดตเพื่อแสดงการแมปการดำเนินการ Dataproc Serverless และพร็อพเพอร์ตี้ของตัวเชื่อมต่อที่เฉพาะเจาะจง
  3. สังเกตไฮเปอร์ลิงก์ชื่อไฟล์ Notebook (เช่น 01_ingestion.ipynb) ภายในบล็อกโหนด การคลิกจะเปิด Notebook ในเอดิเตอร์โดยตรง
  4. คลิก Deployment configuration ในแถบด้านข้างทางซ้ายใต้ไปป์ไลน์การจัดระเบียบ มุมมองนี้แสดงคลัสเตอร์สภาพแวดล้อม dev เป้าหมายและอาร์ติแฟกต์ของที่เก็บข้อมูล GCS เอาต์พุต

สรุปส่วน: คุณสร้างการกำหนดค่าไปป์ไลน์การจัดการเป็นกลุ่มด้วย Agent โดยกำหนดทรัพยากร Dependency ระหว่างงานการนำเข้า dbt และการอนุมานใน Canvas แบบภาพที่โต้ตอบได้

8. ติดตั้งใช้งาน ดำเนินการ และตรวจสอบ

เมื่อกำหนด DAG ในเครื่องแล้ว คุณจะเชื่อมต่อกับสภาพแวดล้อม Managed Airflow ที่จัดสรรไว้ในระหว่างการตั้งค่าและติดตั้งใช้งานไปป์ไลน์

กำหนดค่าบริการที่มีการจัดการสำหรับ Apache Airflow

ก่อนที่จะติดตั้งใช้งาน ให้กำหนดค่าการเชื่อมต่อตัวกำหนดตารางเวลาในการตั้งค่า Data Agent Kit เพื่อให้ส่วนขยายกำหนดเป้าหมายไปยังสภาพแวดล้อม Airflow ที่มีการจัดการ

  1. เปิดแผง Google Cloud Data Agent Kit ในแถบกิจกรรม IDE
  2. คลิกการตั้งค่าในส่วนSETTINGS
  3. เลือกตัวกำหนดเวลาจากเมนูด้านซ้าย
  4. กำหนดการตั้งค่า
    • รหัสโปรเจ็กต์: เลือกรหัสโปรเจ็กต์ที่ใช้งานอยู่
    • ภูมิภาค: เลือก us-central1
    • สภาพแวดล้อม: เลือก cymbal-airflow
  5. คลิกบันทึก

การตั้งค่า Managed Service สำหรับ Apache Airflow

ทําให้ DAG ใช้งานได้

ตอนนี้คุณจะสามารถติดตั้งใช้งานไปป์ไลน์ที่กำหนดค่าไว้ในสภาพแวดล้อม Airflow ที่มีการจัดการได้โดยตรงจาก Canvas แบบภาพ

  1. ในแถบด้านข้างของ Google Cloud Data Agent Kit ให้ขยาย DATA ENGINEERING > Orchestration Pipelines แล้วคลิก fraud_analysis_pipeline.yaml เพื่อเปิด Canvas ของ DAG แบบภาพ
  2. คลิกปุ่มเรียกใช้ไปป์ไลน์สีน้ำเงินที่มุมขวาบนของแถบเครื่องมือ Canvas
  3. ในเครื่องมือเลือกเมนูแบบเลื่อนลงของสภาพแวดล้อม ให้เลือก dev
  4. สังเกตการแจ้งเตือนความคืบหน้าในพื้นที่แสดงสถานะด้านล่าง (Running pipeline: Building pipeline locally...) ส่วนขยายจะคอมไพล์ DAG, แพ็กเกจ Notebook และชิ้นงาน dbt โดยอัตโนมัติ แล้วอัปโหลดไปยัง Bucket GCS ของสภาพแวดล้อม Airflow ที่มีการจัดการ (ขั้นตอนนี้ใช้เวลาประมาณ 3-4 นาที)

การติดตั้งใช้งานไปป์ไลน์จาก Canvas แบบภาพ

ตรวจสอบการเรียกใช้

เมื่อการรวบรวมข้อมูลในเครื่องเสร็จสมบูรณ์และการแจ้งเตือนแบบป๊อปอัปยืนยัน Triggered a new run for pipeline... successfully แล้ว ให้ตรวจสอบการดำเนินการจริง

  1. ในแถบด้านข้างของ Google Cloud Data Agent Kit ให้ขยาย DATA ENGINEERING > Orchestration Pipelines
  2. คลิกการจัดการไปป์ไลน์
  3. ในตารางการจัดการไปป์ไลน์ ให้คลิก fraud_analysis_pipeline เพื่อเปิดประวัติการดำเนินการ

ภาพรวมการจัดการไปป์ไลน์

  1. ในมุมมองประวัติการดำเนินการ ให้เลือกการเรียกใช้ที่ใช้งานอยู่จากปฏิทิน
  2. เมื่อการดำเนินการมีความคืบหน้าในแต่ละงานของไปป์ไลน์ (การนำเข้า การเปลี่ยนรูปแบบ dbt และการอนุมาน) ตัวบ่งชี้สถานะจะอัปเดตและระยะเวลาของงานจะแสดงขึ้น คลิกงานใดก็ได้เพื่อตรวจสอบเอาต์พุตการดำเนินการจริงและบันทึก DAG ของ Airflow

ประวัติการดำเนินการไปป์ไลน์แบบเรียลไทม์และรายละเอียดงาน

สรุปส่วน: คุณได้กำหนดค่าการเชื่อมต่อตัวจัดกำหนดการ Airflow, ทำให้ไปป์ไลน์การวิเคราะห์แบบครบวงจรใช้งานได้กับ Managed Airflow และตรวจสอบการดำเนินการที่ใช้งานจริง รวมถึงยืนยันระบบตั้งแต่บันทึกดิบไปจนถึงการคาดการณ์ขั้นสุดท้ายใน Cloud Spanner

9. ล้างข้อมูล

หากต้องการหลีกเลี่ยงการเรียกเก็บเงินอย่างต่อเนื่องในโปรเจ็กต์ที่อยู่ในระบบคลาวด์ของ Google สำหรับทรัพยากรที่ใช้ใน Codelab นี้ ให้ยกเลิกการใช้งานสภาพแวดล้อมโดยใช้สคริปต์อัตโนมัติ

  1. ในแผงเทอร์มินัล (หรือใน Cloud Shell) ให้ไปที่ไดเรกทอรีสคริปต์แล้วเรียกใช้คำสั่งต่อไปนี้
cd ~/devrel-demos/codelabs/agentic-data-labs/data-science/scripts
chmod +x teardown.sh
./teardown.sh
  1. สคริปต์จะแสดงทรัพยากรทั้งหมดที่วางแผนจะลบและแจ้งให้ยืนยัน
    • สภาพแวดล้อม Airflow ที่มีการจัดการ (cymbal-airflow)
    • อินสแตนซ์ Cloud Spanner (cymbal-fraud)
    • ชุดข้อมูล BigQuery (transactions_dataset_evals)
    • ที่เก็บข้อมูล Cloud Storage (gs://${PROJECT_ID}-fin-clearing-raw และ gs://${PROJECT_ID}-models)
    • บัญชีบริการของ Worker (composer-worker-sa)
  2. พิมพ์ y เพื่อยืนยัน สคริปต์การหยุดทำงานจะนำบริการ GCP ที่จัดสรรทั้งหมดออกและล้างไฟล์ในเครื่อง

10. ยินดีด้วย

คุณได้สร้างไปป์ไลน์การตรวจหาการประพฤติมิชอบแบบครบวงจรที่ครอบคลุม Cloud Storage, BigQuery, Managed Service สำหรับ Apache Spark (Spark Serverless), dbt, Cloud Spanner และ Managed Service สำหรับ Apache Airflow โดยใช้การเขียนโปรแกรมคู่กับ Google Cloud Data Agent Kit ภายใน Antigravity IDE

สิ่งที่คุณทำสำเร็จ

  1. 📥 บันทึกธุรกรรมดิบที่ส่งผ่านข้อมูลไปยังตาราง BigQuery โดยใช้ Managed Service สำหรับ Apache Spark และ Data Agent Kit
  2. 🧹 ข้อมูลที่ซ้ำกันและข้อมูลที่ทำให้เป็นมาตรฐานโดยการสร้างโปรเจ็กต์ dbt ด้วยการทดสอบคุณภาพของข้อมูล
  3. 🤖 ฝึกโมเดล Random Forest แบบกระจายโดยใช้ RandomForestClassifier และส่งออกโมเดลที่ฝึกแล้วไปยัง Cloud Storage
  4. ⚡ การอนุมานแบบกลุ่มที่ดำเนินการในธุรกรรมขาเข้าและกำหนดเส้นทางบันทึกที่มีความเสี่ยงสูงไปยัง Cloud Spanner เพื่อการตรวจสอบ
  5. 🔄 จัดระเบียบ จัดการ และตรวจสอบเวิร์กโฟลว์เป็น DAG ของ Airflow ที่กำหนดเวลาไว้โดยใช้ Managed Service สำหรับ Apache Airflow และเครื่องมือการจัดการ DAG แบบภาพของ IDE

หัวข้อสำคัญ

แนวคิด

สิ่งที่คุณได้เรียนรู้

ชุดเครื่องมือสำหรับตัวแทนด้านข้อมูล

การเขียนโค้ดร่วมกันภายใน IDE โดยใช้ภาษาธรรมชาติเพื่อสร้างสมุดบันทึก PySpark กำหนดค่าโมเดล dbt และกำหนด DAG ของ Airflow

BigQuery

ที่เก็บข้อมูลแบบตารางที่ปรับขนาดได้สำหรับ SQL เชิงวิเคราะห์, การแปลง dbt และการฝึก ML

Spark แบบ Serverless

การดำเนินการแบบ Serverless สำหรับการโหลดข้อมูล PySpark แบบกระจายและการฝึก ML แบบ Random Forest

ตัวเชื่อมต่อ Cloud Spanner

เขียนการคาดการณ์การอนุมานแบบกลุ่มของ Spark ลงในคิวการตรวจสอบฐานข้อมูลการดำเนินงานโดยตรง

การประกาศ DAG ใน YAML

คำจำกัดความของไปป์ไลน์แบบประกาศที่แสดงเป็นกราฟภาพ Airflow แบบอินเทอร์แอกทีฟใน IDE

การจัดการ DAG ด้วยภาพ

การตรวจสอบการขึ้นต่อกันของไปป์ไลน์ การติดตั้งใช้งานใน Managed Airflow และการตรวจสอบประวัติการดำเนินการของงานที่ใช้งานจริงภายใน IDE

ขั้นตอนถัดไป