ARA-C01 Data Engineering — All Questions With Answers

CREATE OR REPLACE DYNAMIC TABLE DT_SALES_AGG
TARGET_LAG = '1 minute'
WAREHOUSE = 'COMPUTE_WH'
AS
SELECT 
    LOCATION_ID,
    SUM(TOTAL_PRICE) as DAILY_REVENUE,
    COUNT(*) as TRANSACTION_COUNT
FROM RAW_SALES
GROUP BY 1;

COPY INTO sales_stage
FROM @my_s3_stage/raw/
FILE_FORMAT = (TYPE = 'JSON')
PATTERN = '.*_2023.*\.json'
ON_ERROR = 'CONTINUE'
PURGE = TRUE;

COPY INTO target_table
FROM (
  SELECT 
    $1:id::int,
    $1:event_time::timestamp,
    $1:payload:user:name::string,
    METADATA$FILENAME,
    METADATA$FILE_ROW_NUMBER
  FROM @my_stage
)
FILE_FORMAT = (TYPE = JSON)
ON_ERROR = 'CONTINUE';

CREATE TASK T2
  WAREHOUSE = 'ETL_WH'
  AFTER T1
  WHEN SYSTEM$STREAM_HAS_DATA('S1')
AS
  INSERT INTO TARGET SELECT * FROM S1;

{
  "pipe": "SALES_PIPE",
  "status": {
    "executionState": "RUNNING",
    "pendingFileCount": 450,
    "lastIngestedTimestamp": "2023-10-25T12:00:00Z"
  }
}

CREATE TASK daily_sales_task
  WAREHOUSE = 'COMPUTE_WH'
  SCHEDULE = 'USING CRON 0 0 * * * UTC'
  WHEN SYSTEM$STREAM_HAS_DATA('sales_stream')
AS
  INSERT INTO sales_summary SELECT * FROM sales_stream;

{
  "query_id": "01b2345...",
  "statistics": {
    "spilling_to_local_storage": "45GB",
    "spilling_to_remote_storage": "120GB",
    "partitions_scanned": 1500,
    "partitions_total": 1500
  }
}