使用 AWS Step Functions 为 Amazon Redshift 编排 ELT 流程

{"Comment":"A simple ETL example that submits a Job to AWS Batch","StartAt":"DBConnectionInit",..."Parallel":{"Type":"Parallel","Next":"SalesFACTInit","ResultPath":"$.status","Branches":[...},"SalesFACTInit":{"Type":"Pass","Next":"SubmitStoreSalesFACTJob","Result":"SalesFACT","ResultPath":"$.stepId"},"SubmitStoreSalesFACTJob":{"Type":"Task","Resource":"arn:aws:lambda:us-west-2:1234567890:function:StepFunctionsSample-JobStatusPol-SubmitJobFunction-5M2HCJIG81R1","Next":"GetStoreSalesFACTJobStatus"},"GetStoreSalesFACTJobStatus":{"Type":"Task","Resource":"arn:aws:lambda:us-west-2:1234567890:function:StepFunctionsSample-JobStatusPoll-CheckJobFunction-1SKER18I6FU24","Next":"CheckStoreSalesFACTJobStatus","InputPath":"$","ResultPath":"$.status"},"CheckStoreSalesFACTJobStatus":{"Type":"Choice","Choices":[{"Variable":"$.status","StringEquals":"FAILED","Next":"FailState"},{"Variable":"$.status","StringEquals":"SUCCEEDED","Next":"GetFinalStoreSalesFACTJobStatus"}],"Default":"StoreSalesFACTWait30Seconds"},...}}

