Realtime Compute for Apache Flink:Aggregation merge engine

Aggregation merge engine

• Merge Policy

  Engine ini mengelompokkan baris berdasarkan primary key yang sama dan menerapkan fungsi agregasi tertentu pada setiap bidang non-primary key.

• Konfigurasi

  Untuk mengaktifkan fitur ini, atur properti tabel 'table.merge-engine' = 'aggregation' saat membuat tabel.

• Penggunaan

  • Saat membuat tabel di Konsol, pilih kebijakan penggabungan (merge policy) dan gunakan konfigurasi lanjutan 'fields.<field_name>.agg' = '<function_name>' untuk menentukan fungsi agregasi untuk setiap bidang non-primary key.

  • Untuk fungsi agregasi yang memerlukan parameter tambahan, seperti delimiter kustom untuk listagg, gunakan format berikut:

    'fields.<field_name>.agg' = '<function_name>',
    'fields.<field_name>.<function_name>.<parameter_name>' = '<parameter_value>'

    Jika Anda tidak menentukan fungsi agregasi untuk suatu bidang, sistem menggunakan last_value_ignore_nulls sebagai perilaku agregasi default.

• Kasus Penggunaan

  • Menghitung total kumulatif dan metrik statistik

  • Memelihara pencacah (counter) dan ukuran (measure)

  • Memantau perubahan nilai maksimum atau minimum

  • Membangun dasbor waktu nyata dan analitik

• Batasan

  • Semantik retraction tidak didukung. Engine ini menggunakan logika agregasi satu arah dan tidak dapat membalikkan agregasi. Misalnya, engine tidak dapat mengurangi nilai yang dihapus dari hasil sum atau membatalkan operasi max untuk memulihkan nilai terbesar kedua.

  • Engine ini memberikan dukungan terbatas untuk operasi delete. Secara default, engine diam-diam mengabaikan operasi delete ('table.delete.behavior' = 'ignore').

  • Anda dapat mengonfigurasi engine agar menolak operasi delete dan mengembalikan error (disable) atau mengizinkan penghapusan fisik seluruh baris (allow).

    'table.delete.behavior' = 'ignore'   -- Nilai default. Diam-diam mengabaikan operasi delete.
    'table.delete.behavior' = 'disable'  -- Menolak operasi delete dan mengembalikan error.
    'table.delete.behavior' = 'allow'    -- Mengizinkan operasi delete.

Contoh

