<jupyter_start><jupyter_text>Load TSV Data From S3/Athena into Redshift We can leverage our previously created table in Amazon Athena with its metadata and schema information stored in the AWS Glue Data Catalog to access our data in S3 through Redshift Spectrum. All we need to do is create an external schema in Redshift, point it to our AWS Glue Data Catalog, and point Redshift to the database we’ve created.<jupyter_code>import boto3
import sagemaker

# Get region 
session = boto3.session.Session()
region_name = session.region_name

# Get SageMaker session & default S3 bucket
sagemaker_session = sagemaker.Session()
bucket = sagemaker_session.default_bucket()

redshift = boto3.client('redshift')
secretsmanager = boto3.client('secretsmanager')<jupyter_output><empty_output><jupyter_text>Setup Redshift Connection Via SQLAlchemyThe Python SQL Toolkit and Object Relational Mapper (https://pypi.org/project/SQLAlchemy/)<jupyter_code>#!pip install -q SQLAlchemy==1.3.13
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import pandas as pd<jupyter_output><empty_output><jupyter_text>Get Redshift Credentials<jupyter_code>import json

secret = secretsmanager.get_secret_value(SecretId='dsoaws_redshift_login')
cred = json.loads(secret['SecretString'])

master_user_name = cred[0]['username']
master_user_pw = cred[1]['password']<jupyter_output><empty_output><jupyter_text>Redshift Configuration Parameters<jupyter_code>redshift_cluster_identifier = 'dsoaws'

database_name_redshift = 'dsoaws'
database_name_athena = 'dsoaws'

redshift_port = '5439'

schema_redshift = 'redshift'
schema_athena = 'athena'

table_name_tsv = 'amazon_reviews_tsv'<jupyter_output><empty_output><jupyter_text>Please Wait for Cluster Status  `Available`<jupyter_code>import time

response = redshift.describe_clusters(ClusterIdentifier=redshift_cluster_identifier)
cluster_status = response['Clusters'][0]['ClusterStatus']
print(cluster_status)

while cluster_status != 'available':
    time.sleep(10)
    response = redshift.describe_clusters(ClusterIdentifier=redshift_cluster_identifier)
    cluster_status = response['Clusters'][0]['ClusterStatus']
    print(cluster_status)<jupyter_output><empty_output><jupyter_text>Get Redshift Endpoint Address & IAM Role<jupyter_code>redshift_endpoint_address = response['Clusters'][0]['Endpoint']['Address']
iam_role = response['Clusters'][0]['IamRoles'][0]['IamRoleArn']

print('Redshift endpoint: {}'.format(redshift_endpoint_address))
print('IAM Role: {}'.format(iam_role))<jupyter_output><empty_output><jupyter_text>Connect to Redshift Database Engine<jupyter_code>engine = create_engine('postgresql://{}:{}@{}:{}/{}'.format(master_user_name, master_user_pw, redshift_endpoint_address, redshift_port, database_name_redshift))<jupyter_output><empty_output><jupyter_text>Configure Session<jupyter_code>session = sessionmaker()
session.configure(bind=engine)
s = session()<jupyter_output><empty_output><jupyter_text>Redshift SpectrumAmazon Redshift Spectrum directly queries data in S3, using the same SQL syntax of Amazon Redshift. You can also run queries that span both the frequently accessed data stored locally in Amazon Redshift and your full datasets stored cost-effectively in S3.To use Redshift Spectrum, your cluster needs authorization to access data catalog in Amazon Athena and your data files in Amazon S3. You provide that authorization by referencing an AWS Identity and Access Management (IAM) role that is attached to your cluster. To use this capability in from your Amazon SageMaker notebook:* Register your Athena database `dsoaws` with Redshift Spectrum* Query Your Data in Amazon S3  Register Athena Database `dsoaws` with Redshift Spectrum to Access the Data Directly in S3 using Glue Data CatalogWith just one command, you can query the S3 data lake from Amazon Redshift without moving any data into our data warehouse. This is the power of Redshift Spectrum. Note the `FROM DATA CATALOG` below.  This is pulling the table and schema information from the Glue Data Catalog (ie. Hive Metastore).<jupyter_code>statement = """
CREATE EXTERNAL SCHEMA IF NOT EXISTS {} FROM DATA CATALOG 
    DATABASE '{}' 
    IAM_ROLE '{}'
    REGION '{}'
    CREATE EXTERNAL DATABASE IF NOT EXISTS
""".format(schema_athena, database_name_athena, iam_role, region_name)

print(statement)
s.execute(statement)
s.commit()<jupyter_output><empty_output><jupyter_text>Run Sample Query on S3 Data through Redshift Spectrum<jupyter_code>statement = """
SELECT product_category, COUNT(star_rating) AS count_star_rating
    FROM {}.{}
    GROUP BY product_category
    ORDER BY count_star_rating DESC
""".format(schema_athena, table_name_tsv)

print(statement)
df = pd.read_sql_query(statement, engine)
df.head(5)<jupyter_output><empty_output><jupyter_text>But now, let’s actually copy some data from S3 into Amazon Redshift. Let’s pull in customer reviews data from the years 2014 and 2015.   Load TSV Data Into RedshiftCreate Redshift tables with Customer Reviews data for each year we wish to load.  Create `redshift` Schema<jupyter_code>statement = """CREATE SCHEMA IF NOT EXISTS {}""".format(schema_redshift)

s = session()
s.execute(statement)
s.commit()<jupyter_output><empty_output><jupyter_text>Create Redshift Tables for Each Year We Wish to Load When you create a table, you can specify one or more columns as the **sort key**. Amazon Redshift stores your data on disk in sorted order according to the sort key. This means, you can optimize your table by choosing a sort key that reflects your most frequently used query types. If you query a lot of recent data, you can specify a timestamp column as the sort key. If you frequently query based on range or equality filtering on one column, you should choose that column as the sort key. As we are going to run a lot of queries in the next chapter filtering on `product_category`, let’s choose that one as our sort key. You can also define a distribution style for every table. When you load data into a table, Redshift distributes the rows of the table among your cluster nodes according to the table’s distribution style. When you run a query, the query optimizer redistributes the rows to the cluster nodes as needed to perform any joins and aggregations. So our goal should be to optimize the rows distribution to minimize needed data movements. There are three distribution styles from which you can choose from: * KEY distribution - distribute the rows according to the values in one column* ALL distribution - distribute a copy of the entire table to every node* EVEN distribution - the rows are distributed across all nodes in a round-robin-fashion which is the default distribution styleFor our table, we’ve chosen **KEY distribution** based on `product_id` as this column has a high cardinality, shows an even distribution and can be used to join with other tables. Now we are ready to copy the data from S3 into our new Redshift table.<jupyter_code># Create table function, pass session, table name prefix and start & end year

def create_redshift_table_tsv(session, table_name_prefix, start_year, end_year):
    for year in range(start_year, end_year + 1, 1):
        current_table_name = table_name_prefix+'_'+str(year)
        statement = """
        CREATE TABLE IF NOT EXISTS redshift.{}( 
             marketplace varchar(2),
             customer_id varchar(8),
             review_id varchar(14),
             product_id varchar(10) DISTKEY,
             product_parent varchar(9),
             product_title varchar(400),
             product_category varchar(24),
             star_rating int,
             helpful_votes int,
             total_votes int,
             vine varchar(1),
             verified_purchase varchar(1),
             review_headline varchar(128),
             review_body varchar(65535),
             review_date varchar(10),
             year int)  SORTKEY (product_category)
        """.format(current_table_name)

        #print(statement)
        session.execute(statement)
    session.commit()
        
    print("Done.")
create_redshift_table_tsv(s, 'amazon_reviews_tsv', 2014, 2015)<jupyter_output><empty_output><jupyter_text>Insert TSV Data into New Redshift TablesFor such bulk inserts, you can either use a `COPY` command, or an `INSERT INTO` command. In general, the `COPY` command is preferred, as it loads data in parallel and more efficiently from Amazon S3, or other supported data sources. If you are loading data or a subset of data from one table into another, you can use the `INSERT INTO` command with a `SELECT` clause for high-performance data insertion. As we’re loading our data from the `athena.amazon_reviews_tsv` table, let’s choose this option.<jupyter_code># INSERT INTO function, pass session, table name prefix and start & end year

def insert_into_redshift_table_tsv(session, table_name_prefix, start_year, end_year):
    for year in range(start_year, end_year + 1, 1):
        print(year)
        current_table_name = table_name_prefix+'_'+str(year)
        statement = """
            INSERT 
            INTO
                redshift.{}
                SELECT
                    marketplace,
                    customer_id,
                    review_id,
                    product_id,
                    product_parent,
                    product_title,
                    product_category,
                    star_rating,
                    helpful_votes,
                    total_votes,
                    vine,
                    verified_purchase,
                    review_headline,
                    review_body,
                    review_date,
                    CAST(DATE_PART_YEAR(TO_DATE(review_date, 'YYYY-MM-DD')) AS INTEGER) AS year
                FROM
                    athena.amazon_reviews_tsv             
                WHERE
                    year = {}
            """.format(current_table_name, year)
        #print(statement)
        session.execute(statement)
        session.commit()        
    print("Done.")<jupyter_output><empty_output><jupyter_text>_The following `INSERT INTO` command can take some time to complete. Please be patient._<jupyter_code>insert_into_redshift_table_tsv(s, 'amazon_reviews_tsv', 2014, 2015)<jupyter_output><empty_output><jupyter_text>You might notice that we use a date conversion to parse the year out of our `review_date` column and store it in a separate `year` column which we then use to filter on all records from 2015. This is an example on how you can simplify ETL tasks, as we’re putting our data transformation logic directly in a `SELECT` query and ingest the result into Redshift. Another way to optimize our tables would be to create them as a sequence of time-series tables, especially when our data has a fixed retention period. Let’s say we want to store data of the last 2 years (24 months) in our data warehouse, and update with new data once a month. If you create one table per month, you can easily remove old data simply by running a `DROP TABLE` command on the corresponding table. This approach is much faster than running a large-scale DELETE process and also saves you from having to run a subsequent VACUUM process to reclaim space and re-sort the rows.<jupyter_code>%%javascript

try {
    Jupyter.notebook.save_checkpoint();
    Jupyter.notebook.session.delete();
}
catch(err) {
    // NoOp
}<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.06952714535901927, 'fraction_numerical': 0.008231173380035026, 'mean_word_length': 4.053539823008849, 'pattern_counts': {'":': 0, '<': 61, '<?xml version=': 0, '>': 61, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '15', 'dir': 'jupyter-structured-clean-dedup', 'id': '123573', 'license': 'Apache-2.0', 'n_tokens_mistral': 3264, 'n_tokens_neox': 3084, 'n_words': 1187, 'path': '04_ingest/07_Load_TSV_Data_From_Athena_Into_Redshift.ipynb', 'repo_name': 'bwcx-aws/data-science-on-aws-workshop', 'size': '11420'}