This article describes how to implement a data warehouse by subscribing to business events from Quavo and using a combination of AWS tools and Snowflake to consume those events. Please not note that this is only one approach and there may be a different approach that is more appropriate for your organization.
...
It is recommended that an Authentication Profile is used, however basic API Key Authentication is supported.
Events
Quavo will configure QFD so it knows which events should be tracked and published.
Event Stream
QFD will automatically process event queues and send events to the defined endpoints based on the defined settings.
...
This creates Business Event instances.
Requeuing Business Events
This utility provides method to requeue events by TranmissionStatus, Start/End DateTime window, or specific EventId:
...
Snowflake ingests each S3 file as it is notified via the SQS queue. For each event type a stage, snowpipe, staging table, structured final table, merge task, and staging table cleanup task should be set up. A storage integration (quavo_snowflake_qfd_{dev|stg|prod}_s3) and file format (QFD.{DEV|STG|PROD}.JSON_FILE_FORMAT are also created but these are already existing and reused for all event types and will not need to be recreated.
To start you need to be in the "DATAARCHITECT" role for the schema you are working in. For this example we are using QFDDEVDATAARCHITECT. You should be developing and testing initially in the DEV schema before making the same changes in the STG and PROD schemas.
Snowflake Stage
Code Block | ||
---|---|---|
| ||
create or replace stage QFD.DEV.ACCOUNTING storage_integration = "quavo_snowflake_qfd_dev_s3" url = 's3://quavo-snowflake-qfd-dev/load/Accounting/' file_format = QFD.DEV.JSON_FILE_FORMAT; |
Staging Table (Unstructured)
This is an unstructured database table that contains a single VARIANT column containing the raw JSON records.
Code Block | ||
---|---|---|
| ||
create or replace TABLE QFD.DEV.ACCOUNTING_STG (
EVENTPAYLOAD VARIANT
); |
Snowpipe
Code Block | ||
---|---|---|
| ||
create or replace pipe QFD.DEV.ACCOUNTING auto_ingest=true as copy into qfd.dev.accounting_stg
from @qfd.dev.accounting
file_format = QFD.DEV.JSON_FILE_FORMAT
MATCH_BY_COLUMN_NAME = NONE; |
Staging Table (Unstructured)
This is an unstructured database table that contains a single VARIANT column containing the raw JSON records.
Code Block | ||
---|---|---|
| ||
create or replace TABLE QFD.DEV.ACCOUNTING_STG (
EVENTPAYLOAD VARIANT
); |
Stream on Staging Table
The stream monitors for appended rows in the staging table. This is used in the merge task.
Code Block | ||
---|---|---|
| ||
create or replace stream QFD.DEV.ACCOUNTING_STREAM_APPEND_ONLY on table ACCOUNTING_STG append_only = true; |
...
Code Block | ||
---|---|---|
| ||
create or replace TABLE QFD.DEV.ACCOUNTING (
TENANTID VARCHAR(32),
CLIENTID VARCHAR(64),
ENTRYIDENTIFIER VARCHAR(64),
CLAIMID VARCHAR(32),
DISPUTEID VARCHAR(32),
PERFORMEDON TIMESTAMP_NTZ(9),
PERFORMEDBYOPERATORID VARCHAR(128),
EXECUTEDON TIMESTAMP_NTZ(9),
EXECUTEDBYOPERATORID VARCHAR(128),
COLLECTIONNAME VARCHAR(64),
DEBITCREDIT VARCHAR(16),
AMOUNT NUMBER(9,2),
REASON VARCHAR(64),
EXECUTIONMETHOD VARCHAR(32),
STEPIDENTIFIER VARCHAR(64),
EVENTDATETIME TIMESTAMP_NTZ(9),
EVENTIDENTIFIER VARCHAR(64),
EVENTITEMID VARCHAR(64),
primary key (EVENTITEMID, ENTRYIDENTIFIER)
);
GRANT SELECT ON TABLE QFD.DEV.ACCOUNTING to ROLE QFDDEVREADONLY; |
Merge Task
This task will merge the unstructured EVENTPAYLOAD data from the staging table into the final structured table.
...