CREATE TABLE product_stats (
    product_id BIGINT,
    price DOUBLE,
    sales BIGINT,
    last_update_time TIMESTAMP(3),
    PRIMARY KEY (product_id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.price.agg' = 'max',
    'fields.sales.agg' = 'sum'
    -- Tidak ada fungsi agregasi yang ditentukan untuk last_update_time, sehingga last_value_ignore_nulls digunakan secara default.
);

-- Masukkan data. Catatan dengan primary key yang sama secara otomatis diagregasi.
INSERT INTO product_stats VALUES
    (1, 23.0, 15, TIMESTAMP '2024-01-01 10:00:00'),
    (1, 30.2, 20, TIMESTAMP '2024-01-01 11:00:00');

SELECT * FROM product_stats WHERE product_id = 1;
-- Output:
-- +------------+-------+-------+---------------------+
-- | product_id | price | sales | last_update_time    |
-- +------------+-------+-------+---------------------+
-- |          1 |  30.2 |    35 | 2024-01-01 11:00:00 |
-- +------------+-------+-------+---------------------+
-- price: 30.2 (nilai maksimum dari 23.0 dan 30.2)
-- sales: 35 (jumlah dari 15 dan 20)
-- last_update_time: 2024-01-01 11:00:00 (nilai non-null terakhir)

Fungsi agregasi yang didukung

CREATE TABLE test_sum (
    id BIGINT,
    amount DECIMAL(10, 2),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.amount.agg' = 'sum'
);

INSERT INTO test_sum VALUES (1, 100.50), (1, 200.75);
SELECT * FROM test_sum WHERE id = 1;
-- Output:
-- +----+---------+
-- | id | amount  |
-- +----+---------+
-- |  1 | 301.25  |
-- +----+---------+

CREATE TABLE test_product (
    id BIGINT,
    discount_factor DOUBLE,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.discount_factor.agg' = 'product'
);

INSERT INTO test_product VALUES (1, 0.9), (1, 0.8);
SELECT * FROM test_product WHERE id = 1;
-- Output:
-- +----+-----------------------+
-- | id | discount_factor       |
-- +----+-----------------------+
-- |  1 | 0.7200000000000001    |
-- +----+-----------------------+
-- Catatan: Karena aritmetika floating-point presisi ganda IEEE 754, hasilnya adalah 0.7200000000000001, bukan 0.72.
-- Untuk perhitungan yang presisi, kami merekomendasikan menggunakan tipe data DECIMAL daripada DOUBLE.

CREATE TABLE test_max (
    id BIGINT,
    temperature DOUBLE,
    reading_time TIMESTAMP(3),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.temperature.agg' = 'max',
    'fields.reading_time.agg' = 'max'
);

INSERT INTO test_max VALUES
    (1, 25.5, TIMESTAMP '2024-01-01 10:00:00'),
    (1, 28.3, TIMESTAMP '2024-01-01 11:00:00');
SELECT * FROM test_max WHERE id = 1;
-- Output:
-- +----+-------------+---------------------+
-- | id | temperature | reading_time        |
-- +----+-------------+---------------------+
-- |  1 |        28.3 | 2024-01-01 11:00:00 |
-- +----+-------------+---------------------+

CREATE TABLE test_min (
    id BIGINT,
    lowest_price DECIMAL(10, 2),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.lowest_price.agg' = 'min'
);

INSERT INTO test_min VALUES (1, 99.99), (1, 79.99), (1, 89.99);
SELECT * FROM test_min WHERE id = 1;
-- Output:
-- +----+--------------+
-- | id | lowest_price |
-- +----+--------------+
-- |  1 |        79.99 |
-- +----+--------------+

CREATE TABLE test_last_value (
    id BIGINT,
    status STRING,
    last_login TIMESTAMP(3),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.status.agg' = 'last_value',
    'fields.last_login.agg' = 'last_value'
);

INSERT INTO test_last_value VALUES (1, 'online', TIMESTAMP '2024-01-01 10:00:00');
INSERT INTO test_last_value VALUES (1, 'offline', TIMESTAMP '2024-01-01 11:00:00');
INSERT INTO test_last_value VALUES (1, CAST(NULL AS STRING), TIMESTAMP '2024-01-01 12:00:00');
-- Nilai null menimpa nilai 'offline' sebelumnya.
SELECT * FROM test_last_value WHERE id = 1;
-- Output:
-- +----+--------+---------------------+
-- | id | status | last_login          |
-- +----+--------+---------------------+
-- |  1 | null   | 2024-01-01 12:00:00 |
-- +----+--------+---------------------+

CREATE TABLE test_last_value_ignore_nulls (
    id BIGINT,
    email STRING,
    phone STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.email.agg' = 'last_value_ignore_nulls',
    'fields.phone.agg' = 'last_value_ignore_nulls'
);

INSERT INTO test_last_value_ignore_nulls VALUES (1, 'user@example.com', '123-456');
INSERT INTO test_last_value_ignore_nulls VALUES (1, CAST(NULL AS STRING), '789-012');
-- Nilai null untuk email diabaikan, dan nilai sebelumnya dipertahankan.
INSERT INTO test_last_value_ignore_nulls VALUES (1, 'new@example.com', CAST(NULL AS STRING));
-- Nilai null untuk phone diabaikan, dan nilai sebelumnya dipertahankan.
SELECT * FROM test_last_value_ignore_nulls WHERE id = 1;
-- Output:
-- +----+-----------------+---------+
-- | id | email           | phone   |
-- +----+-----------------+---------+
-- |  1 | new@example.com | 789-012 |
-- +----+-----------------+---------+

CREATE TABLE test_first_value (
    id BIGINT,
    first_purchase_date DATE,
    first_product STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.first_purchase_date.agg' = 'first_value',
    'fields.first_product.agg' = 'first_value'
);

INSERT INTO test_first_value VALUES (1, '2024-01-01', 'ProductA'), (1, '2024-02-01', 'ProductB');
-- Catatan kedua diabaikan, dan nilai pertama disimpan.
SELECT * FROM test_first_value WHERE id = 1;
-- Output:
-- +----+---------------------+---------------+
-- | id | first_purchase_date | first_product |
-- +----+---------------------+---------------+
-- |  1 | 2024-01-01          | ProductA      |
-- +----+---------------------+---------------+

CREATE TABLE test_listagg (
    id BIGINT,
    tags1 STRING,
    tags2 STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.tags1.agg' = 'listagg',
    'fields.tags2.agg' = 'listagg',
    'fields.tags2.listagg.delimiter' = ';'   -- Tentukan delimiter kustom.
);

INSERT INTO test_listagg VALUES (1, 'developer', 'developer'), (1, 'java', 'java'), (1, 'flink', 'flink');
SELECT * FROM test_listagg WHERE id = 1;
-- Output:
-- +----+----------------------+----------------------+
-- | id | tags1                | tags2                |
-- +----+----------------------+----------------------+
-- |  1 | developer,java,flink | developer;java;flink |
-- +----+----------------------+----------------------+

CREATE TABLE test_string_agg (
    id BIGINT,
    tags STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.tags.agg' = 'string_agg',
    'fields.tags.string_agg.delimiter' = ';'   -- Tentukan delimiter kustom.
);

INSERT INTO test_string_agg VALUES (1, 'developer'), (1, 'java'), (1, 'flink');
SELECT * FROM test_string_agg WHERE id = 1;
-- Output:
-- +----+----------------------+
-- | id | tags                 |
-- +----+----------------------+
-- |  1 | developer;java;flink |
-- +----+----------------------+

CREATE TABLE user_visits (
    user_id BIGINT,
    visit_bitmap BYTES,
    PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.visit_bitmap.agg' = 'rbm32'
);

-- Masukkan nilai RoaringBitmap yang telah diserialisasi (literal heksadesimal).
-- Bitmap {1,2}
INSERT INTO user_visits VALUES (1, x'3A30000001000000000001001000000001000200');
-- Bitmap {2,3}
INSERT INTO user_visits VALUES (1, x'3A30000001000000000001001000000002000300');

SELECT * FROM user_visits WHERE user_id = 1;
-- Hasil: visit_bitmap berisi union {1,2,3}.

CREATE TABLE session_interactions (
    session_id BIGINT,
    interaction_bitmap BYTES,
    PRIMARY KEY (session_id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.interaction_bitmap.agg' = 'rbm64'
);

-- Masukkan nilai Roaring64Bitmap yang telah diserialisasi.
-- Hasil: interaction_bitmap berisi union dari semua bitmap yang dimasukkan.

CREATE TABLE test_bool_and (
    id BIGINT,
    has_all_permissions BOOLEAN,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.has_all_permissions.agg' = 'bool_and'
);

INSERT INTO test_bool_and VALUES (1, true), (1, true), (1, false);
SELECT * FROM test_bool_and WHERE id = 1;
-- Output:
-- +----+---------------------+
-- | id | has_all_permissions |
-- +----+---------------------+
-- |  1 | false               |
-- +----+---------------------+

CREATE TABLE test_bool_or (
    id BIGINT,
    has_any_alert BOOLEAN,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'table.merge-engine' = 'aggregation',
    'fields.has_any_alert.agg' = 'bool_or'
);

INSERT INTO test_bool_or VALUES (1, false), (1, false), (1, true);
SELECT * FROM test_bool_or WHERE id = 1;
-- Output:
-- +----+---------------+
-- | id | has_any_alert |
-- +----+---------------+
-- |  1 | true          |
-- +----+---------------+