Mengisi ulang data historis dengan alur

Dalam rekayasa data, pengisian ulang mengacu pada proses pemrosesan data historis secara retroaktif melalui alur data yang dirancang untuk memproses data saat ini atau streaming.

Biasanya, ini adalah alur terpisah yang mengirim data ke dalam tabel yang ada. Ilustrasi berikut menunjukkan aliran pengisian ulang yang mengirim data historis ke tabel perunggu dalam pipa saluran Anda.

Aliran isi ulang menambahkan data historis ke alur kerja yang sudah ada

Beberapa skenario yang mungkin memerlukan isi ulang:

  • Proses data historis dari sistem warisan untuk melatih model pembelajaran mesin (ML) atau membangun dasbor analisis tren historis.
  • Memproses ulang subkumpulan data karena masalah kualitas data dengan sumber data hulu.
  • Persyaratan bisnis Anda berubah dan Anda perlu mengisi ulang data untuk periode waktu yang berbeda yang tidak tercakup oleh alur awal.
  • Logika bisnis Anda berubah dan Anda perlu memproses ulang data historis dan saat ini.

Isi ulang di alur Lakeflow didukung dengan alur tambahan khusus yang menggunakan ONCE opsi . Lihat append_flow atau CREATE FLOW (pipelines) untuk informasi selengkapnya tentang opsi tersebut ONCE .

Pertimbangan saat mengisi ulang data historis ke dalam tabel streaming

  • Biasanya, tambahkan data ke tabel streaming perunggu. Lapisan perak dan emas hilir akan mengambil data baru dari lapisan perunggu.
  • Pastikan alur Anda dapat menangani data duplikat dengan baik jika data yang sama ditambahkan beberapa kali.
  • Pastikan skema data historis kompatibel dengan skema data saat ini.
  • Pertimbangkan ukuran volume data dan waktu pemrosesan SLA yang diperlukan, dan dengan demikian mengonfigurasi ukuran kluster dan batch.

Contoh: Menambahkan isi ulang ke alur yang sudah ada

Dalam contoh ini, katakanlah Anda memiliki alur yang menyerap data pendaftaran peristiwa mentah dari sumber penyimpanan cloud, mulai 01 Jan 2025. Anda kemudian menyadari bahwa Anda ingin menambahkan data historis tiga tahun sebelumnya untuk keperluan pelaporan dan analisis data hilir. Semua data berada dalam satu lokasi, dipartisi menurut tahun, bulan, dan hari, dalam format JSON.

Alur awal

Berikut adalah kode alur awal yang secara bertahap menyerap data pendaftaran peristiwa mentah dari penyimpanan cloud.

Phyton

from pyspark import pipelines as dp

source_root_path = spark.conf.get("registration_events_source_root_path")
begin_year = spark.conf.get("begin_year")
incremental_load_path = f"{source_root_path}/*/*/*"

# create a streaming table and the default flow to ingest streaming events
@dp.table(name="registration_events_raw", comment="Raw registration events")
def ingest():
    return (
        spark
        .readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.inferColumnTypes", "true")
        .option("cloudFiles.maxFilesPerTrigger", 100)
        .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
        .option("modifiedAfter", "2025-01-01T00:00:00.000+00:00")
        .load(incremental_load_path)
        .where(f"year(timestamp) >= {begin_year}") # safeguard to not process data before begin_year
    )

SQL

-- create a streaming table and the default flow to ingest streaming events
CREATE OR REFRESH STREAMING LIVE TABLE registration_events_raw AS
SELECT * FROM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/*/*/*",
  format => "json",
  inferColumnTypes => true,
  maxFilesPerTrigger => 100,
  schemaEvolutionMode => "addNewColumns",
  modifiedAfter => "2024-12-31T23:59:59.999+00:00"
)
WHERE year(timestamp) >= '2025'; -- safeguard to not process data before begin_year

Di sini kami menggunakan modifiedAfter opsi Auto Loader untuk memastikan kami tidak memproses semua data dari jalur penyimpanan cloud. Pemrosesan bertahap dihentikan pada batas itu.

Petunjuk / Saran

Sumber data lainnya, seperti Kafka, Kinesis, dan Azure Event Hubs memiliki opsi pembaca yang setara untuk mencapai perilaku yang sama.

Mengisi ulang data dari 3 tahun sebelumnya

Sekarang Anda ingin menambahkan satu atau beberapa alur untuk mengisi ulang data sebelumnya. Dalam contoh ini, lakukan langkah-langkah berikut:

  • Gunakan alur append once. Ini melakukan pengisian ulang satu kali tanpa terus berjalan setelah isi ulang pertama. Kode tetap berada di alur kerja Anda, dan jika alur kerja dimuat ulang sepenuhnya, pemulihan data akan dijalankan kembali.
  • Buat tiga aliran pengisian ulang, satu untuk setiap tahun (dalam hal ini, data dibagi per tahun di jalur). Untuk Python, kami membuat parameter pembuatan alur, tetapi di SQL kami mengulangi kode tiga kali, sekali untuk setiap alur.

