استراتژی‌های حذف مجدد BigQuery – انجمن DEV

بیان مسأله

متن نوشته

• اجازه دهید فرض کنیم که خطوط لوله داده‌ای داریم که پیام‌ها را به جداول Google BigQuery می‌ریزند (اجازه دهید آنها را جداول خام بنامیم).
• ممکن است پیام های تکراری در جدول خام به دلایلی مانند:
  • پیام های تکراری ارسال شده از منبع
  • پیام چندین بار به دلیل مشکلات شبکه درج شد و مجدداً بین خط لوله داده و پرس و جو بزرگ (اگرچه می توان با استفاده از شناسه های درخواست منحصر به فرد در حین بارگیری داده ها در BQ تا حدودی این مشکل را برطرف کرد)
• BigQuery ایندکس منحصر به فرد ندارد

مفروضات

• پیام‌های تکراری دارای شناسه یکسانی هستند (شاید message_id)، اگرچه ممکن است با هر متادیتای اضافی که توسط خط لوله داده تزریق می‌شود، متفاوت باشند، به‌عنوان مثال: receive_timestamp
• پیام دارای یک مهر زمانی تجاری است که توسط سیستم های بالادستی ارائه می شود
• پیام‌ها ممکن است دیر برسد (شاید چند روز)
• فرض کنید یک جدول سفارشات خام با این فیلدها داریم:
  • message_id: شناسه منحصر به فرد شناسایی یک پیام
  • message_timestamp: مهر زمانی ارائه شده توسط سیستم های منبع
  • order_id: شناسه منحصر به فرد برای شناسایی یک سفارش
  • order_amount: مبلغ در مقابل سفارش
  • receive_timestamp: دریافت زمان تزریق شده توسط خط لوله جذب

هدف، واقعگرایانه

• این پیام‌ها باید به‌عنوان یک محصول داده به شکل حذف‌شده در معرض دید مصرف‌کنندگان قرار گیرند و مصرف‌کنندگان نباید نگران تکراری‌سازی باشند.
• اثربخشی هزینه. توجه: صورت‌حساب BQ بر اساس تعداد بایت‌های اسکن‌شده و ذخیره‌شده براساس مدل قیمت‌گذاری درخواستی انجام می‌شود.

راه حل 1

یکی از راه ها ایجاد نمایه ای است که ورودی های مجزا را از جدول خام انتخاب می کند. به عنوان مثال مشاهده SQL:

CREATE OR REPLACE VIEW views.orders AS
SELECT
  DISTINCT
    message_id,
    message_timestamp,
    order_id,
    order_amount
FROM raw.orders

اکنون یک مصرف کننده می تواند این نما را با یک SQL مانند: SELECT * FROM views.orders WHERE DATE(message_timestamp) BETWEEN '2023-01-01' AND '2023-01-31' برای دریافت سفارش های ثبت شده در ماه ژانویه.

اگر جدول سفارشات خام در قسمت message_timestamp پارتیشن بندی شده باشد، BiqQuery فقط ورودی های متعلق به پارتیشن ژانویه را پرس و جو می کند.

با این حال، اگر به انعطاف بیشتری در معیارهای منحصر به فرد نیاز داشته باشیم، کلمه کلیدی DISTINCT کمکی نمی کند. موارد استفاده مثال:

• receive_timestamp باید بخشی از view باشد: یک پیام ممکن است چندین ورودی با دریافت_timestamp های مختلف داشته باشد، در این صورت حذف مجدد توسط message_id کار نخواهد کرد. اگرچه در این مورد خاص می توان استدلال کرد که دریافت_timestamp نباید در معرض مصرف کنندگان قرار گیرد.
• در صورت وجود چندین ورودی با message_id یکسان، ورودی با آخرین زمان دریافت باید انتخاب شود

راه حل 2

ما می‌توانیم با استفاده از یک بند واجد شرایط در نما، انعطاف‌پذیری بیشتری در معیارهای منحصربه‌فرد ایجاد کنیم:

CREATE OR REPLACE VIEW views.orders AS
SELECT
  *
FROM raw.orders
QUALIFY ROW_NUMBER() OVER (PARTITION BY message_id ORDER BY receive_timestamp DESC) = 1

با این حال، اگر جدول توسط message_timestamp پارتیشن بندی شده باشد، کار می کند، و مصرف کنندگان با فیلتر کردن در message_timestamp با استفاده از پرس و جوی مانند: SELECT * FROM views.orders WHERE DATE(message_timestamp) BETWEEN '2023-01-01' AND '2023-01-31'، آن پارتیشن های اسکن شده را کاهش نمی دهد و BQ تمام پارتیشن ها را اسکن می کند.

اگر می توانستیم بند WHERE را قبل از بند QUALIFY قرار دهیم، کار می کرد، چیزی شبیه به:

CREATE OR REPLACE VIEW views.orders AS
SELECT
  *
FROM raw.orders
WHERE DATE(message_timestamp) BETWEEN '2023-01-01' AND '2023-01-31'
QUALIFY ROW_NUMBER() OVER (PARTITION BY message_id ORDER BY receive_timestamp DESC) = 1

اما هنگام تعریف view، پیام_timestamp دقیقی که باید برای فیلتر کردن استفاده شود را نمی‌دانیم، زیرا فقط در زمان اجرا شناخته می‌شود.

راه حل 3

اگر بتوانیم عبارت WHERE را قبل از عبارت QUALIFY داشته باشیم و مقادیر را در زمان اجرا تزریق کنیم، چه می‌شود؟

توابع جدول می توانند این کار را انجام دهند!

CREATE OR REPLACE TABLE FUNCTION functions.orders(start_date DATE, end_date DATE) AS 
SELECT 
  *
