diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index f6c4b2e7..a535e485 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -88,8 +88,9 @@ jobs: - example: iceberg-dedup path: iceberg-data-deduplication test_commands: | - compile data_compaction_package.json - compile data_delete_package.json + compile datagen_package_local.json + compile data_compaction_package_local.json + compile data_delete_package_local.json - example: iot-sensor path: iot-sensor-metrics diff --git a/.gitignore b/.gitignore index 2d906009..ad6aded2 100644 --- a/.gitignore +++ b/.gitignore @@ -8,6 +8,11 @@ cache */myenv warehouse +# Performance benchmarking +node_modules/ +iceberg-data/ +sqrl-iceberg-data/ + # Created by https://www.toptal.com/developers/gitignore/api/python # Edit at https://www.toptal.com/developers/gitignore?templates=python @@ -225,8 +230,3 @@ Temporary Items py-venv # End of https://www.toptal.com/developers/gitignore/api/macos - -##performance benchmarking -node_modules/ -iceberg-data/ -sqrl-iceberg-data/ diff --git a/iceberg-data-deduplication/data_compaction.sqrl b/iceberg-data-deduplication/data_compaction.sqrl index e57e3cca..e6845efd 100644 --- a/iceberg-data-deduplication/data_compaction.sqrl +++ b/iceberg-data-deduplication/data_compaction.sqrl @@ -1,8 +1,8 @@ -- This script runs a deduplication compaction on Iceberg tables -- It currently assumes a single primary partition column and a single time bucket partition column based on seconds -IMPORT tables.{{tableName}}.*; -IMPORT tables.compaction_tracker.*; +IMPORT tables.data_{{env}}.*; +IMPORT tables.tracker_{{env}}.*; IMPORT udfs.read_partition_sizes; -- Collect previously compacted partitions @@ -73,7 +73,7 @@ _PartitionsToCompact.partition_id := CAST(partition_id AS {{partitionColType}}); _InputData := SELECT /*+ BROADCAST(c) */ d.* -FROM {{tableName}} AS d +FROM DedupData AS d JOIN _PartitionsToCompact AS c ON d.{{partitionCol}} = c.partition_id AND d.{{timeBucketCol}} <= c.max_time_bucket; @@ -82,7 +82,7 @@ _InputData.{{timeBucketCol}} := 0; _DistInputData := DISTINCT _InputData ON {{distinctKeyCol}} ORDER BY {{timestampCol}} DESC; -INSERT OVERWRITE {{tableName}} +INSERT OVERWRITE DedupData SELECT * FROM _DistInputData -- filter only when "deleteFlagCol" has a value {{#deleteFlagCol}} diff --git a/iceberg-data-deduplication/data_compaction_package.json b/iceberg-data-deduplication/data_compaction_package_local.json similarity index 97% rename from iceberg-data-deduplication/data_compaction_package.json rename to iceberg-data-deduplication/data_compaction_package_local.json index b9caff94..1ffc6d3f 100644 --- a/iceberg-data-deduplication/data_compaction_package.json +++ b/iceberg-data-deduplication/data_compaction_package_local.json @@ -4,6 +4,7 @@ "script": { "main": "data_compaction.sqrl", "config": { + "env": "local", "warehouse": "sqrl-iceberg-data", "catalogType": "hadoop", "catalogName": "mycatalog", diff --git a/iceberg-data-deduplication/data_compaction_package_test.json b/iceberg-data-deduplication/data_compaction_package_test.json new file mode 100644 index 00000000..f664f867 --- /dev/null +++ b/iceberg-data-deduplication/data_compaction_package_test.json @@ -0,0 +1,34 @@ +{ + "version": "1", + "enabled-engines": ["flink"], + "script": { + "main": "data_compaction.sqrl", + "config": { + "env": "test", + "warehouse": "s3://sqrl-examples-data-bucket/datasqrl-examples/iceberg-canary", + "catalogType": "glue", + "catalogName": "canary", + "databaseName": "canary", + "tableName": "iceberg_dedup_test", + "distinctKeyCol": "account_id", + "partitionCol": "account_id", + "partitionColType": "bigint", + "timeBucketCol": "time_bucket", + "timestampCol": "ts", + "deleteFlagCol": null, + "sortKey": null, + "bufferSeconds": 2, + "bucketSeconds": 5, + "newDataNormalizerInMB": 1, + "maxTotalPartitionSizeInMB": 10 + } + }, + "engines": { + "flink": { + "config": { + "execution.runtime-mode": "batch", + "table.exec.source.idle-timeout": "1 ms" + } + } + } +} diff --git a/iceberg-data-deduplication/data_delete.sqrl b/iceberg-data-deduplication/data_delete.sqrl index 6f3af526..364a4ee6 100644 --- a/iceberg-data-deduplication/data_delete.sqrl +++ b/iceberg-data-deduplication/data_delete.sqrl @@ -1,4 +1,4 @@ -IMPORT tables.compaction_tracker.*; +IMPORT tables.tracker_{{env}}.*; IMPORT udfs.delete_duplicated_data; _DataToDelete := diff --git a/iceberg-data-deduplication/data_delete_package.json b/iceberg-data-deduplication/data_delete_package_local.json similarity index 95% rename from iceberg-data-deduplication/data_delete_package.json rename to iceberg-data-deduplication/data_delete_package_local.json index 5fca7cc6..1f33b037 100644 --- a/iceberg-data-deduplication/data_delete_package.json +++ b/iceberg-data-deduplication/data_delete_package_local.json @@ -4,6 +4,7 @@ "script": { "main": "data_delete.sqrl", "config": { + "env": "local", "warehouse": "sqrl-iceberg-data", "catalogType": "hadoop", "catalogName": "mycatalog", diff --git a/iceberg-data-deduplication/data_delete_package_test.json b/iceberg-data-deduplication/data_delete_package_test.json new file mode 100644 index 00000000..527758f0 --- /dev/null +++ b/iceberg-data-deduplication/data_delete_package_test.json @@ -0,0 +1,23 @@ +{ + "version": "1", + "enabled-engines": ["flink"], + "script": { + "main": "data_delete.sqrl", + "config": { + "env": "test", + "warehouse": "s3://sqrl-examples-data-bucket/datasqrl-examples/iceberg-canary", + "catalogType": "glue", + "catalogName": "canary", + "databaseName": "canary", + "commitWaitSec": 1, + "stopDeleteFilter": "" + } + }, + "engines": { + "flink": { + "config": { + "execution.runtime-mode": "batch" + } + } + } +} diff --git a/iceberg-data-deduplication/datagen.sqrl b/iceberg-data-deduplication/datagen.sqrl index 93a320c7..29b1d542 100644 --- a/iceberg-data-deduplication/datagen.sqrl +++ b/iceberg-data-deduplication/datagen.sqrl @@ -1,8 +1,8 @@ IMPORT tables.min_source AS _CDC; -IMPORT tables.{{tableName}}.*; +IMPORT tables.data_{{env}}.*; CountRecords := SELECT COUNT(*) AS num_records FROM _CDC.SrcRecords; EXPORT CountRecords TO logger.records; -INSERT INTO {{tableName}} SELECT c.*, EXTRACT(SECOND FROM c.ts) + 1 AS time_bucket FROM _CDC.SrcRecords c; +INSERT INTO DedupData SELECT c.*, EXTRACT(SECOND FROM c.ts) + 1 AS time_bucket FROM _CDC.SrcRecords c; diff --git a/iceberg-data-deduplication/datagen_package.json b/iceberg-data-deduplication/datagen_package_local.json similarity index 95% rename from iceberg-data-deduplication/datagen_package.json rename to iceberg-data-deduplication/datagen_package_local.json index f916c5dd..2c415ee8 100644 --- a/iceberg-data-deduplication/datagen_package.json +++ b/iceberg-data-deduplication/datagen_package_local.json @@ -4,6 +4,7 @@ "script": { "main": "datagen.sqrl", "config": { + "env": "local", "warehouse": "sqrl-iceberg-data", "catalogType": "hadoop", "catalogName": "mycatalog", diff --git a/iceberg-data-deduplication/datagen_package_test.json b/iceberg-data-deduplication/datagen_package_test.json new file mode 100644 index 00000000..02dc1576 --- /dev/null +++ b/iceberg-data-deduplication/datagen_package_test.json @@ -0,0 +1,24 @@ +{ + "version": "1", + "enabled-engines": ["flink"], + "script": { + "main": "datagen.sqrl", + "config": { + "env": "test", + "warehouse": "s3://sqrl-examples-data-bucket/datasqrl-examples/iceberg-canary", + "catalogType": "glue", + "catalogName": "canary", + "databaseName": "canary", + "tableName": "iceberg_dedup_test" + } + }, + "engines": { + "flink": { + "config": { + "execution.runtime-mode": "batch", + "execution.checkpointing.interval": "8 s", + "table.exec.source.idle-timeout": "1 ms" + } + } + } +} diff --git a/iceberg-data-deduplication/tables/cdc_data.sqrl b/iceberg-data-deduplication/tables/data_local.sqrl similarity index 95% rename from iceberg-data-deduplication/tables/cdc_data.sqrl rename to iceberg-data-deduplication/tables/data_local.sqrl index e4932321..2f3870dd 100644 --- a/iceberg-data-deduplication/tables/cdc_data.sqrl +++ b/iceberg-data-deduplication/tables/data_local.sqrl @@ -1,4 +1,4 @@ -CREATE TABLE `cdc_data` ( +CREATE TABLE `DedupData` ( `account_id` BIGINT NOT NULL, `id` BIGINT NOT NULL, `pt` STRING NOT NULL, diff --git a/iceberg-data-deduplication/tables/data_test.sqrl b/iceberg-data-deduplication/tables/data_test.sqrl new file mode 100644 index 00000000..7cd3e0af --- /dev/null +++ b/iceberg-data-deduplication/tables/data_test.sqrl @@ -0,0 +1,20 @@ +CREATE TABLE `DedupData` ( + `account_id` BIGINT NOT NULL, + `id` BIGINT NOT NULL, + `pt` STRING NOT NULL, + `op` STRING NOT NULL, + `payload` STRING NOT NULL, + `ts` TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL, + `time_bucket` BIGINT NOT NULL, + PRIMARY KEY (`account_id`, `id`) NOT ENFORCED +) PARTITIONED BY (`account_id`, `id`, `time_bucket`) +WITH ( + 'connector' = 'iceberg', + 'catalog-name' = '{{catalogName}}', + 'catalog-database' = '{{databaseName}}', + 'catalog-table' = '{{tableName}}', + 'catalog-impl' = 'org.apache.iceberg.aws.glue.GlueCatalog', + 'io-impl' = 'org.apache.iceberg.aws.s3.S3FileIO', + 'warehouse' = '{{warehouse}}', + 'format-version' = '2' +); diff --git a/iceberg-data-deduplication/tables/compaction_tracker.sqrl b/iceberg-data-deduplication/tables/tracker_local.sqrl similarity index 100% rename from iceberg-data-deduplication/tables/compaction_tracker.sqrl rename to iceberg-data-deduplication/tables/tracker_local.sqrl diff --git a/iceberg-data-deduplication/tables/tracker_test.sqrl b/iceberg-data-deduplication/tables/tracker_test.sqrl new file mode 100644 index 00000000..de199793 --- /dev/null +++ b/iceberg-data-deduplication/tables/tracker_test.sqrl @@ -0,0 +1,22 @@ +CREATE TABLE `CompactionTracker` ( + `table_name` STRING NOT NULL, + `job_id` STRING NOT NULL, + `partition_id` STRING NOT NULL, + `partition_col_name` STRING NOT NULL, + `partition_time_bucket` BIGINT, + `new_rel_percentage` DOUBLE NOT NULL, + `new_abs_percentage` DOUBLE NOT NULL, + `max_time_bucket` BIGINT NOT NULL, + `action` STRING NOT NULL, + `action_time` TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL +) PARTITIONED BY (`table_name`) +WITH ( + 'connector' = 'iceberg', + 'catalog-name' = '{{catalogName}}', + 'catalog-database' = '{{databaseName}}', + 'catalog-table' = 'compaction_tracker', + 'catalog-impl' = 'org.apache.iceberg.aws.glue.GlueCatalog', + 'io-impl' = 'org.apache.iceberg.aws.s3.S3FileIO', + 'warehouse' = '{{warehouse}}', + 'format-version' = '2' +);