Skip to content

Commit e97df5f

Browse files
RyanL1997YANG-DBSwiddis
authored
[Manual Backport 2.x] update vpc flow with flint-s3 based DDL assets and dashboard (#1721)
* update vpc flow with flint-s3 based DDL assets and dashboard (#1691) * update vpc flow with flint-s3 based DDL assets and dashboard Signed-off-by: YANGDB <[email protected]> * update MV to use auto sync refresh Signed-off-by: YANGDB <[email protected]> * Update server/adaptors/integrations/__data__/repository/aws_vpc_flow/assets/vpc_agg_30min_connections_mv-1.0.0.sql Co-authored-by: Simeon Widdis <[email protected]> Signed-off-by: YANGDB <[email protected]> * Update server/adaptors/integrations/__data__/repository/aws_vpc_flow/assets/vpc_window-agg_60min_dest_ip_total-bytes_mv-1.0.0.sql Co-authored-by: Simeon Widdis <[email protected]> Signed-off-by: YANGDB <[email protected]> * Update server/adaptors/integrations/__data__/repository/aws_vpc_flow/assets/vpc_window-agg_60min_dest_ip_cardinality_mv-1.0.0.sql Co-authored-by: Simeon Widdis <[email protected]> Signed-off-by: YANGDB <[email protected]> * Update server/adaptors/integrations/__data__/repository/aws_vpc_flow/assets/vpc_live_week_mv-1.0.0.sql Co-authored-by: Simeon Widdis <[email protected]> Signed-off-by: YANGDB <[email protected]> * Update server/adaptors/integrations/__data__/repository/aws_vpc_flow/assets/vpc_agg_60min_connections_mv-1.0.0.sql Co-authored-by: Simeon Widdis <[email protected]> Signed-off-by: YANGDB <[email protected]> * remove comments from MV queries Signed-off-by: YANGDB <[email protected]> * update to remove HIVE as not supported yet in current EMR version Signed-off-by: YANGDB <[email protected]> * update refresh sync rate Signed-off-by: YANGDB <[email protected]> * update watermark_delay Signed-off-by: YANGDB <[email protected]> * add live only workflow & dashboard add pre-agg + live workflow & dashboard add refresh-workflow for the pre-agg queries Signed-off-by: YANGDB <[email protected]> * add live only workflow & dashboard add pre-agg + live workflow & dashboard add refresh-workflow for the pre-agg queries Signed-off-by: YANGDB <[email protected]> * add live all only workflow & dashboard Signed-off-by: YANGDB <[email protected]> * correct table name hard coded issue Signed-off-by: YANGDB <[email protected]> * add vpc table creation based on CSV format Signed-off-by: YANGDB <[email protected]> --------- Signed-off-by: YANGDB <[email protected]> Co-authored-by: Simeon Widdis <[email protected]> * Change the vpc flow's refresh interval to 15mins and watermark delay to 1 min Signed-off-by: Ryan Liang <[email protected]> --------- Signed-off-by: YANGDB <[email protected]> Signed-off-by: Ryan Liang <[email protected]> Co-authored-by: YANGDB <[email protected]> Co-authored-by: Simeon Widdis <[email protected]>
1 parent f1e2ce1 commit e97df5f

18 files changed

+1108
-35
lines changed

‎server/adaptors/integrations/__data__/repository/aws_vpc_flow/assets/aws_vpc_flow_flint-live-1.0.0.ndjson

Lines changed: 18 additions & 0 deletions
Large diffs are not rendered by default.

‎server/adaptors/integrations/__data__/repository/aws_vpc_flow/assets/aws_vpc_flow_flint-pre_agg-1.0.0.ndjson

Lines changed: 19 additions & 0 deletions
Large diffs are not rendered by default.

‎server/adaptors/integrations/__data__/repository/aws_vpc_flow/assets/create_mv_vpc-1.0.0.sql

Lines changed: 0 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -1,30 +0,0 @@
1-
CREATE MATERIALIZED VIEW {table_name}__mview AS
2-
SELECT
3-
CAST(FROM_UNIXTIME(start) AS TIMESTAMP) as `@timestamp`,
4-
version as `aws.vpc.version`,
5-
account_id as `aws.vpc.account-id`,
6-
interface_id as `aws.vpc.interface-id`,
7-
srcaddr as `aws.vpc.srcaddr`,
8-
dstaddr as `aws.vpc.dstaddr`,
9-
CAST(srcport AS LONG) as `aws.vpc.srcport`,
10-
CAST(dstport AS LONG) as `aws.vpc.dstport`,
11-
protocol as `aws.vpc.protocol`,
12-
CAST(packets AS LONG) as `aws.vpc.packets`,
13-
CAST(bytes AS LONG) as `aws.vpc.bytes`,
14-
CAST(FROM_UNIXTIME(start) AS TIMESTAMP) as `aws.vpc.start`,
15-
CAST(FROM_UNIXTIME(end) AS TIMESTAMP) as `aws.vpc.end`,
16-
action as `aws.vpc.action`,
17-
log_status as `aws.vpc.log-status`,
18-
CASE
19-
WHEN regexp(dstaddr, '(10\\..*)|(192\\.168\\..*)|(172\\.1[6-9]\\..*)|(172\\.2[0-9]\\..*)|(172\\.3[0-1]\\.*)')
20-
THEN 'ingress'
21-
ELSE 'egress'
22-
END AS `aws.vpc.flow-direction`
23-
FROM
24-
{table_name}
25-
WITH (
26-
auto_refresh = 'true',
27-
checkpoint_location = '{s3_checkpoint_location}',
28-
watermark_delay = '1 Minute',
29-
extra_options = '{ "{table_name}": { "maxFilesPerTrigger": "10" }}'
30-
)

‎server/adaptors/integrations/__data__/repository/aws_vpc_flow/assets/create_table_parquet_vpc-1.0.0.sql

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
CREATE EXTERNAL TABLE IF NOT EXISTS {table_name} (
2+
cloud STRUCT<
3+
account_uid: STRING,
4+
region: STRING,
5+
zone: STRING,
6+
provider: STRING
7+
>,
8+
src_endpoint STRUCT<
9+
port: INT,
10+
svc_name: STRING,
11+
ip: STRING,
12+
intermediate_ips: ARRAY<STRING>,
13+
interface_uid: STRING,
14+
vpc_uid: STRING,
15+
instance_uid: STRING,
16+
subnet_uid: STRING
17+
>,
18+
dst_endpoint STRUCT<
19+
port: INT,
20+
svc_name: STRING,
21+
ip: STRING,
22+
intermediate_ips: ARRAY<STRING>,
23+
interface_uid: STRING,
24+
vpc_uid: STRING,
25+
instance_uid: STRING,
26+
subnet_uid: STRING
27+
>,
28+
connection_info STRUCT<
29+
protocol_num: INT,
30+
tcp_flags: INT,
31+
protocol_ver: STRING,
32+
boundary_id: INT,
33+
boundary: STRING,
34+
direction_id: INT,
35+
direction: STRING
36+
>,
37+
traffic STRUCT<
38+
packets: BIGINT,
39+
bytes: BIGINT
40+
>,
41+
time BIGINT,
42+
start_time BIGINT,
43+
end_time BIGINT,
44+
status_code STRING,
45+
severity_id INT,
46+
severity STRING,
47+
class_name STRING,
48+
class_uid INT,
49+
category_name STRING,
50+
category_uid INT,
51+
activity_name STRING,
52+
activity_id INT,
53+
disposition STRING,
54+
disposition_id INT,
55+
type_uid INT,
56+
type_name STRING,
57+
region STRING,
58+
accountid STRING,
59+
eventday STRING
60+
)
61+
USING json
62+
LOCATION '{s3_bucket_location}'

‎server/adaptors/integrations/__data__/repository/aws_vpc_flow/assets/vpc_agg_30min_connections_mv-1.0.0.sql

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
CREATE MATERIALIZED VIEW IF NOT EXISTS {table_name}__agg_30_min_connections_mview AS
2+
SELECT
3+
CAST(from_unixtime(CAST((start_time / 1000) AS BIGINT) DIV 1800 * 1800) AS TIMESTAMP) AS interval_start_time,
4+
CAST(from_unixtime((CAST((start_time / 1000) AS BIGINT) DIV 1800 * 1800) + 1799) AS TIMESTAMP) AS interval_end_time,
5+
6+
status_code as `aws.vpc.status_code`,
7+
CAST(IFNULL(connection_info['direction'], 'Unknown') AS STRING) AS `aws.vpc.connection.direction`,
8+
CAST(IFNULL(src_endpoint.svc_name, 'Unknown') AS STRING) AS `aws.vpc.pkt-src-aws-service`,
9+
CAST(IFNULL(dst_endpoint.svc_name, 'Unknown') AS STRING) AS `aws.vpc.pkt-dst-aws-service`,
10+
11+
accountid as `aws.vpc.account-id`,
12+
region as `aws.vpc.region`,
13+
14+
COUNT(*) AS total_connections,
15+
SUM(CAST(IFNULL(traffic.bytes, 0) AS LONG)) AS total_bytes,
16+
SUM(CAST(IFNULL(traffic.packets, 0) AS LONG)) AS total_packets
17+
FROM
18+
{table_name}
19+
GROUP BY
20+
CAST((start_time / 1000) AS BIGINT) DIV 1800 * 1800,
21+
region,
22+
accountid,
23+
status_code,
24+
src_endpoint.svc_name,
25+
dst_endpoint.svc_name,
26+
connection_info['direction']
27+
ORDER BY
28+
interval_start_time
29+
WITH (
30+
auto_refresh = false
31+
)
32+

‎server/adaptors/integrations/__data__/repository/aws_vpc_flow/assets/vpc_agg_30min_connections_refresh-1.0.0.sql

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
REFRESH MATERIALIZED VIEW {table_name}__agg_30_min_connections_mview

‎server/adaptors/integrations/__data__/repository/aws_vpc_flow/assets/vpc_agg_60min_connections_mv-1.0.0.sql

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
CREATE MATERIALIZED VIEW IF NOT EXISTS {table_name}__agg_60_min_connections_mview AS
2+
SELECT
3+
date_trunc('hour', from_unixtime(start_time / 1000)) AS interval_start_time,
4+
date_trunc('hour', from_unixtime(start_time / 1000)) + INTERVAL 1 HOUR AS interval_end_time,
5+
6+
status_code as `aws.vpc.status_code`,
7+
CAST(IFNULL(connection_info['direction'], 'Unknown') AS STRING) AS `aws.vpc.connection.direction`,
8+
CAST(IFNULL(src_endpoint.svc_name, 'Unknown') AS STRING) AS `aws.vpc.pkt-src-aws-service`,
9+
CAST(IFNULL(dst_endpoint.svc_name, 'Unknown') AS STRING) AS `aws.vpc.pkt-dst-aws-service`,
10+
11+
accountid as `aws.vpc.account-id`,
12+
region as `aws.vpc.region`,
13+
14+
COUNT(*) AS total_connections,
15+
SUM(CAST(IFNULL(traffic.bytes, 0) AS LONG)) AS total_bytes,
16+
SUM(CAST(IFNULL(traffic.packets, 0) AS LONG)) AS total_packets
17+
FROM
18+
{table_name}
19+
GROUP BY
20+
date_trunc('hour', from_unixtime(start_time / 1000)),
21+
region,
22+
accountid,
23+
status_code,
24+
src_endpoint.svc_name,
25+
dst_endpoint.svc_name,
26+
connection_info['direction']
27+
ORDER BY
28+
interval_start_time
29+
WITH (
30+
auto_refresh = false
31+
)
32+

‎server/adaptors/integrations/__data__/repository/aws_vpc_flow/assets/vpc_agg_60min_connections_refresh-1.0.0.sql

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
REFRESH MATERIALIZED VIEW {table_name}__agg_60_min_connections_mview

‎server/adaptors/integrations/__data__/repository/aws_vpc_flow/assets/vpc_live_all_mv-1.0.0.sql

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
CREATE MATERIALIZED VIEW {table_name}__week_live_mview AS
2+
SELECT
3+
cloud.account_uid AS `aws.vpc.cloud_account_uid`,
4+
cloud.region AS `aws.vpc.cloud_region`,
5+
cloud.zone AS `aws.vpc.cloud_zone`,
6+
cloud.provider AS `aws.vpc.cloud_provider`,
7+
8+
CAST(IFNULL(src_endpoint.port, 0) AS LONG) AS `aws.vpc.srcport`,
9+
CAST(IFNULL(src_endpoint.svc_name, 'Unknown') AS STRING) AS `aws.vpc.pkt-src-aws-service`,
10+
CAST(IFNULL(src_endpoint.ip, '0.0.0.0') AS STRING) AS `aws.vpc.srcaddr`,
11+
CAST(IFNULL(src_endpoint.interface_uid, 'Unknown') AS STRING) AS `aws.vpc.src-interface_uid`,
12+
CAST(IFNULL(src_endpoint.vpc_uid, 'Unknown') AS STRING) AS `aws.vpc.src-vpc_uid`,
13+
CAST(IFNULL(src_endpoint.instance_uid, 'Unknown') AS STRING) AS `aws.vpc.src-instance_uid`,
14+
CAST(IFNULL(src_endpoint.subnet_uid, 'Unknown') AS STRING) AS `aws.vpc.src-subnet_uid`,
15+
16+
CAST(IFNULL(dst_endpoint.port, 0) AS LONG) AS `aws.vpc.dstport`,
17+
CAST(IFNULL(dst_endpoint.svc_name, 'Unknown') AS STRING) AS `aws.vpc.pkt-dst-aws-service`,
18+
CAST(IFNULL(dst_endpoint.ip, '0.0.0.0') AS STRING) AS `aws.vpc.dstaddr`,
19+
CAST(IFNULL(dst_endpoint.interface_uid, 'Unknown') AS STRING) AS `aws.vpc.dst-interface_uid`,
20+
CAST(IFNULL(dst_endpoint.vpc_uid, 'Unknown') AS STRING) AS `aws.vpc.dst-vpc_uid`,
21+
CAST(IFNULL(dst_endpoint.instance_uid, 'Unknown') AS STRING) AS `aws.vpc.dst-instance_uid`,
22+
CAST(IFNULL(dst_endpoint.subnet_uid, 'Unknown') AS STRING) AS `aws.vpc.dst-subnet_uid`,
23+
CASE
24+
WHEN regexp(dst_endpoint.ip, '(10\\..*)|(192\\.168\\..*)|(172\\.1[6-9]\\..*)|(172\\.2[0-9]\\..*)|(172\\.3[0-1]\\.*)')
25+
THEN 'ingress'
26+
ELSE 'egress'
27+
END AS `aws.vpc.flow-direction`,
28+
29+
CAST(IFNULL(connection_info['protocol_num'], 0) AS INT) AS `aws.vpc.connection.protocol_num`,
30+
CAST(IFNULL(connection_info['tcp_flags'], '0') AS STRING) AS `aws.vpc.connection.tcp_flags`,
31+
CAST(IFNULL(connection_info['protocol_ver'], '0') AS STRING) AS `aws.vpc.connection.protocol_ver`,
32+
CAST(IFNULL(connection_info['boundary'], 'Unknown') AS STRING) AS `aws.vpc.connection.boundary`,
33+
CAST(IFNULL(connection_info['direction'], 'Unknown') AS STRING) AS `aws.vpc.connection.direction`,
34+
35+
CAST(IFNULL(traffic.packets, 0) AS LONG) AS `aws.vpc.packets`,
36+
CAST(IFNULL(traffic.bytes, 0) AS LONG) AS `aws.vpc.bytes`,
37+
38+
CAST(FROM_UNIXTIME(time / 1000) AS TIMESTAMP) AS `@timestamp`,
39+
CAST(FROM_UNIXTIME(start_time / 1000) AS TIMESTAMP) AS `start_time`,
40+
CAST(FROM_UNIXTIME(start_time / 1000) AS TIMESTAMP) AS `interval_start_time`,
41+
CAST(FROM_UNIXTIME(end_time / 1000) AS TIMESTAMP) AS `end_time`,
42+
status_code AS `aws.vpc.status_code`,
43+
44+
severity AS `aws.vpc.severity`,
45+
class_name AS `aws.vpc.class_name`,
46+
category_name AS `aws.vpc.category_name`,
47+
activity_name AS `aws.vpc.activity_name`,
48+
disposition AS `aws.vpc.disposition`,
49+
type_name AS `aws.vpc.type_name`,
50+
51+
region AS `aws.vpc.region`,
52+
accountid AS `aws.vpc.account-id`
53+
FROM
54+
{table_name}
55+
WITH (
56+
auto_refresh = true,
57+
refresh_interval = '15 Minute',
58+
checkpoint_location = '{s3_checkpoint_location}',
59+
watermark_delay = '1 Minute',
60+
extra_options = '{ "{table_name}": { "maxFilesPerTrigger": "10" }}'
61+
)
62+

‎server/adaptors/integrations/__data__/repository/aws_vpc_flow/assets/vpc_live_week_mv-1.0.0.sql

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
CREATE MATERIALIZED VIEW {table_name}__week_live_mview AS
2+
SELECT
3+
cloud.account_uid AS `aws.vpc.cloud_account_uid`,
4+
cloud.region AS `aws.vpc.cloud_region`,
5+
cloud.zone AS `aws.vpc.cloud_zone`,
6+
cloud.provider AS `aws.vpc.cloud_provider`,
7+
8+
CAST(IFNULL(src_endpoint.port, 0) AS LONG) AS `aws.vpc.srcport`,
9+
CAST(IFNULL(src_endpoint.svc_name, 'Unknown') AS STRING) AS `aws.vpc.pkt-src-aws-service`,
10+
CAST(IFNULL(src_endpoint.ip, '0.0.0.0') AS STRING) AS `aws.vpc.srcaddr`,
11+
CAST(IFNULL(src_endpoint.interface_uid, 'Unknown') AS STRING) AS `aws.vpc.src-interface_uid`,
12+
CAST(IFNULL(src_endpoint.vpc_uid, 'Unknown') AS STRING) AS `aws.vpc.src-vpc_uid`,
13+
CAST(IFNULL(src_endpoint.instance_uid, 'Unknown') AS STRING) AS `aws.vpc.src-instance_uid`,
14+
CAST(IFNULL(src_endpoint.subnet_uid, 'Unknown') AS STRING) AS `aws.vpc.src-subnet_uid`,
15+
16+
CAST(IFNULL(dst_endpoint.port, 0) AS LONG) AS `aws.vpc.dstport`,
17+
CAST(IFNULL(dst_endpoint.svc_name, 'Unknown') AS STRING) AS `aws.vpc.pkt-dst-aws-service`,
18+
CAST(IFNULL(dst_endpoint.ip, '0.0.0.0') AS STRING) AS `aws.vpc.dstaddr`,
19+
CAST(IFNULL(dst_endpoint.interface_uid, 'Unknown') AS STRING) AS `aws.vpc.dst-interface_uid`,
20+
CAST(IFNULL(dst_endpoint.vpc_uid, 'Unknown') AS STRING) AS `aws.vpc.dst-vpc_uid`,
21+
CAST(IFNULL(dst_endpoint.instance_uid, 'Unknown') AS STRING) AS `aws.vpc.dst-instance_uid`,
22+
CAST(IFNULL(dst_endpoint.subnet_uid, 'Unknown') AS STRING) AS `aws.vpc.dst-subnet_uid`,
23+
CASE
24+
WHEN regexp(dst_endpoint.ip, '(10\\..*)|(192\\.168\\..*)|(172\\.1[6-9]\\..*)|(172\\.2[0-9]\\..*)|(172\\.3[0-1]\\.*)')
25+
THEN 'ingress'
26+
ELSE 'egress'
27+
END AS `aws.vpc.flow-direction`,
28+
29+
CAST(IFNULL(connection_info['protocol_num'], 0) AS INT) AS `aws.vpc.connection.protocol_num`,
30+
CAST(IFNULL(connection_info['tcp_flags'], '0') AS STRING) AS `aws.vpc.connection.tcp_flags`,
31+
CAST(IFNULL(connection_info['protocol_ver'], '0') AS STRING) AS `aws.vpc.connection.protocol_ver`,
32+
CAST(IFNULL(connection_info['boundary'], 'Unknown') AS STRING) AS `aws.vpc.connection.boundary`,
33+
CAST(IFNULL(connection_info['direction'], 'Unknown') AS STRING) AS `aws.vpc.connection.direction`,
34+
35+
CAST(IFNULL(traffic.packets, 0) AS LONG) AS `aws.vpc.packets`,
36+
CAST(IFNULL(traffic.bytes, 0) AS LONG) AS `aws.vpc.bytes`,
37+
38+
CAST(FROM_UNIXTIME(time / 1000) AS TIMESTAMP) AS `@timestamp`,
39+
CAST(FROM_UNIXTIME(start_time / 1000) AS TIMESTAMP) AS `start_time`,
40+
CAST(FROM_UNIXTIME(start_time / 1000) AS TIMESTAMP) AS `interval_start_time`,
41+
CAST(FROM_UNIXTIME(end_time / 1000) AS TIMESTAMP) AS `end_time`,
42+
status_code AS `aws.vpc.status_code`,
43+
44+
severity AS `aws.vpc.severity`,
45+
class_name AS `aws.vpc.class_name`,
46+
category_name AS `aws.vpc.category_name`,
47+
activity_name AS `aws.vpc.activity_name`,
48+
disposition AS `aws.vpc.disposition`,
49+
type_name AS `aws.vpc.type_name`,
50+
51+
region AS `aws.vpc.region`,
52+
accountid AS `aws.vpc.account-id`
53+
FROM
54+
{table_name},
55+
(SELECT MAX(CAST(FROM_UNIXTIME(start_time / 1000) AS TIMESTAMP)) AS max_start_time FROM {table_name}) AS latest
56+
WHERE
57+
CAST(FROM_UNIXTIME(start_time / 1000) AS TIMESTAMP) >= DATE_SUB(latest.max_start_time, 7)
58+
WITH (
59+
auto_refresh = false
60+
)
61+

‎server/adaptors/integrations/__data__/repository/aws_vpc_flow/assets/vpc_live_week_refresh-1.0.0.sql

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
REFRESH MATERIALIZED VIEW {table_name}__week_live_mview

‎server/adaptors/integrations/__data__/repository/aws_vpc_flow/assets/vpc_window-agg_60min_dest_ip_cardinality_mv-1.0.0.sql

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
CREATE MATERIALIZED VIEW IF NOT EXISTS {table_name}__window_agg_60_min_network_ip_bytes_mview AS
2+
WITH hourly_buckets AS (
3+
SELECT
4+
date_trunc('hour', from_unixtime(start_time / 1000)) AS interval_start_time,
5+
CAST(IFNULL(dst_endpoint.ip, '0.0.0.0') AS STRING) AS dstaddr,
6+
SUM(CAST(IFNULL(traffic.bytes, 0) AS LONG)) AS total_bytes
7+
FROM
8+
{table_name}
9+
GROUP BY
10+
interval_start_time,
11+
dstaddr
12+
),
13+
ranked_addresses AS (
14+
SELECT
15+
CAST(interval_start_time AS TIMESTAMP),
16+
dstaddr,
17+
total_bytes,
18+
RANK() OVER (PARTITION BY interval_start_time ORDER BY total_bytes DESC) AS bytes_rank
19+
FROM
20+
hourly_buckets
21+
)
22+
SELECT
23+
CAST(interval_start_time AS TIMESTAMP),
24+
dstaddr,
25+
total_bytes
26+
FROM
27+
ranked_addresses
28+
WHERE
29+
bytes_rank <= 50
30+
ORDER BY
31+
interval_start_time ASC,
32+
bytes_rank ASC
33+
WITH (
34+
auto_refresh = false
35+
)
36+

‎server/adaptors/integrations/__data__/repository/aws_vpc_flow/assets/vpc_window-agg_60min_dest_ip_cardinality_refresh-1.0.0.sql

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
REFRESH MATERIALIZED VIEW {table_name}__window_agg_60_min_network_ip_bytes_mview

‎server/adaptors/integrations/__data__/repository/aws_vpc_flow/assets/vpc_window-agg_60min_dest_ip_total-bytes_mv-1.0.0.sql

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
CREATE MATERIALIZED VIEW IF NOT EXISTS {table_name}__window_agg_60_min_network_ip_cardinality_mview AS
2+
WITH hourly_buckets AS (
3+
SELECT
4+
date_trunc('hour', from_unixtime(start_time / 1000)) AS interval_start_time,
5+
CAST(IFNULL(dst_endpoint.ip, '0.0.0.0') AS STRING) AS dstaddr,
6+
COUNT(*) AS total_count
7+
FROM
8+
{table_name}
9+
GROUP BY
10+
interval_start_time,
11+
dstaddr
12+
),
13+
ranked_addresses AS (
14+
SELECT
15+
CAST(interval_start_time AS TIMESTAMP),
16+
dstaddr,
17+
total_count,
18+
RANK() OVER (PARTITION BY interval_start_time ORDER BY total_count DESC) AS addr_rank
19+
FROM
20+
hourly_buckets
21+
)
22+
SELECT
23+
CAST(interval_start_time AS TIMESTAMP),
24+
dstaddr,
25+
total_count
26+
FROM
27+
ranked_addresses
28+
WHERE
29+
addr_rank <= 50
30+
ORDER BY
31+
interval_start_time ASC,
32+
addr_rank ASC
33+
WITH (
34+
auto_refresh = false
35+
)
36+

‎server/adaptors/integrations/__data__/repository/aws_vpc_flow/assets/vpc_window-agg_60min_dest_ip_total-bytes_refresh-1.0.0.sql

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
REFRESH MATERIALIZED VIEW {table_name}__window_agg_60_min_network_ip_cardinality_mview

0 commit comments

Comments
 (0)