{ "cells": [ { "cell_type": "code", "execution_count": 31, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "The dotenv extension is already loaded. To reload it, use:\n", " %reload_ext dotenv\n" ] } ], "source": [ "import datetime\n", "import json\n", "import itertools\n", "import concurrent.futures\n", "import os\n", "\n", "from snowflake import connector\n", "\n", "%load_ext dotenv" ] }, { "cell_type": "code", "execution_count": 32, "metadata": {}, "outputs": [], "source": [ "#%reload_ext dotenv" ] }, { "cell_type": "code", "execution_count": 33, "metadata": {}, "outputs": [], "source": [ "AWS_ACCESS_KEY_ID = os.environ.get('AWS_ACCESS_KEY_ID')\n", "AWS_SECRET_ACCESS_KEY = os.environ.get('AWS_SECRET_ACCESS_KEY')\n", "AWS_REGION = os.environ.get('AWS_REGION')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Utils" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Snowflake" ] }, { "cell_type": "code", "execution_count": 34, "metadata": {}, "outputs": [], "source": [ "SNOWFLAKE_DATABASE = os.environ.get('SNOWFLAKE_DATABASE')\n", "SNOWFLAKE_SCHEMA = os.environ.get('SNOWFLAKE_SCHEMA')\n", "snowflake_connection_params = dict(\n", " account=os.environ.get('SNOWFLAKE_ACCOUNT'),\n", " role=os.environ.get('SNOWFLAKE_ROLE'),\n", " user=os.environ.get('SNOWFLAKE_USER'),\n", " password=os.environ.get('SNOWFLAKE_PASSWORD'),\n", " database=os.environ.get('SNOWFLAKE_DATABASE'),\n", " schema=os.environ.get('SNOWFLAKE_SCHEMA'),\n", " warehouse=os.environ.get('SNOWFLAKE_WAREHOUSE'),\n", " autocommit=True,\n", " session_parameters={\n", " 'CLIENT_STORE_TEMPORARY_CREDENTIAL': True,\n", " }\n", ")\n", "\n", "snowflake_connection = connector.connect(**snowflake_connection_params)\n", "cursor = snowflake_connection.cursor()" ] }, { "cell_type": "code", "execution_count": 35, "metadata": {}, "outputs": [], "source": [ "copy_file_sql = \"\"\"\n", "COPY INTO {db}.{schema}.{temp_staging_raw_table}\n", "FROM '{s3_path}'\n", "FILE_FORMAT = (\n", " FIELD_DELIMITER='\\t'\n", " RECORD_DELIMITER='\\n'\n", " DATE_FORMAT=YYYYMMDD\n", " COMPRESSION=AUTO\n", " TRIM_SPACE=TRUE\n", " EMPTY_FIELD_AS_NULL=TRUE\n", " SKIP_HEADER=3\n", ")\n", "ON_ERROR=CONTINUE\n", "FORCE = TRUE\n", "CREDENTIALS=(\n", " AWS_KEY_ID='{aws_key_id}'\n", " AWS_SECRET_KEY='{aws_secret_key}'\n", ");\n", "\"\"\"\n", "\n", "activity_report_table = \"\"\"\n", "CREATE OR REPLACE TRANSIENT TABLE {db}.{schema}.amazon_unlimited_daily_activity_report\n", "(\n", " reportingstartdate DATE,\n", " activitytype INTEGER,\n", " customerid VARCHAR,\n", " streamsourceid VARCHAR,\n", " vendorid VARCHAR,\n", " vendorname VARCHAR,\n", " smetrackproductid VARCHAR,\n", " providergenre VARCHAR,\n", " artistname VARCHAR,\n", " trackname VARCHAR,\n", " isrctracknumber VARCHAR,\n", " upccode VARCHAR,\n", " producttitle VARCHAR,\n", " grid VARCHAR,\n", " statementtypekey VARCHAR,\n", " locationoftrackinplaylist VARCHAR,\n", " devicetype VARCHAR,\n", " ostype VARCHAR,\n", " referralsourcetype VARCHAR,\n", " royaltybearingplay INTEGER,\n", " timestamp TIMESTAMP,\n", " lengthofstream INTEGER,\n", " numberofweeklystreams INTEGER,\n", " selectionsourcetype VARCHAR,\n", " selectionsourcedetail VARCHAR,\n", " serviceloggingtimestamp TIMESTAMP,\n", " utcoffset VARCHAR,\n", " transctiontype INTEGER\n", ");\"\"\"\n", "\n", "playlist_report_table = \"\"\"\n", "CREATE OR REPLACE TRANSIENT TABLE {db}.{schema}.amazon_unlimited_daily_playlist_report\n", "(\n", " reportingstartdate DATE,\n", " playlistid VARCHAR,\n", " playlistname VARCHAR,\n", " playlistgenre VARCHAR,\n", " numberoftracks INTEGER\n", ");\"\"\"\n", "\n", "user_report_table = \"\"\"\n", "CREATE OR REPLACE TRANSIENT TABLE {db}.{schema}.amazon_unlimited_daily_user_report\n", "(\n", " reportingstartdate DATE,\n", " customerid VARCHAR,\n", " country VARCHAR,\n", " location VARCHAR,\n", " cohortmonthofcurrentstreamingtier INTEGER,\n", " subscriptionproduct VARCHAR,\n", " subscriptiontype VARCHAR,\n", " subscriptiondetail VARCHAR,\n", " gender VARCHAR,\n", " birthdate VARCHAR,\n", " numberoftotalplaylists INTEGER,\n", " numberofpersonalplaylists INTEGER,\n", " numberofthirdpartyplaylists INTEGER,\n", " numberofsmeplaylists INTEGER,\n", " numberofpromotedplaylists INTEGER,\n", " numberofsmeaccountfollows INTEGER,\n", " numberoftotaltracksincollection INTEGER,\n", " numberofsmetracksincollection INTEGER\n", ");\"\"\"\n" ] }, { "cell_type": "code", "execution_count": 36, "metadata": {}, "outputs": [], "source": [ "for table_template in [activity_report_table, playlist_report_table, user_report_table]:\n", " sql = table_template.format(db=SNOWFLAKE_DATABASE, schema=SNOWFLAKE_SCHEMA)\n", " cursor.execute(sql)" ] }, { "cell_type": "code", "execution_count": 37, "metadata": {}, "outputs": [], "source": [ "def try_copy_file(s3_file_path, snowflake_table):\n", " sql = copy_file_sql.format(\n", " db=SNOWFLAKE_DATABASE, \n", " schema=SNOWFLAKE_SCHEMA,\n", " temp_staging_raw_table=snowflake_table,\n", " s3_path=s3_file_path,\n", " aws_key_id=AWS_ACCESS_KEY_ID,\n", " aws_secret_key=AWS_SECRET_ACCESS_KEY\n", " )\n", " return cursor.execute(sql).fetchone()\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Data Gathering" ] }, { "cell_type": "code", "execution_count": 38, "metadata": {}, "outputs": [], "source": [ "start_date = datetime.datetime.fromisoformat('2018-01-01')\n", "scan_dates = [start_date + datetime.timedelta(days=i) \n", " for i in range((datetime.datetime.now() - start_date).days - 1)]\n", "scan_dates = [start_date + datetime.timedelta(days=i) \n", " for i in range((datetime.datetime.fromisoformat('2018-03-31') - start_date).days + 1)]" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "scan_dates" ] }, { "cell_type": "code", "execution_count": 40, "metadata": { "scrolled": true }, "outputs": [], "source": [ "param_dates = [(date.strftime('%Y-%m-%d'), date.strftime('%Y%m%d')) for date in scan_dates]\n", "orgs = ['ORCA', 'ORED']\n", "countries = ['AT', 'AU', 'DE', 'ES', 'FR', 'GB', 'IT', 'US']\n", "report_types = ['Activity', 'Playlist', 'User']" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "s3_file_path_template = (\n", " 's3://cucumbers/AmazonUnlimited/archives'\n", " '/{date0}/clean/{org}_{country}_{date1}_Daily_{report_type}_Report.txt.gz')\n", "\n", "\n", "file_params = []\n", "\n", "for (date0, date1), org, country, report_type in itertools.product(param_dates, orgs, countries, report_types):\n", " s3_file_path= s3_file_path_template.format(\n", " date0=date0, date1=date1, org=org, country=country, report_type=report_type)\n", " table_name = 'amazon_unlimited_daily_{}_report'.format(report_type.lower())\n", " file_params.append({\n", " 'upload_params': (s3_file_path, table_name),\n", " 'file_meta': [date0, org, country, report_type]\n", " })\n", "\n", "upload_results = []\n", "with concurrent.futures.ThreadPoolExecutor(max_workers=6) as executor:\n", " future_to_url = {\n", " executor.submit(try_copy_file, *params['upload_params']): params['file_meta'] \n", " for params in file_params}\n", " for future in concurrent.futures.as_completed(future_to_url):\n", " file_meta = future_to_url[future]\n", " upload_result = future.result()\n", " upload_results.append([file_meta, upload_result])\n", " print(file_meta)\n", " \n", "with open('copy_results_201801-201803.json', 'w') as f:\n", " json.dump(upload_results, f)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "!ls" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "upload_results" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.7.1" } }, "nbformat": 4, "nbformat_minor": 2 }