{"DBConnection":{.."SalesFACT":{"jobName":"my-job","jobDefinition":"arn:aws:batch:us-west-2:1234567890:job-definition/JobDefinition-cd6aa175c07fb2a:1","jobQueue":"arn:aws:batch:us-west-2:1234567890:job-queue/JobQueue-217beecdb0caa3f","wait_time":60,"containerOverrides":{"environment":[{"name":"BATCH_FILE_TYPE","value":"script_psql"},{"name":"BATCH_FILE_S3_URL","value":"s3://salamander-us-east-1/reinvent2018/ant353/etlscript/psql_rs.sh"},{"name":"BATCH_FILE_SQL_S3_URL","value":"s3://salamander-us-east-1/reinvent2018/ant353/etlscript/store_sales.sql"},{"name":"DATASET_DATE","value":"2003-01-02"}]}}}

\set s3datareadrolevar 'aws_iam_role=' :s3datareadrole-- This transform ETL will refresh data for the store_sales table-- Start a new transactionbegin transaction;
-- Create a stg_store_sales staging table and COPY data from input S3 location it with updated rows from SALES_UPDATEDROP TABLE if exists public.stg_store_sales;CREATE TABLE public.stg_store_sales(sold_date DATE ENCODE lzo,sold_time INTEGER ENCODE lzo,i_item_id CHAR(16) ENCODE lzo,c_customer_id CHAR(16) ENCODE lzo,cd_demo_sk INTEGER ENCODE lzo,hd_income_band_sk INTEGER ENCODE lzo,hd_buy_potential CHAR(15) ENCODE lzo,hd_dep_count INTEGER ENCODE lzo,hd_vehicle_count INTEGER ENCODE lzo,ca_address_id CHAR(16) ENCODE lzo,s_store_id CHAR(16) ENCODE lzo,p_promo_id CHAR(16) ENCODE lzo,ss_ticket_number INTEGER ENCODE lzo,ss_quantity INTEGER ENCODE lzo,ss_wholesale_cost NUMERIC(7,2) ENCODE lzo,ss_list_price NUMERIC(7,2) ENCODE lzo,ss_sales_price NUMERIC(7,2) ENCODE lzo,ss_ext_discount_amt NUMERIC(7,2) ENCODE lzo,ss_ext_sales_price NUMERIC(7,2) ENCODE lzo,ss_ext_wholesale_cost NUMERIC(7,2) ENCODE lzo,ss_ext_list_price NUMERIC(7,2) ENCODE lzo,ss_ext_tax NUMERIC(7,2) ENCODE lzo,ss_coupon_amt NUMERIC(7,2) ENCODE lzo,ss_net_paid NUMERIC(7,2) ENCODE lzo,ss_net_paid_inc_tax NUMERIC(7,2) ENCODE lzo,ss_net_profit NUMERIC(7,2) ENCODE lzo)BACKUP NODISTSTYLE EVEN;
\set s3loc 's3://salamander-us-east-1/reinvent2018/ant353/store_sales/saledate=' :dt '/'-- COPY input data to the staging tablecopy public.stg_store_salesfrom:'s3loc'CREDENTIALS :'s3datareadrolevar'DELIMITER '~' gzip region 'us-east-1';
-- Delete any rows from target store_sales for the input date for idempotencydelete from store_sales where ss_sold_date_sk in (select d_date_sk from date_dim where d_date=:'dt');--Insert data from staging table to the target TABLEINSERT INTO store_sales(ss_sold_date_sk,ss_sold_time_sk,ss_item_sk,ss_customer_sk,ss_cdemo_sk,ss_hdemo_sk,ss_addr_sk,ss_store_sk,ss_promo_sk,ss_ticket_number,ss_quantity,ss_wholesale_cost,ss_list_price,ss_sales_price,ss_ext_discount_amt,ss_ext_sales_price,ss_ext_wholesale_cost,ss_ext_list_price,ss_ext_tax,ss_coupon_amt,ss_net_paid,ss_net_paid_inc_tax,ss_net_profit)SELECT date_dim.d_date_sk ss_sold_date_sk,time_dim.t_time_sk ss_sold_time_sk,item.i_item_sk ss_item_sk,customer.c_customer_sk ss_customer_sk,customer_demographics.cd_demo_sk ss_cdemo_sk,household_demographics.hd_demo_sk ss_hdemo_sk,customer_address.ca_address_sk ss_addr_sk,store.s_store_sk ss_store_sk,promotion.p_promo_sk ss_promo_sk,ss_ticket_number,ss_quantity,ss_wholesale_cost,ss_list_price,ss_sales_price,ss_ext_discount_amt,ss_ext_sales_price,ss_ext_wholesale_cost,ss_ext_list_price,ss_ext_tax,ss_coupon_amt,ss_net_paid,ss_net_paid_inc_tax,ss_net_profitFROM stg_store_sales AS store_salesJOIN date_dim ON store_sales.sold_date = date_dim.d_dateLEFT JOIN time_dim ON store_sales.sold_time = time_dim.t_timeLEFT JOIN itemON store_sales.i_item_id = item.i_item_idAND i_rec_end_date IS NULLLEFT JOIN customer ON store_sales.c_customer_id = customer.c_customer_idLEFT JOIN customer_demographics ON store_sales.cd_demo_sk = customer_demographics.cd_demo_skLEFT JOIN household_demographicsON store_sales.hd_income_band_sk = household_demographics.hd_income_band_skAND store_sales.hd_buy_potential = household_demographics.hd_buy_potentialAND store_sales.hd_dep_count = household_demographics.hd_dep_countAND store_sales.hd_vehicle_count = household_demographics.hd_vehicle_countLEFT JOIN customer_address ON store_sales.ca_address_id = customer_address.ca_address_idLEFT JOIN storeON store_sales.s_store_id = store.s_store_idAND s_rec_end_date IS NULLLEFT JOIN promotion ON store_sales.p_promo_id = promotion.p_promo_id;
--drop staging table
DROP TABLE if exists public.stg_store_sales;
-- End transaction and commitend transaction;

“{ "DBConnection":{ "jobName":"…alue":"s3://salamander-us-east-1/reinvent2018/ant353/etlscript/store_sales.sql" }, { "name":"DATASET_DATE", "value":"2003-01-02" } ] } } }”

SELECT c_last_name,c_first_name,ca_city,bought_city,ss_ticket_number,amt,profitFROM (SELECT ss_ticket_number,ss_customer_sk,ca_city bought_city,Sum(ss_coupon_amt) amt,Sum(ss_net_profit) profitFROM store_sales,date_dim,store,household_demographics,customer_addressWHERE store_sales.ss_sold_date_sk = date_dim.d_date_skAND store_sales.ss_store_sk = store.s_store_skAND store_sales.ss_hdemo_sk = household_demographics.hd_demo_skAND store_sales.ss_addr_sk = customer_address.ca_address_skAND ( household_demographics.hd_dep_count = 6OR household_demographics.hd_vehicle_count = 0 )AND d_date = '2003-01-02'GROUP BY ss_ticket_number,ss_customer_sk,ss_addr_sk,ca_city) dn,customer,customer_address current_addrWHERE ss_customer_sk = c_customer_skAND customer.c_current_addr_sk = current_addr.ca_address_skAND current_addr.ca_city <> bought_cityORDER BY c_last_name,c_first_name,ca_city,bought_city,ss_ticket_numberLIMIT 100;