Create materialized view on external table #1407

dai-chen · 2023-03-07T23:44:13Z

Is your feature request related to a problem?

In #1379, we've verified query acceleration capability and maintainance of secondary index. This issue will focus on materialized view implementation.

What solution would you like?

Will work on the following task for prototype, demo and development afterwards:

Extend Spark grammar with CREATE MATERIALIZED VIEW
Refresh MV data by Spark streaming job
Add tumbling window function

What alternatives have you considered?

Alternative solution and reference for this work:

Apache Carbondata: https://carbondata.apache.org/index.html

Do you have any additional context?

Please find more details in:

acarbonetto · 2023-03-08T20:42:42Z

Do we have a standard for creating "internal" or "plugin" indices?

Store materialized view data in OpenSearch index

dai-chen · 2023-03-13T21:05:18Z

Do we have a standard for creating "internal" or "plugin" indices?

Store materialized view data in OpenSearch index

For internal indices, there is hidden or system index in OpenSearch. In our case, MV data maybe regular index.

dai-chen · 2023-03-13T22:04:47Z

Proof of Concepts

Setup

Please refer to #1379 (comment)

Cleanup

#Drop tables
DROP TABLE maximus_alb_logs;
DROP TABLE alb_logs_metrics;

#Clean MV and index previous created
rm -rf spark-warehouse/alb_logs_metrics
rm -rf spark-warehouse/indexes/*

#Remove _delta folder on S3

#This checkpoint folder location is hardcoding in code for now
rm -rf /tmp/delta/_checkpoints/*

Test with ALB Logs

General Case: MV with Aggregation for Log-to-Metric Transformation

#Create temp table for simulating ingestion
CREATE TABLE IF NOT EXISTS alb_logs_temp
(
  type string,
  time timestamp,
  elb string,
  client_ip string,
  client_port int,
  target_ip string,
  target_port int,
  request_processing_time double,
  target_processing_time double,
  response_processing_time double,
  elb_status_code int,
  target_status_code string,
  received_bytes bigint,
  sent_bytes bigint,
  request_verb string,
  request_url string,
  request_proto string,
  user_agent string,
  ssl_cipher string,
  ssl_protocol string,
  target_group_arn string,
  trace_id string,
  domain_name string,
  chosen_cert_arn string,
  matched_rule_priority string,
  request_creation_time string,
  actions_executed string,
  redirect_url string,
  lambda_error_reason string,
  target_port_list string,
  target_status_code_list string,
  classification string,
  classification_reason string
)
USING PARQUET
LOCATION "s3a://maximus-alb-logs/";

#Create Maimxus table with auto refresh enabled
CREATE EXTERNAL TABLE IF NOT EXISTS maximus_alb_logs
(
  type string,
  time timestamp,
  elb string,
  client_ip string,
  client_port int,
  target_ip string,
  target_port int,
  request_processing_time double,
  target_processing_time double,
  response_processing_time double,
  elb_status_code int,
  target_status_code string,
  received_bytes bigint,
  sent_bytes bigint,
  request_verb string,
  request_url string,
  request_proto string,
  user_agent string,
  ssl_cipher string,
  ssl_protocol string,
  target_group_arn string,
  trace_id string,
  domain_name string,
  chosen_cert_arn string,
  matched_rule_priority string,
  request_creation_time string,
  actions_executed string,
  redirect_url string,
  lambda_error_reason string,
  target_port_list string,
  target_status_code_list string,
  classification string,
  classification_reason string
)
USING DELTA
LOCATION "s3a://maximus-alb-logs/"
TBLPROPERTIES ('auto_refresh'='true');

#Create MV for total count with 1 minute window
CREATE MATERIALIZED VIEW alb_logs_metrics
AS
SELECT
  window.start AS startTime,
  window.end AS endTime,
  COUNT(*) AS totalCount
FROM maximus_alb_logs
GROUP BY TUMBLE(time, '1 Minutes');

#Add 2 same records at 16:30:00 by running twice
INSERT INTO alb_logs_temp
VALUES
(
  "https", --type
  CAST("2023-03-13 16:30:00.000000" AS TIMESTAMP), --time
  "app/elb1",      --elb
  "10.212.10.100", --client_ip
  41950,           --client_port
  "10.212.20.1",   --target_ip
  443,   --target_port
  0.002, --request_processing_time
  0.046, --target_processing_time
  0.0,   --response_processing_time
  403,   --elb_status_code
  "403", --target_status_code
  211,   --received_bytes
  364,   --sent_bytes
  "GET", --request_verb
  "https://192.168.1.100:443/solr/", --request_url
  NULL,  --request_proto
  NULL,  --user_agent
  NULL,  --ssl_cipher
  NULL,  --ssl_protocol
  NULL,  --target_group_arn
  NULL,  --trace_id
  NULL,  --domain_name
  NULL,  --chosen_cert_arn
  NULL,  --matched_rule_priority
  NULL,  --request_creation_time
  NULL,  --actions_executed
  NULL,  --redirect_url
  NULL,  --lambda_error_reason
  NULL,  --target_port_list
  NULL,  --target_status_code_list
  NULL,  --classification
  NULL   --classification_reason
);

#No records in MV due to records not under watermark
SELECT * FROM alb_logs_metrics;
(None)

#Add a record at 16:32:00 which trigger the first 2 records fired
INSERT INTO alb_logs_temp
VALUES
(
  "https", --type
  CAST("2023-03-13 16:32:00.000000" AS TIMESTAMP), --time
  "app/elb1",      --elb
  "10.212.10.100", --client_ip
  41950,           --client_port
  "10.212.20.1",   --target_ip
  443,   --target_port
  0.002, --request_processing_time
  0.046, --target_processing_time
  0.0,   --response_processing_time
  403,   --elb_status_code
  "403", --target_status_code
  211,   --received_bytes
  364,   --sent_bytes
  "GET", --request_verb
  "https://192.168.1.100:443/solr/", --request_url
  NULL,  --request_proto
  NULL,  --user_agent
  NULL,  --ssl_cipher
  NULL,  --ssl_protocol
  NULL,  --target_group_arn
  NULL,  --trace_id
  NULL,  --domain_name
  NULL,  --chosen_cert_arn
  NULL,  --matched_rule_priority
  NULL,  --request_creation_time
  NULL,  --actions_executed
  NULL,  --redirect_url
  NULL,  --lambda_error_reason
  NULL,  --target_port_list
  NULL,  --target_status_code_list
  NULL,  --classification
  NULL   --classification_reason
);

SELECT * FROM alb_logs_metrics;
2023-03-13 16:30:00    2023-03-13 16:31:00    2

#Add another one at 16:35:00
INSERT INTO alb_logs_temp
VALUES
(
  "https", --type
  CAST("2023-03-13 16:35:00.000000" AS TIMESTAMP), --time
  "app/elb1",      --elb
  "10.212.10.100", --client_ip
  41950,           --client_port
  "10.212.20.1",   --target_ip
  443,   --target_port
  0.002, --request_processing_time
  0.046, --target_processing_time
  0.0,   --response_processing_time
  403,   --elb_status_code
  "403", --target_status_code
  211,   --received_bytes
  364,   --sent_bytes
  "GET", --request_verb
  "https://192.168.1.100:443/solr/", --request_url
  NULL,  --request_proto
  NULL,  --user_agent
  NULL,  --ssl_cipher
  NULL,  --ssl_protocol
  NULL,  --target_group_arn
  NULL,  --trace_id
  NULL,  --domain_name
  NULL,  --chosen_cert_arn
  NULL,  --matched_rule_priority
  NULL,  --request_creation_time
  NULL,  --actions_executed
  NULL,  --redirect_url
  NULL,  --lambda_error_reason
  NULL,  --target_port_list
  NULL,  --target_status_code_list
  NULL,  --classification
  NULL   --classification_reason
);

SELECT * FROM alb_logs_metrics;
2023-03-13 16:30:00    2023-03-13 16:31:00    2
2023-03-13 16:32:00    2023-03-13 16:33:00    1

Special Case: MV as Covering Index (Streaming Raw Data into OpenSearch Index)

#Create MV to simply load all data
CREATE MATERIALIZED VIEW alb_logs_raw
AS
SELECT * FROM maximus_alb_logs;

#Add records
INSERT INTO alb_logs_temp
VALUES
(
  "https", --type
  CAST("2023-03-13 16:30:00.000000" AS TIMESTAMP), --time
  "app/elb1",      --elb
  "10.212.10.100", --client_ip
  41950,           --client_port
  "10.212.20.1",   --target_ip
  443,   --target_port
  0.002, --request_processing_time
  0.046, --target_processing_time
  0.0,   --response_processing_time
  403,   --elb_status_code
  "403", --target_status_code
  211,   --received_bytes
  364,   --sent_bytes
  "GET", --request_verb
  "https://192.168.1.100:443/solr/", --request_url
  NULL,  --request_proto
  NULL,  --user_agent
  NULL,  --ssl_cipher
  NULL,  --ssl_protocol
  NULL,  --target_group_arn
  NULL,  --trace_id
  NULL,  --domain_name
  NULL,  --chosen_cert_arn
  NULL,  --matched_rule_priority
  NULL,  --request_creation_time
  NULL,  --actions_executed
  NULL,  --redirect_url
  NULL,  --lambda_error_reason
  NULL,  --target_port_list
  NULL,  --target_status_code_list
  NULL,  --classification
  NULL   --classification_reason
);

#Because this is not streaming aggregation, data is output to MV immediately
SELECT * FROM alb_logs_raw;
https    2023-03-13 16:30:00    app/elb1    10.212.10.100    41950    10.212.20.1    443    0.002    0.046    0.0    403    403    211    364    GET    https://192.168.1.100:443/solr/    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL
Time taken: 0.289 seconds, Fetched 1 row(s)

#Add another record
INSERT INTO alb_logs_temp
VALUES
(
  "https", --type
  CAST("2023-03-13 16:32:00.000000" AS TIMESTAMP), --time
  "app/elb1",      --elb
  "10.212.10.100", --client_ip
  41950,           --client_port
  "10.212.20.1",   --target_ip
  443,   --target_port
  0.002, --request_processing_time
  0.046, --target_processing_time
  0.0,   --response_processing_time
  403,   --elb_status_code
  "403", --target_status_code
  211,   --received_bytes
  364,   --sent_bytes
  "GET", --request_verb
  "https://192.168.1.100:443/solr/", --request_url
  NULL,  --request_proto
  NULL,  --user_agent
  NULL,  --ssl_cipher
  NULL,  --ssl_protocol
  NULL,  --target_group_arn
  NULL,  --trace_id
  NULL,  --domain_name
  NULL,  --chosen_cert_arn
  NULL,  --matched_rule_priority
  NULL,  --request_creation_time
  NULL,  --actions_executed
  NULL,  --redirect_url
  NULL,  --lambda_error_reason
  NULL,  --target_port_list
  NULL,  --target_status_code_list
  NULL,  --classification
  NULL   --classification_reason
);

SELECT * FROM alb_logs_raw;
https    2023-03-13 16:30:00    app/elb1    10.212.10.100    41950    10.212.20.1    443    0.002    0.046    0.0    403    403    211    364    GET    https://192.168.1.100:443/solr/    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL
https    2023-03-13 16:32:00    app/elb1    10.212.10.100    41950    10.212.20.1    443    0.002    0.046    0.0    403    403    211    364    GET    https://192.168.1.100:443/solr/    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL    NULL
Time taken: 0.216 seconds, Fetched 2 row(s)

dai-chen added feature spark integration labels Mar 7, 2023

dai-chen self-assigned this Mar 7, 2023

github-actions bot added the untriaged label Mar 7, 2023

dai-chen removed the untriaged label Mar 7, 2023

penghuo mentioned this issue Jul 11, 2023

[Feature] OpenSearch and Apache Spark Integration opensearch-project/opensearch-spark#3

Closed

dai-chen closed this as completed Mar 13, 2023

dai-chen mentioned this issue Mar 13, 2023

Query rewrite with secondary index and materialized view #1435

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Create materialized view on external table #1407

Create materialized view on external table #1407

dai-chen commented Mar 7, 2023 •

edited

Loading

acarbonetto commented Mar 8, 2023

dai-chen commented Mar 13, 2023

dai-chen commented Mar 13, 2023 •

edited

Loading

Create materialized view on external table #1407

Create materialized view on external table #1407

Comments

dai-chen commented Mar 7, 2023 • edited Loading

acarbonetto commented Mar 8, 2023

dai-chen commented Mar 13, 2023

dai-chen commented Mar 13, 2023 • edited Loading

Proof of Concepts

Setup

Cleanup

Test with ALB Logs

General Case: MV with Aggregation for Log-to-Metric Transformation

Special Case: MV as Covering Index (Streaming Raw Data into OpenSearch Index)

dai-chen commented Mar 7, 2023 •

edited

Loading

dai-chen commented Mar 13, 2023 •

edited

Loading