使用 AWS Step Functions 和 AWS Glue 编排基于 Amazon Redshift 的 ETL 工作流（二）

import pgimport boto3import base64from botocore.exceptions import ClientErrorimport json
#uses session manager name to return connection and credential informationdef connection_info(db):
  session = boto3.session.Session()  client = session.client(    service_name='secretsmanager'  )
  get_secret_value_response = client.get_secret_value(SecretId=db)
  if 'SecretString' in get_secret_value_response:    secret = json.loads(get_secret_value_response['SecretString'])  else:    secret = json.loads(base64.b64decode(get_secret_value_response['SecretBinary']))
  return secret
#creates a connection to the clusterdef get_connection(db,db_creds):
  con_params = connection_info(db_creds)
  rs_conn_string = "host=%s port=%s dbname=%s user=%s password=%s" % (con_params['host'], con_params['port'], db, con_params['username'], con_params['password'])  rs_conn = pg.connect(dbname=rs_conn_string)  rs_conn.query("set statement_timeout = 1200000")
  return rs_conn
#submits a query to the clusterdef query(con,statement):    res = con.query(statement)    return res

from redshift_module import pygresql_redshift_common as rs_commonimport sysfrom awsglue.utils import getResolvedOptionsimport boto3
#get job argsargs = getResolvedOptions(sys.argv,['db','db_creds','bucket','file'])db = args['db']db_creds = args['db_creds']bucket = args['bucket']file = args['file']
#get sql statementss3 = boto3.client('s3') sqls = s3.get_object(Bucket=bucket, Key=file)['Body'].read().decode('utf-8')sqls = sqls.split(';')
#get database connectionprint('connecting...')con = rs_common.get_connection(db,db_creds)
#run each sql statementprint("connected...running query...")results = []for sql in sqls[:-1]:    sql = sql + ';'    result = rs_common.query(con, sql)    print(result)    results.append(result)
print(results)

   CREATE EXTERNAL SCHEMA amzreviews    from data catalog   database 'amzreviews'   iam_role 'rolearn'   CREATE EXTERNAL database IF NOT EXISTS;
   CREATE EXTERNAL TABLE amzreviews.reviews(     marketplace varchar(10),     customer_id varchar(15),     review_id varchar(15),     product_id varchar(25),     product_parent varchar(15),     product_title varchar(50),     star_rating int,     helpful_votes int,     total_votes int,     vine varchar(5),     verified_purchase varchar(5),     review_headline varchar(25),     review_body varchar(1024),     review_date date,     year int)   PARTITIONED BY (      product_category varchar(25))   ROW FORMAT SERDE      'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'    STORED AS INPUTFORMAT      'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'    OUTPUTFORMAT      'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'   LOCATION     's3://amazon-reviews-pds/parquet/';


   ALTER TABLE amzreviews.reviews ADD   partition(product_category='Apparel')    location 's3://amazon-reviews-pds/parquet/product_category=Apparel/'   partition(product_category='Automotive')    location 's3://amazon-reviews-pds/parquet/product_category=Automotive'   partition(product_category='Baby')    location 's3://amazon-reviews-pds/parquet/product_category=Baby'   partition(product_category='Beauty')    location 's3://amazon-reviews-pds/parquet/product_category=Beauty'   partition(product_category='Books')    location 's3://amazon-reviews-pds/parquet/product_category=Books'   partition(product_category='Camera')    location 's3://amazon-reviews-pds/parquet/product_category=Camera'   partition(product_category='Grocery')    location 's3://amazon-reviews-pds/parquet/product_category=Grocery'   partition(product_category='Furniture')    location 's3://amazon-reviews-pds/parquet/product_category=Furniture'   partition(product_category='Watches')    location 's3://amazon-reviews-pds/parquet/product_category=Watches'   partition(product_category='Lawn_and_Garden')    location 's3://amazon-reviews-pds/parquet/product_category=Lawn_and_Garden';
   CREATE TABLE reviews(     marketplace varchar(10),     customer_id varchar(15),     review_id varchar(15),     product_id varchar(25) DISTKEY,     product_parent varchar(15),     product_title varchar(50),     star_rating int,     helpful_votes int,     total_votes int,     vine varchar(5),     verified_purchase varchar(5),     review_date date,     year int,     product_category varchar(25))
     SORTKEY (        review_date       );