{ "cells": [ { "cell_type": "code", "execution_count": 1, "id": "1f80cd23-69c4-4572-94d2-135823821822", "metadata": { "tags": [] }, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "/home/ec2-user/anaconda3/envs/python3/lib/python3.10/site-packages/pandas/core/computation/expressions.py:21: UserWarning: Pandas requires version '2.8.0' or newer of 'numexpr' (version '2.7.3' currently installed).\n", " from pandas.core.computation.check import NUMEXPR_INSTALLED\n" ] } ], "source": [ "#snowflake connector and pytest\n", "!pip -q install snowflake-connector-python pytest pytest-sugar \n", "!pip -q install pyecharts absl-py\n", "!pip -q install statsmodels\n", "!pip -q install pmdarima\n", "import snowflake.connector" ] }, { "cell_type": "code", "execution_count": 2, "id": "bbfa9fa9-7275-4a00-ad8c-a156298b18e9", "metadata": { "tags": [] }, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "DEBUG:absl:READY!!!\n" ] } ], "source": [ "# main imports\n", "import snowflake.connector\n", "import pandas as pd\n", "import pyecharts as echarts\n", "import os\n", "import boto3 \n", "\n", "import math\n", "import random\n", "import scipy\n", "import numpy as np\n", "import time\n", "\n", "from datetime import datetime, date, timedelta\n", "\n", "# utils\n", "from getpass import getpass\n", "\n", "# logging and re\n", "from absl import logging\n", "import re\n", "\n", "\n", "log_level = \"DEBUG\"\n", "ticket_code = \"EXP_1\"\n", "\n", "logging.set_verbosity(log_level)\n", "logging.debug(\"READY!!!\")" ] }, { "cell_type": "markdown", "id": "3214d17c-47cd-4ca0-be07-6d7d6763a965", "metadata": { "tags": [] }, "source": [ "## Direct Snowflake Connection" ] }, { "cell_type": "code", "execution_count": 3, "id": "08d79171-cf43-4f91-ba10-bb1d3e0cd03d", "metadata": { "tags": [] }, "outputs": [], "source": [ "sec_id = 'dev/sagemaker-notebook-instance/SNOWFLAKE_PASSWORD'\n", "\n", "\n", "def get_secret_value(name, version=None):\n", " \"\"\"Gets the value of a secret.\n", "\n", " Version (if defined) is used to retrieve a particular version of\n", " the secret.\n", "\n", " \"\"\"\n", " secrets_client = boto3.client(\"secretsmanager\")\n", " kwargs = {'SecretId': name}\n", " if version is not None:\n", " kwargs['VersionStage'] = version\n", " response = secrets_client.get_secret_value(**kwargs)\n", " return response\n", "\n", "\n", "def get_snowflake_creds(username=\"SAGEMAKER\", account=\"orchard\",\n", " warehouse=\"DEV_OWS_ENGINEERING\"):\n", " \"\"\"\n", " Fetches and returns snowflake creds for connecting to snowflake\n", "\n", " Please use this within the scope of a function if using this on a shared instance\n", " This is so that the password is in memory only when its needed and gets dropped \n", " once its no longer required.\n", "\n", " returns:\n", " - creds (dict) - a dictionary containing user creds\n", "\n", " \"\"\"\n", " creds = {\n", " \"user\": username,\n", " \"password\": get_secret_value(sec_id)['SecretString'],\n", " \"account\": \"orchard\",\n", " \"warehouse\": warehouse,\n", " \"protocol\": 'https'\n", " }\n", " return creds\n", "\n", "\n", "def snowflake_connector_factory(creds=None):\n", " \"\"\"\n", " A Factory for creating snowflake connectors.\n", "\n", " This returns the cursor after opening a session with snowflake.\n", "\n", " params:\n", " - creds - snowflake credentials \n", "\n", " returns:\n", " - cursor - snowflake session cursor\n", " \"\"\"\n", " try:\n", " if creds:\n", " _creds = creds\n", " else:\n", " _creds = get_snowflake_creds()\n", " return snowflake.connector.connect(**_creds).cursor()\n", " except Exception as e:\n", " logging.error(f\"Something went wrong - {str(e)}\")\n", "\n", "\n", "def _is_version_number(s):\n", " \"Check and returns true if its a version number\"\n", " return re.search(\"^[0-9][.0-9]*[0-9]$\", s) is not None\n", "\n", "\n", "def test_connection():\n", " \"\"\" tests connection to snowflake \"\"\"\n", " with snowflake_connector_factory() as cs:\n", " try:\n", " cs.execute(\"SELECT current_version()\")\n", " one_row = cs.fetchone()\n", " assert len(one_row) == 1\n", " assert _is_version_number(one_row[0])\n", " logging.info(f\"Your snowflake version - {one_row[0]} PASSED!\")\n", " except Exception as e:\n", " logging.error(f\"Something went wrong - {str(e)}\")" ] }, { "cell_type": "code", "execution_count": 4, "id": "e7d0a5ab-d332-42ba-ae46-b9e877691971", "metadata": { "tags": [] }, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "INFO:absl:Your snowflake version - 7.42.2 PASSED!\n" ] } ], "source": [ "test_connection()" ] }, { "cell_type": "markdown", "id": "4f117e55-aa31-4beb-892c-6cbb8f075d33", "metadata": {}, "source": [] }, { "cell_type": "code", "execution_count": 5, "id": "aad061a5-f42f-4719-a80e-3423ae92f9a5", "metadata": { "tags": [] }, "outputs": [], "source": [ "with snowflake_connector_factory() as cs:\n", " try:\n", " cs.execute(\"USE WAREHOUSE DEV_PERFORMANCE_WAREHOUSE;\")\n", " cs.execute(\"\"\"\n", "with moments_table as (\n", "select * \n", "from DEV_ENGINEERING.EIMPARA.MOMENTS_FOURIERTABLE_APR_SEPT_22_V2_TIKTOK_DATA\n", "),\n", "\n", "trackinfo as \n", " (\n", " select\n", " facts.prod.dim_release.GENREID, \n", " facts.prod.dim_release.RELEASEID,\n", " facts.prod.dim_track.upc,\n", " facts.prod.dim_track.isrc,\n", " facts.prod.dim_release.sale_start_date,\n", " row_number() over (partition by facts.prod.dim_track.isrc order by facts.prod.dim_release.sale_start_date desc) as row_number\n", " from facts.prod.dim_track\n", " inner join moments_table ON facts.prod.dim_track.ISRC = moments_table.ISRC\n", " inner join facts.prod.dim_release on facts.prod.dim_release.RELEASEID = facts.prod.dim_track.upc \n", " ),\n", "\n", "-- trackinfo_extra as (\n", "-- select\n", "-- GENREID, \n", "-- RELEASEID as UPC\n", "-- from facts.prod.dim_release\n", "-- ),\n", "\n", "genre as (\n", "select \n", "GENREID, \n", "GENRENAME \n", "from facts.prod.dim_genre\n", "),\n", "\n", "joined_table as (\n", "select \n", "moments_table.*,\n", "trackinfo.GENREID,\n", "genre.GENRENAME\n", "from moments_table\n", "\n", "LEFT JOIN trackinfo ON moments_table.ISRC = trackinfo.ISRC\n", "LEFT JOIN genre ON trackinfo.GENREID = genre.GENREID\n", "where trackinfo.row_number = 1\n", ")\n", "\n", "select * from joined_table;\n", " \"\"\")\n", " rows = cs.fetchall()\n", " except Exception as e:\n", " logging.error(f\"Something went wrong - {str(e)}\")" ] }, { "cell_type": "code", "execution_count": 6, "id": "906e745a", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "(7225400, 9)\n", "(7225400, 9)\n" ] } ], "source": [ "data_df = pd.DataFrame(rows, columns=map(lambda meta: meta[0], cs.description))\n", "df = data_df.drop_duplicates().copy()\n", "print(data_df.shape)\n", "print(df.shape)" ] }, { "cell_type": "code", "execution_count": 7, "id": "caf4c8f2", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "51610" ] }, "execution_count": 7, "metadata": {}, "output_type": "execute_result" } ], "source": [ "df['ISRC'].nunique()" ] }, { "cell_type": "code", "execution_count": 8, "id": "7113a190", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Min activity date: 2022-04-22\n", "Max activity date: 2022-09-08\n" ] } ], "source": [ "df['ACTIVITY_DATE'] = pd.to_datetime(df['ACTIVITY_DATE'])\n", "df['ACTIVITY_DATE'] = df['ACTIVITY_DATE'].dt.date\n", "\n", "print('Min activity date: ', df['ACTIVITY_DATE'].min())\n", "print('Max activity date: ', df['ACTIVITY_DATE'].max())" ] }, { "cell_type": "code", "execution_count": 9, "id": "7d8c5cd5", "metadata": {}, "outputs": [], "source": [ "def save_dataframe_s3(df):\n", " \n", " '''Saving table to S3'''\n", " \n", " s3 = boto3.client('s3')\n", " bucket_name = 'dev-cucumbers'\n", " today = datetime.today().strftime('%Y%m%d-%H%M%S')\n", " filepath = \"eimpara/TikTok_analysis/genre_level_analysis.{}.csv\".format(today)\n", " csv_buffer = df.to_csv(index=False).encode('utf-8')\n", " s3.put_object(Body=csv_buffer, Bucket=bucket_name, Key=filepath)\n", " print(f\"Table saved to S3 bucket: {bucket_name}, with file name: {filepath}\")" ] }, { "cell_type": "code", "execution_count": 10, "id": "34371ce3", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Table saved to S3 bucket: dev-cucumbers, with file name: eimpara/TikTok_analysis/genre_level_analysis.20231204-145402.csv\n" ] } ], "source": [ "save_dataframe_s3(df)" ] }, { "cell_type": "code", "execution_count": 11, "id": "9420629e", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "/home/ec2-user/anaconda3/envs/python3/lib/python3.10/site-packages/fsspec/registry.py:272: UserWarning: Your installed version of s3fs is very old and known to cause\n", "severe performance issues, see also https://github.com/dask/dask/issues/10276\n", "\n", "To fix, you should specify a lower version bound on s3fs, or\n", "update the current installation.\n", "\n", " warnings.warn(s3_msg)\n" ] }, { "data": { "text/plain": [ "(7225400, 9)" ] }, "execution_count": 11, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# test = pd.read_csv('s3://dev-cucumbers/eimpara/TikTok_analysis/genre_level_analysis.20231204-145402.csv')\n", "# test.shape" ] }, { "cell_type": "code", "execution_count": null, "id": "e95b9e1c", "metadata": {}, "outputs": [], "source": [] } ], "metadata": { "kernelspec": { "display_name": "conda_python3", "language": "python", "name": "conda_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.10.13" } }, "nbformat": 4, "nbformat_minor": 5 }