Performance & Usage Telemetry: Store Astrato activity in Snowflake + Setup Astrato Activity Workbook
How does it work
Section titled “How does it work”Astrato appends metadata to the end of each query, subsequently allowing this to be viewed in Snowflake. Snowflake already contains data about which user has accessed each table, warehouse etc.
In this article we will cover how to store this data in a new database, using only one block of code.
The data that Astrato appends is shown below.
-- {"objectId":"f4ed7820-9884-45ba-b078-86fc260d5511","sheetId":"3c2fa835-05ce-47f8-9489-3b99a76adc0b","workbookId":"d8ccb7b8-970e-477f-a093-6e862a431e21"}To supplement this, you can export the lookup associating workbooks, sheets and users, from the administration section https://app.astrato.io/administration/exports.
Scheduled metadata share guide
Section titled “Scheduled metadata share guide”Manual data import guide
Section titled “Manual data import guide”In the step below, code can be run to create these table, enabling an easy upload of your data.
Step 1: Setup & Schedule Data Pipeline
Section titled “Step 1: Setup & Schedule Data Pipeline”To setup a storage area, you need to create a database called ASTRATO_QUERY_LOGS and populate it on schedule. The code to do this is below. Altering the structure means that Astrato will not be able to provide a compatible analytics workbook. We recommend reviewing the scheduled task to ensure this is run at an appropriate time.
✏️ Edit the warehouse name in the code below since
MY_WHmay not exist
For the manual file upload approach: Copy & run the code below in Snowflake
Section titled “For the manual file upload approach: Copy & run the code below in Snowflake”call STORE_ASTRATO_QUERY_HISTORY();--📝 🅿🆁🅾🅲🅴🅳🆄🆁🅴 Create the stored procedureCREATE OR REPLACE PROCEDURE STORE_ASTRATO_QUERY_HISTORY() RETURNS STRING LANGUAGE SQL AS $$BEGINcreate SCHEMA IF NOT EXISTS ASTRATO_LOGS;create SCHEMA IF NOT EXISTS ASTRATO_METADATA;CREATE TABLE IF NOT EXISTS ASTRATO_METADATA.ASTRATO_METADATA_COLLECTIONS ( ID VARCHAR(36), COLLECTIONNAME VARCHAR(255), WORKBOOKNAME VARCHAR(255), EXPORTED_AT TIMESTAMP_NTZ(9) DEFAULT CURRENT_TIMESTAMP());CREATE TABLE IF NOT EXISTS ASTRATO_METADATA.ASTRATO_METADATA_GROUPS ( GROUPID VARCHAR(36), GROUPNAME VARCHAR(255), USERID VARCHAR(36), EXPORTED_AT TIMESTAMP_NTZ(9) DEFAULT CURRENT_TIMESTAMP());CREATE TABLE IF NOT EXISTS ASTRATO_METADATA.ASTRATO_METADATA_SEMANTIC_LAYERS ( ID VARCHAR(36), NAME VARCHAR(255), DATACONNECTION VARCHAR(255), DATACONNECTIONID VARCHAR(36), EXPORTED_AT TIMESTAMP_NTZ(9) DEFAULT CURRENT_TIMESTAMP());CREATE TABLE IF NOT EXISTS ASTRATO_METADATA.ASTRATO_METADATA_USERS ( ID VARCHAR(36), EMAIL VARCHAR(255), NAME VARCHAR(255), FIRSTNAME VARCHAR(255), LASTNAME VARCHAR(255), EXPORTED_AT TIMESTAMP_NTZ(9) DEFAULT CURRENT_TIMESTAMP());CREATE TABLE IF NOT EXISTS ASTRATO_METADATA.ASTRATO_METADATA_USERS_PER_COLLECTION ( COLLECTIONID VARCHAR(36), ID VARCHAR(36), NAME_OR_EMAIL VARCHAR(255), USERACCESSACL VARCHAR(255), EXPORTED_AT TIMESTAMP_NTZ(9) DEFAULT CURRENT_TIMESTAMP());CREATE TABLE IF NOT EXISTS ASTRATO_METADATA.ASTRATO_METADATA_USERS_PER_SEMANTIC_LAYER ( SEMANTICLAYERID VARCHAR(36), ID VARCHAR(36), NAME_OR_EMAIL VARCHAR(255), USERACCESSACL VARCHAR(255), EXPORTED_AT TIMESTAMP_NTZ(9) DEFAULT CURRENT_TIMESTAMP());CREATE TABLE IF NOT EXISTS ASTRATO_METADATA.ASTRATO_METADATA_WORKBOOKS ( ID VARCHAR(36), NAME VARCHAR(255), DESCRIPTION VARCHAR(255), SEMANTICLAYERID VARCHAR(36), EXPORTED_AT TIMESTAMP_NTZ(9) DEFAULT CURRENT_TIMESTAMP());CREATE TABLE IF NOT EXISTS ASTRATO_METADATA.ASTRATO_METADATA_WORKBOOK_OBJECTS ( OBJECTNAME VARCHAR(255), OBJECTID VARCHAR(36), SHEETID VARCHAR(36), EXPORTED_AT TIMESTAMP_NTZ(9) DEFAULT CURRENT_TIMESTAMP());CREATE TABLE IF NOT EXISTS ASTRATO_METADATA.ASTRATO_METADATA_WORKBOOK_SHEETS ( ID VARCHAR(36), SHEETNAME VARCHAR(255), WORKBOOKNAME VARCHAR(255), EXPORTED_AT TIMESTAMP_NTZ(9) DEFAULT CURRENT_TIMESTAMP());create SCHEMA IF NOT EXISTS ASTRATO_LOGS;create TABLE IF NOT EXISTS ASTRATO_QUERY_LOGS.ASTRATO_LOGS.QUERY_HISTORY ( QUERY_ID VARCHAR(16777216), QUERY_TAG VARCHAR(16777216), QUERY_TEXT VARCHAR(16777216), EXECUTION_TIMESTAMP TIMESTAMP_LTZ(6), USER_NAME VARCHAR(16777216), WAREHOUSE_NAME VARCHAR(16777216), DATABASE_NAME VARCHAR(16777216), SCHEMA_NAME VARCHAR(16777216), START_TIME TIMESTAMP_LTZ(6), END_TIME TIMESTAMP_LTZ(6), ERROR_MESSAGE VARCHAR(16777216), ROWS_PRODUCED NUMBER(38,0), ROWS_INSERTED NUMBER(38,0), PERCENTAGE_SCANNED_FROM_CACHE FLOAT, EXECUTION_TIME NUMBER(38,0), COMPILATION_TIME NUMBER(38,0), EXTERNAL_FUNCTION_TOTAL_INVOCATIONS NUMBER(38,0), BYTES_SPILLED_TO_LOCAL_STORAGE NUMBER(38,0), BYTES_SPILLED_TO_REMOTE_STORAGE NUMBER(38,0), BYTES_SENT_OVER_THE_NETWORK NUMBER(38,0), CREDITS_USED_CLOUD_SERVICES FLOAT, WAREHOUSE_SIZE VARCHAR(16777216), WAREHOUSE_TYPE VARCHAR(16777216), ROLE_NAME VARCHAR(16777216), ASTRATO_QUERY_META_DATA VARCHAR(16777216), ASTRATO_QUERY_FIELDS VARCHAR(20000), ASTRATO_USER_ROLE VARCHAR(1670), ASTRATO_WORKBOOK_STATE VARCHAR(1670), ASTRATO_QUERY_SOURCE_TYPE VARCHAR(1670), astrato_user_id VARCHAR(1670), astrato_object_id VARCHAR(1670), astrato_sheet_id VARCHAR(1670), astrato_workbook_id VARCHAR(1670), ASTRATO_reportName VARCHAR(200), ASTRATO_reportRunId VARCHAR(200), ASTRATO_reportRunType VARCHAR(200), ASTRATO_reportDeliveryMethod VARCHAR(200), BYTES_SCANNED NUMBER(38,0), QUEUED_PROVISIONING_TIME NUMBER(38,0), QUEUED_OVERLOAD_TIME NUMBER(38,0),cortex_model VARCHAR(20000), cortex_prompt VARCHAR(20000), cortex_model_parameters VARCHAR(20000), cortex_response_format VARCHAR(20000));create TABLE IF NOT EXISTS ASTRATO_QUERY_LOGS.ASTRATO_LOGS.QUERIES_PER_TABLE ( QUERY_ID VARCHAR(16777216), DATABASE_NAME VARCHAR(16777216), SCHEMA_NAME VARCHAR(16777216), TABLE_NAME VARCHAR(16777216), TOTAL_MATCHES NUMBER(18,0), MATCH_POSITION NUMBER(38,0), ESTIMATED_CREDITS_USED_PER_TABLE FLOAT);------------------------------------------------------------------ CREATE TABLE ENDS -- Insert new query history into the target table INSERT INTO ASTRATO_QUERY_LOGS.ASTRATO_LOGS.QUERY_HISTORY(SELECT QUERY_ID, QUERY_TAG, QUERY_TEXT, START_TIME AS EXECUTION_TIMESTAMP, USER_NAME, WAREHOUSE_NAME, DATABASE_NAME, SCHEMA_NAME, START_TIME, END_TIME, ERROR_MESSAGE, ROWS_PRODUCED, ROWS_INSERTED, percentage_scanned_from_cache, EXECUTION_TIME, COMPILATION_TIME, EXTERNAL_FUNCTION_TOTAL_INVOCATIONS, BYTES_SPILLED_TO_LOCAL_STORAGE, BYTES_SPILLED_TO_REMOTE_STORAGE, BYTES_SENT_OVER_THE_NETWORK, credits_used_cloud_services, warehouse_size, warehouse_type, role_name, split_part(query_text, '--', -1) as astrato_query_meta_data, TRY_PARSE_JSON(astrato_query_meta_data):"fields"::STRING AS ASTRATO_QUERY_FIELDS, TRY_PARSE_JSON(astrato_query_meta_data):"roleType"::STRING AS ASTRATO_USER_ROLE, replace(TRY_PARSE_JSON(astrato_query_meta_data):"workbookState"::STRING,'depublished','unpublished') AS ASTRATO_WORKBOOK_STATE, TRY_PARSE_JSON(astrato_query_meta_data):"type"::STRING AS ASTRATO_QUERY_SOURCE_TYPE, TRY_PARSE_JSON(astrato_query_meta_data):"userId"::STRING AS astrato_user_id, TRY_PARSE_JSON(astrato_query_meta_data):"objectId"::STRING AS astrato_object_id, TRY_PARSE_JSON(astrato_query_meta_data):"sheetId"::STRING AS astrato_sheet_id, TRY_PARSE_JSON(astrato_query_meta_data):"workbookId"::STRING AS astrato_workbook_id, TRY_PARSE_JSON(astrato_query_meta_data):"reportName"::STRING AS ASTRATO_reportName, -- Added 2025-03-07 TRY_PARSE_JSON(astrato_query_meta_data):"reportRunId"::STRING AS ASTRATO_reportRunId, -- Added 2025-03-07 TRY_PARSE_JSON(astrato_query_meta_data):"reportRunType"::STRING AS ASTRATO_reportRunType, -- Added 2025-03-07 TRY_PARSE_JSON(astrato_query_meta_data):"reportDeliveryMethod"::STRING AS ASTRATO_reportDeliveryMethod, -- Added 2025-03-07 BYTES_SCANNED, QUEUED_PROVISIONING_TIME, QUEUED_OVERLOAD_TIME, REGEXP_SUBSTR(IFF( QUERY_TEXT ILIKE '%AI_COMPLETE%', query_text, null), 'model\\s*=>\\s*''([^'']*)''', 1, 1, 'e', 1) AS cortex_model, REGEXP_SUBSTR(IFF( QUERY_TEXT ILIKE '%AI_COMPLETE%', query_text, null), 'prompt\\s*=>\\s*''([\\s\\S]*?)''\\s*,\\s*model_parameters', 1, 1, 'e', 1) AS cortex_prompt, TRY_PARSE_JSON( REGEXP_SUBSTR(IFF( QUERY_TEXT ILIKE '%AI_COMPLETE%', query_text, null), 'model_parameters\\s*=>\\s*PARSE_JSON\\(''([^'']*)''\\)', 1, 1, 'e', 1) ) AS cortex_model_parameters, TRY_PARSE_JSON( REGEXP_SUBSTR(IFF( QUERY_TEXT ILIKE '%AI_COMPLETE%', query_text, null), 'response_format\\s*=>\\s*([^\\n\\r\\)]*)', 1, 1, 'e', 1) ) AS cortex_response_formatFROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORYWHERE --QUERY_TAG ='Astrato' AND --⚠️UNCOMMENT to store ONLY Astrato activity QUERY_ID NOT IN ( SELECT QUERY_ID FROM ASTRATO_QUERY_LOGS.ASTRATO_LOGS.QUERY_HISTORY ) and date(START_TIME) >= '2024-06-24');INSERT INTO ASTRATO_QUERY_LOGS.ASTRATO_LOGS.QUERIES_PER_TABLE( QUERY_ID, DATABASE_NAME, schema_name, table_name, total_matches, match_position, estimated_credits_used_per_table)WITH ExtractedTables AS ( SELECT QUERY_ID, DATABASE_NAME, credits_used_cloud_services, query_text, REGEXP_SUBSTR(query_text, '(FROM|JOIN)\\s+([A-Za-z0-9_]+)\\.([A-Za-z0-9_]+)', 1, 1, 'i') AS match_part, 1 AS match_position FROM ASTRATO_QUERY_LOGS.ASTRATO_LOGS.QUERY_HISTORY UNION ALL SELECT QUERY_ID, DATABASE_NAME, credits_used_cloud_services, et.query_text, REGEXP_SUBSTR(et.query_text, '(FROM|JOIN)\\s+([A-Za-z0-9_]+)\\.([A-Za-z0-9_]+)', 1, match_position + 1, 'i') AS match_part, match_position + 1 FROM ExtractedTables et WHERE REGEXP_SUBSTR(et.query_text, '(FROM|JOIN)\\s+([A-Za-z0-9_]+)\\.([A-Za-z0-9_]+)', 1, et.match_position + 1, 'i') IS NOT NULL),MatchedTables AS ( SELECT QUERY_ID, DATABASE_NAME, credits_used_cloud_services, reverse(TRIM(REGEXP_SUBSTR(reverse(match_part),'([A-Za-z0-9_]+)\\.([A-Za-z0-9_]+)',1,1,'i',2))) AS schema_name, reverse(TRIM(REGEXP_SUBSTR(reverse(match_part),'([A-Za-z0-9_]+)\\.([A-Za-z0-9_]+)',1,1,'i',1))) AS table_name, match_position, query_text FROM ExtractedTables WHERE match_part IS NOT NULL)SELECT QUERY_ID, DATABASE_NAME, schema_name, table_name, COUNT(*) OVER (PARTITION BY QUERY_ID) AS total_matches, -- Count total matches per QUERY_ID match_position, (credits_used_cloud_services / COUNT(*) OVER (PARTITION BY QUERY_ID)) AS estimated_credits_used_per_tableFROM MatchedTables;------🆆🅾🆁🅺🅱🅾🅾🅺 🆂🅴🆂🆂🅸🅾🅽🆂create or replace table workbook_sessions as (WITH RankedQueries AS ( SELECT ASTRATO_USER_ID, ASTRATO_WORKBOOK_ID, START_TIME, END_TIME, ROW_NUMBER() OVER ( PARTITION BY ASTRATO_USER_ID, ASTRATO_WORKBOOK_ID ORDER BY START_TIME ) AS QUERY_RANK FROM ASTRATO_QUERY_LOGS.ASTRATO_LOGS.QUERY_HISTORY),SessionizedData AS ( SELECT R1.ASTRATO_USER_ID, R1.ASTRATO_WORKBOOK_ID, R1.START_TIME, R1.END_TIME, CASE WHEN TIMESTAMPDIFF('minute', LAG(R1.END_TIME) OVER (PARTITION BY R1.ASTRATO_USER_ID, R1.ASTRATO_WORKBOOK_ID ORDER BY R1.START_TIME), R1.START_TIME) > 5 THEN 1 ELSE 0 END AS NEW_SESSION_FLAG FROM RankedQueries R1),SessionGroups AS ( SELECT ASTRATO_USER_ID, ASTRATO_WORKBOOK_ID, START_TIME, END_TIME, SUM(NEW_SESSION_FLAG) OVER ( PARTITION BY ASTRATO_USER_ID, ASTRATO_WORKBOOK_ID ORDER BY START_TIME ) AS SESSION_GROUP FROM SessionizedData),SessionMetrics AS ( SELECT ASTRATO_USER_ID, ASTRATO_WORKBOOK_ID, SESSION_GROUP, MIN(START_TIME) AS WORKBOOK_SESSION_START_TIME, MAX(END_TIME) AS WORKBOOK_SESSION_END_TIME FROM SessionGroups GROUP BY ASTRATO_USER_ID, ASTRATO_WORKBOOK_ID, SESSION_GROUP),LaggedSessionMetrics AS ( SELECT ASTRATO_USER_ID, ASTRATO_WORKBOOK_ID, WORKBOOK_SESSION_START_TIME, WORKBOOK_SESSION_END_TIME, LAG(WORKBOOK_SESSION_END_TIME) OVER ( PARTITION BY ASTRATO_USER_ID ORDER BY WORKBOOK_SESSION_START_TIME ASC ) AS PREVIOUS_WORKBOOK_SESSION_END_TIME FROM SessionMetrics),OverallSessionGroups AS ( SELECT ASTRATO_USER_ID, WORKBOOK_SESSION_START_TIME, WORKBOOK_SESSION_END_TIME, CASE WHEN TIMESTAMPDIFF('minute', PREVIOUS_WORKBOOK_SESSION_END_TIME, WORKBOOK_SESSION_START_TIME) > 5 THEN 1 ELSE 0 END AS NEW_OVERALL_SESSION_FLAG FROM LaggedSessionMetrics),OverallSessionIDs AS ( SELECT ASTRATO_USER_ID, WORKBOOK_SESSION_START_TIME, WORKBOOK_SESSION_END_TIME, SUM(NEW_OVERALL_SESSION_FLAG) OVER ( PARTITION BY ASTRATO_USER_ID -- Scoped by user ORDER BY WORKBOOK_SESSION_START_TIME ) AS OVERALL_SESSION_ID FROM OverallSessionGroups),ReadableSessions AS ( SELECT sm.ASTRATO_USER_ID, sm.ASTRATO_WORKBOOK_ID, md5(CONCAT( sm.ASTRATO_USER_ID, '_', sm.ASTRATO_WORKBOOK_ID, '_', TO_CHAR(sm.WORKBOOK_SESSION_START_TIME, 'YYYYMMDDHH24MISS') )) AS WORKBOOK_SESSION_ID, md5(CONCAT( sm.ASTRATO_USER_ID, '_', os.OVERALL_SESSION_ID )) AS OVERALL_SESSION_ID, -- Scoped and unique sm.WORKBOOK_SESSION_START_TIME, sm.WORKBOOK_SESSION_END_TIME, TIMESTAMPDIFF('second', sm.WORKBOOK_SESSION_START_TIME, sm.WORKBOOK_SESSION_END_TIME) AS WORKBOOK_SESSION_LENGTH_SECONDS FROM SessionMetrics sm JOIN OverallSessionIDs os ON sm.ASTRATO_USER_ID = os.ASTRATO_USER_ID AND sm.WORKBOOK_SESSION_START_TIME = os.WORKBOOK_SESSION_START_TIME AND sm.WORKBOOK_SESSION_END_TIME = os.WORKBOOK_SESSION_END_TIME)SELECT *FROM ReadableSessionsWHERE ASTRATO_USER_ID IS NOT NULL AND WORKBOOK_SESSION_LENGTH_SECONDS > 5);------🅾🆅🅴🆁🅰🅻🅻 🆂🅴🆂🆂🅸🅾🅽🆂create or replace table ASTRATO_QUERY_LOGS.ASTRATO_LOGS.SESSIONS as WITH OverallSessions AS ( SELECT ASTRATO_USER_ID, OVERALL_SESSION_ID AS SESSION_ID, MIN(WORKBOOK_SESSION_START_TIME) AS OVERALL_SESSION_START_TIME, MAX(WORKBOOK_SESSION_END_TIME) AS OVERALL_SESSION_END_TIME, TIMESTAMPDIFF('second', MIN(WORKBOOK_SESSION_START_TIME), MAX(WORKBOOK_SESSION_END_TIME) ) AS OVERALL_SESSION_LENGTH_SECONDS, COUNT(DISTINCT ASTRATO_WORKBOOK_ID) AS WORKBOOKS_IN_SESSION FROM ASTRATO_QUERY_LOGS.ASTRATO_LOGS.WORKBOOK_SESSIONS GROUP BY ASTRATO_USER_ID, OVERALL_SESSION_ID)SELECT *FROM OverallSessions--ORDER BY ASTRATO_USER_ID, SESSION_ID;------🅵🅸🅴🅻🅳🆂 🅸🅽 🆀🆄🅴🆁🅸🅴🆂create or replace table ASTRATO_QUERY_LOGS.ASTRATO_LOGS.FIELDS_IN_QUERY as ( SELECT query_id, TRIM(value,'"') AS astrato_query_field -- Extracting individual fields from the array FROM query_history, LATERAL FLATTEN(input => TRY_PARSE_JSON(astrato_query_meta_data):fields));------Date Dim------create or replace table ASTRATO_QUERY_LOGS.ASTRATO_LOGS.DATE_DIM as ( Select distinct start_time, date(start_time) as start_date, CASE WHEN START_TIME >= CURRENT_DATE() - INTERVAL '30 DAYS' THEN 'Last 30 Days' ELSE 'Older' end as "Last 30 Days", DATE_TRUNC('WEEK', start_time) as "Week Start Date", DATE_TRUNC('MONTH', start_time) as "Month Year", month(start_time) as "Month"From( SELECT v6.OVERALL_SESSION_START_TIME AS START_TIME FROM ASTRATO_LOGS.SESSIONS AS v6 Union all SELECT v5.START_TIME AS START_TIME FROM ASTRATO_LOGS.QUERY_HISTORY AS v5)); RETURN 'Astrato Query history updated successfully';END;$$; --CALL STORE_ASTRATO_QUERY_HISTORY();--Uncomment for initial run (recommended) --📝 🆃🅰🆂🅺 Schedule the stored procedure using Snowflake tasks (optional) CREATE OR REPLACE TASK QUERY_HISTORY_TASK WAREHOUSE = MY_WH SCHEDULE = 'USING CRON 0 0 * * 0 UTC' -- Runs weekly on Sunday at midnight AS CALL STORE_ASTRATO_QUERY_HISTORY(); --📝 🆃🅰🆂🅺 Start task, CREATE TASK always creates tasks in SUSPENDED state. (optional) ALTER TASK QUERY_HISTORY_TASK RESUME; -- 🆅🅸🅴🆆 🅳🅰🆃🅰select *from ASTRATO_QUERY_LOGS.ASTRATO_LOGS.QUERY_HISTORY fetch 15; -----------------------------------------Step 2: Install Workbook
Section titled “Step 2: Install Workbook”Once you have completed the above, create a data connection in Astrato which connects to the database ASTRATO_QUERY_LOGS. Now you’re ready to add/install the workbook.
Step 3: Schedule Astrato metadata sync
Section titled “Step 3: Schedule Astrato metadata sync”Astrato metadata contains lookup values between IDs found in the logs and entities from Astrato, such as names of workbooks, groups, users, collections etc.
The schema will always be called “Astrato Metadata”.