Bila Anda mengerjakan proyek sendiri dan tidak menggunakan komputasi tanpa server, Anda mungkin ingin memperbarui pekerja maksimum untuk jalur pemrosesan. Meningkatkan jumlah pekerja maksimum memastikan Anda memiliki sumber daya untuk memproses data historis sekaligus memproses data streaming saat ini sesuai SLA yang diharapkan.

Petunjuk / Saran

Jika Anda menggunakan komputasi tanpa server dengan penskalaan otomatis yang ditingkatkan (default), maka kluster Anda secara otomatis meningkat ukurannya saat beban Anda meningkat.

Phyton

from pyspark import pipelines as dp

source_root_path = spark.conf.get("registration_events_source_root_path")
begin_year = spark.conf.get("begin_year")
backfill_years = spark.conf.get("backfill_years") # e.g. "2024,2023,2022"
incremental_load_path = f"{source_root_path}/*/*/*"

# meta programming to create append once flow for a given year (called later)
def setup_backfill_flow(year):
    backfill_path = f"{source_root_path}/year={year}/*/*"
    @dp.append_flow(
        target="registration_events_raw",
        once=True,
        name=f"flow_registration_events_raw_backfill_{year}",
        comment=f"Backfill {year} Raw registration events")
    def backfill():
        return (
            spark
            .read
            .format("json")
            .option("inferSchema", "true")
            .load(backfill_path)
        )

# create the streaming table
dp.create_streaming_table(name="registration_events_raw", comment="Raw registration events")

# append the original incremental, streaming flow
@dp.append_flow(
        target="registration_events_raw",
        name="flow_registration_events_raw_incremental",
        comment="Raw registration events")
def ingest():
    return (
        spark
        .readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.inferColumnTypes", "true")
        .option("cloudFiles.maxFilesPerTrigger", 100)
        .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
        .option("modifiedAfter", "2024-12-31T23:59:59.999+00:00")
        .load(incremental_load_path)
        .where(f"year(timestamp) >= {begin_year}")
    )

# parallelize one time multi years backfill for faster processing
# split backfill_years into array
for year in backfill_years.split(","):
    setup_backfill_flow(year) # call the previously defined append_flow for each year

SQL

-- create the streaming table
CREATE OR REFRESH STREAMING TABLE registration_events_raw;

-- append the original incremental, streaming flow
CREATE FLOW
  registration_events_raw_incremental
AS INSERT INTO
  registration_events_raw BY NAME
SELECT * FROM STREAM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/*/*/*",
  format => "json",
  inferColumnTypes => true,
  maxFilesPerTrigger => 100,
  schemaEvolutionMode => "addNewColumns",
  modifiedAfter => "2024-12-31T23:59:59.999+00:00"
)
WHERE year(timestamp) >= '2025';


-- one time backfill 2024
CREATE FLOW
  registration_events_raw_backfill_2024
AS INSERT INTO ONCE
  registration_events_raw BY NAME
SELECT * FROM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/year=2024/*/*",
  format => "json",
  inferColumnTypes => true
);

-- one time backfill 2023
CREATE FLOW
  registration_events_raw_backfill_2023
AS INSERT INTO ONCE
  registration_events_raw BY NAME
SELECT * FROM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/year=2023/*/*",
  format => "json",
  inferColumnTypes => true
);

-- one time backfill 2022
CREATE FLOW
  registration_events_raw_backfill_2022
AS INSERT INTO ONCE
  registration_events_raw BY NAME
SELECT * FROM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/year=2022/*/*",
  format => "json",
  inferColumnTypes => true
);

Implementasi ini menyoroti beberapa pola penting.

Pemisahan kekhawatiran

  • Pemrosesan inkremental tidak bergantung pada operasi backfill.
  • Setiap alur memiliki pengaturan konfigurasi dan pengoptimalannya sendiri.
  • Ada perbedaan yang jelas antara operasi inkremental dan isi ulang.

Eksekusi terkontrol

  • Menggunakan opsi ONCE memastikan bahwa setiap pemenuhan kembali berjalan persis sekali.
  • Aliran isi ulang tetap berada di grafik alur, tetapi menjadi tidak aktif setelah selesai. Ini siap secara otomatis untuk digunakan pada pembaruan penuh.
  • Ada jejak audit yang jelas dari operasi isi ulang dalam definisi alur.

Pengoptimalan pemrosesan

  • Anda dapat membagi isi ulang besar menjadi beberapa backfill yang lebih kecil untuk pemrosesan yang lebih cepat, atau untuk mengontrol pemrosesan.
  • Menggunakan autoscaling yang ditingkatkan secara dinamis menskalakan ukuran kluster berdasarkan beban kluster saat ini.

Evolusi skema

  • Menggunakan schemaEvolutionMode="addNewColumns" menangani perubahan skema dengan lancar.
  • Anda memiliki inferensi skema yang konsisten di seluruh data historis dan saat ini.
  • Ada penanganan kolom baru yang aman dalam data yang lebih baru.

Contoh: Mengisi ulang target SCD selama migrasi

Skenario migrasi yang umum adalah tabel dimensi yang berubah perlahan (SCD) yang sudah ada dalam sistem warisan dengan sejarah akumulasi bertahun-tahun, tetapi feed perubahan aslinya sudah tidak tersedia lagi. Karena peristiwa perubahan asli sudah hilang, Anda memutar ulang riwayat tabel lama itu ke target baru AUTO CDC satu kali, lalu menghubungkan feed CDC baru untuk selanjutnya. Untuk informasi lebih lanjut tentang AUTO CDC dan tipe SCD, lihat API AUTO CDC: Menyederhanakan penangkapan data perubahan dengan alur.

Pola ini merupakan aliran sekali jalan AUTO CDC ke tabel streaming yang sama yang ditargetkan oleh aliran AUTO CDC yang sedang berlangsung. Target AUTO CDC hanya menerima aliran AUTO CDC, jadi seed juga harus berupa aliran AUTO CDC. Alur append sederhana INSERT INTO ONCE ke tabel yang sama tidak lolos validasi:

  1. Buat tabel streaming target yang akan digunakan oleh alur Anda AUTO CDC .
  2. Inisialisasi riwayat lama satu kali menggunakan alur AUTO CDC ONCE yang membaca tabel SCD lama sebagai stream, diurutkan berdasarkan kolom awal validitas lama. Mainkan ulang baris legacy sebagai event perubahan daripada membentuknya sendiri. AUTO CDC membuat kolom riwayat __START_AT dan __END_AT untuk target SCD Tipe 2, jadi jangan tulis kolom-kolom tersebut secara langsung.
  3. Sambungkan AUTO CDC aliran yang sedang berjalan yang membaca umpan perubahan terbaru. AUTO CDC menentukan pengurutan per kunci, jadi proses peralihan harus dilakukan untuk tiap kunci bisnis secara terpisah: perubahan live pertama untuk setiap kunci harus diurutkan setelah perubahan hasil seeding terakhir untuk kunci yang sama. Nilai urutan yang hanya lebih lambat dari maksimum warisan global masih dapat dianggap basi untuk satu kunci individual, dan perubahan langsung pertama kunci tersebut kemudian diabaikan atau diurutkan secara salah.

Kode berikut membuat tabel streaming yang menggunakan langkah-langkah di atas:

CREATE OR REFRESH STREAMING TABLE customers_history;

-- One-time seed: replay the legacy history as change events
CREATE FLOW customers_history_seed
AS AUTO CDC ONCE INTO customers_history
FROM stream(legacy.customers_scd2)
KEYS (customer_id)
SEQUENCE BY valid_from
STORED AS SCD TYPE 2;

-- Ongoing live CDC into the same target
CREATE FLOW customers_history_cdc
AS AUTO CDC INTO customers_history
FROM stream(customers_cdc_bronze)
KEYS (customer_id)
SEQUENCE BY change_timestamp
STORED AS SCD TYPE 2;

Kedua alur harus sama dalam hal kunci, tipe SCD, dan tipe data kolom pengurutan masing-masing. Dalam contoh sebelumnya, kedua aliran diurutkan berdasarkan stempel waktu, yang menggunakan satu waktu peralihan untuk memisahkan data historis awal dari aliran data langsung. Jika tabel lama mengurutkan berdasarkan nilai dengan tipe yang berbeda dari live feed, lakukan cast pada salah satunya agar tipenya cocok.

Bentuk yang sama berlaku untuk target SCD Tipe 1: ubah STORED AS SCD TYPE 2 menjadi STORED AS SCD TYPE 1 di kedua alur, dan target hanya menyimpan baris terkini untuk setiap kunci. Sebelum bergantung pada salah satu format, validasikan pada sampel kunci bahwa perubahan langsung pertama untuk kunci yang di-seed menghasilkan tepat satu versi baru dan menutup versi sebelumnya secara benar. Celah urutan per tombol biasanya muncul pada langkah tersebut.

Sumber daya tambahan

  • Last updated on