FROM raw.orders
WHERE DATE(message_timestamp) BETWEEN start_date AND end_date
QUALIFY ROW_NUMBER() OVER (PARTITION BY message_id ORDER BY receive_timestamp DESC) = 1

سپس مصرف کنندگان می توانند این تابع را به صورت زیر جستجو کنند:

SELECT * FROM functions.orders('2023-01-01', '2023-01-31') WHERE <other filters>

در اینجا فقط پارتیشن ژانویه توسط BQ اسکن می شود.

به طور خلاصه:

• ما می‌توانیم توابعی ایجاد کنیم (اجازه دهید آنها را توابع جدول خام بنامیم) در بالای جداول خام که محدوده‌های داده را به عنوان ورودی می‌پذیرند و آن پارامترها را قبل از QUALIFY به عبارت WHERE تزریق می‌کنند.
• بر اساس نیازهای تجاری، ممکن است در صورت نیاز، توابع مرتبه بالاتر را با ترکیب یک یا چند تابع جدول خام (در دامنه های مختلف) ایجاد کنیم تا تاریخ ورودی تا پایین ترین سطح منتشر شود.

راه حل 4-الف

آیا می توانیم معنایی یک جدول را حفظ کنیم؟ یک راه برای انجام این کار این است که:

• ابتدا داده های خام را در یک جدول خام تقسیم بندی شده بر اساس زمان دریافت بریزید
• برای برداشتن آخرین پارتیشن از جدول خام و ادغام آن در یک جدول بدون تکرار، یک پرس و جو زمانبندی شده را اجرا کنید (جدول dedulicated توسط message_timestamp پارتیشن بندی می شود)
• پارتیشن های قدیمی را از جدول خام حذف کنید

نمونه پرس و جوی برنامه ریزی شده ای که می تواند روزانه برای اسکن رکوردهای خام دیروز و ادغام آنها اجرا شود:

MERGE INTO dedup.orders dest USING (
SELECT 
  *
FROM raw_partitioned_by_receive_timestamp.orders
WHERE DATE(receive_timestamp) = CURRENT_DATE()-1
QUALIFY ROW_NUMBER() OVER (PARTITION BY message_id ORDER BY receive_timestamp DESC) = 1
) src
ON dest.message_id = src.message_id

WHEN NOT MATCHED THEN INSERT(message_id, message_timestamp, order_id, order_amount, receive_timestamp)
VALUES (src.message_id, src.message_timestamp, src.order_id, src.order_amount, src.receive_timestamp)

این درخواست برنامه ریزی شده اسکن می کند:

• فقط یک پارتیشن از جدول خام
• یک ستون (message_id زیرا بخشی از عبارت merge ON است) از تمام پارتیشن های جدول حذف شده. آیا می توانیم این را بهینه کنیم؟

همچنین توجه داشته باشید که در زمان پرس و جو (به عنوان مثال: هنگامی که یک مصرف کننده از جدول حذف شده درخواست می کند تا سفارشات را به مدت یک ماه دریافت کند)، پارتیشن های جدول حذف شده نیز بر اساس ستون پارتیشن بر اساس بند کجا اسکن می شوند.

راه حل 4-B

ما راه حل قبلی را با جایگزینی پرس و جوی ادغام زمان بندی شده با یک رویه ذخیره شده برنامه ریزی شده دو مرحله ای بهینه می کنیم.

1. به عنوان اولین گام، ما حداقل و حداکثر message_timestamp را از آخرین پارتیشن جدول خام به صورت زیر جستجو می کنیم:

SELECT
  DATE(min(message_timestamp)) AS minimum_message_timestamp,
  DATE(max(message_timestamp)) AS maximum_message_tiemstamp
FROM raw_partitioned_by_receive_timestamp.orders
WHERE
  DATE(receive_timestamp) = CURRENT_DATE()-1

1. ما میتوانیم:
   1. ورودی ها را از جدول حذف شده در پنجره minimal_message_timestamp و maximum_message_tiemstamp انتخاب کنید
   2. آخرین پارتیشن جدول خام را با ورودی های انتخاب شده در مرحله قبل مقایسه کنید تا ورودی های جدیدی را که باید به جدول حذف شده اضافه شوند و سپس فقط آن ورودی ها را درج کنید.

INSERT INTO dedup.orders
WITH existing AS (
  SELECT
    message_id
  FROM dedup.orders
  WHERE
    message_timestamp BETWEEN <<minimum_message_timestamp obtained in step 1>> AND <<maximum_message_timestamp obtained in step 1>>
)
SELECT
  raw.*
FROM raw_partitioned_by_receive_timestamp.orders raw
LEFT JOIN existing
  ON raw.message_id = existing.message_id
WHERE
  receive_timestamp = CURRENT_DATE()-1
  AND existing.message_id IS NULL

توجه: مهم است که این دو مرحله را به طور جداگانه انجام دهید و آنها را در یک پرس و جوی INSERT ترکیب نکنید، زیرا پارتیشن مبتنی بر جایی که فقط در صورتی سود هزینه را ارائه می دهد که عملوندها به صورت ایستا ارائه شوند.

این اسکن می کند:

• یک ستون (message_timestamp) آخرین پارتیشن جدول خام
• همه ستون ها از آخرین پارتیشن جدول خام
• یک ستون (message_id) از چند پارتیشن از جدول حذف شده. تعداد پارتیشن های اسکن شده به حداقل و حداکثر زمان پیام به دست آمده از آخرین پارتیشن جدول خام بستگی دارد.
• در زمان پرس و جو، پارتیشن های انتخاب شده از جدول حذف شده نیز اسکن می شوند

مبادلات

از نقطه نظر هزینه، این ماتریس تصمیم گیری است: