Usage Metrics Extractor.ipynb

{"cells":[{"cell_type":"markdown","source":["## IMPORTS"],"metadata":{"nteract":{"transient":{"deleting":false}}},"id":"bf4ec2d8-c7a2-44b7-afb3-1d39308995f7"},{"cell_type":"code","source":["import requests\n","import time\n","import pandas as pd\n","from datetime import datetime\n","from pyspark.sql.functions import *\n","from pyspark.sql.types import *"],"outputs":[],"execution_count":null,"metadata":{},"id":"856d4815-b8e8-44c4-b3ef-adccc832a534"},{"cell_type":"markdown","source":["## URIs + HEADERS for requests"],"metadata":{"nteract":{"transient":{"deleting":false}}},"id":"4d89c8b8-1dd7-46f9-bf5f-c0dc8b55432d"},{"cell_type":"code","source":["pbi_resource = \"https://analysis.windows.net/powerbi/api\"\n","pbi_Uri = 'https://api.powerbi.com/v1.0/myorg/'"],"outputs":[],"execution_count":null,"metadata":{"jupyter":{"outputs_hidden":false,"source_hidden":false},"nteract":{"transient":{"deleting":false}}},"id":"dc78c524-cde5-48d8-bdf9-c7128b4f6939"},{"cell_type":"markdown","source":["## Unified functions"],"metadata":{"nteract":{"transient":{"deleting":false}}},"id":"d672d87e-7f42-4845-9c07-a7540dece280"},{"cell_type":"code","source":["def get_token():\n","    return mssparkutils.credentials.getToken(pbi_resource)"],"outputs":[],"execution_count":null,"metadata":{"jupyter":{"outputs_hidden":false,"source_hidden":false},"nteract":{"transient":{"deleting":false}}},"id":"fc226ef5-90a2-4a6d-9bc1-c16745aa888d"},{"cell_type":"code","source":["def get_powerbiAPIclusterURI():\n","    fullurl = pbi_Uri+'datasets'\n","    pbi_access_token = get_token()\n","    headers = {'Content-Type': 'application/json', 'Authorization': f'Bearer {pbi_access_token}'}\n","    response = requests.get(fullurl, headers=headers)\n","    unaltered = response.json()['@odata.context']\n","    stripped = unaltered.split('/')\n","    return f'https://{stripped[2]}/beta/myorg/groups'\n","\n","clusteredURI = get_powerbiAPIclusterURI()"],"outputs":[],"execution_count":null,"metadata":{"jupyter":{"outputs_hidden":false,"source_hidden":false},"nteract":{"transient":{"deleting":false}}},"id":"6448c4da-4db9-4929-a25f-14fc1d7f2034"},{"cell_type":"markdown","source":["### Functions definition"],"metadata":{"nteract":{"transient":{"deleting":false}}},"id":"8899202f-01ba-40e5-a665-ea7b9ac11352"},{"cell_type":"code","source":["def get_AccessibleWorkspaces():\n","    fullUrl = pbi_Uri+\"/groups?$filter=type eq 'Workspace'\"\n","    pbi_access_token = get_token()\n","    headers = {'Content-Type': 'application/json', 'Authorization': f'Bearer {pbi_access_token}'}\n","    response = requests.get(fullUrl, headers=headers)\n","    return response.json()['value']\n","\n","def get_WorkspaceUsageMetricsId(wsId):\n","    fullurl = f'{clusteredURI}/{wsId}/usageMetricsReportV2'\n","    # print('Asked for token')\n","    pbi_access_token = get_token()\n","    # print('Token received')\n","\n","    headers = {'Content-Type': 'application/json', 'Authorization': f'Bearer {pbi_access_token}'}\n","    attemps = 0\n","    while attemps < 4:\n","       # print('Starting ' + str(attemps))\n","        try:\n","            response = requests.get(fullurl, headers=headers, timeout=60)\n","        #    print('Received data')\n","            return response.json()['models'][0]['dbName']\n","        except:\n","        #    print('Fallin asleep')\n","            time.sleep(30)\n","        #    print('Awaken')\n","            attemps += 1\n","\n","def post_ExecuteQuery(wsId, dsId, daxQ):\n","    fullurl = f'{pbi_Uri}/groups/{wsId}/datasets/{dsId}/executeQueries'\n","    pbi_access_token = get_token()\n","    headers = {'Content-Type': 'application/json', 'Authorization': f'Bearer {pbi_access_token}'}\n","    content = {\"queries\": [{\"query\": daxQ}],\"serializerSettings\": {\"includeNulls\": True}}\n","    attemps = 0\n","    while attemps < 4:\n","        try:\n","            response = requests.post(fullurl, json=content, headers = headers, timeout=60).json()['results'][0]['tables'][0]\n","            return response\n","        except:\n","            time.sleep(30)\n","            attemps += 1\n","\n","def replace_ColumnNames(frame, replacingName):\n","    return frame.toDF(*(c.replace(replacingName,'').replace('[','').replace(']','') for c in frame.columns))\n","\n","def extract_DataFrame(response_list, preCreatedSchema):\n","    rl = spark.createDataFrame(response_list)\n","    # print('Frame Created')\n","    # display(rl)\n","    rl = rl.select(explode(rl.rows)).select(\"col\").rdd.flatMap(lambda x: x).collect()\n","    #print('Columns Extracted')\n","    if not schema:\n","        rl = spark.createDataFrame(rl)\n","    else:\n","        rl = spark.createDataFrame(rl, preCreatedSchema)\n","    #print('Extraction Completed')\n","    return rl"],"outputs":[],"execution_count":null,"metadata":{"jupyter":{"outputs_hidden":false,"source_hidden":false},"nteract":{"transient":{"deleting":false}}},"id":"52be93a9-b427-40dd-b652-ba57455575fe"},{"cell_type":"markdown","source":["### Tables"],"metadata":{"nteract":{"transient":{"deleting":false}}},"id":"7b08a153-9ccc-4be7-8f20-9674eda19079"},{"cell_type":"code","source":["tblsInLK = spark.catalog.listTables()\n","tblsReady = len(tblsInLK)\n","if tblsReady == 0:\n","    doesTableUMDExists = 0\n","else:\n","    tblsInLK = spark.createDataFrame(pd.DataFrame(tblsInLK))\n","    doesTableUMDExists = tblsInLK.filter(col('name') == 'UsageMetricsDatasets').count()"],"outputs":[],"execution_count":null,"metadata":{"jupyter":{"outputs_hidden":false,"source_hidden":false},"nteract":{"transient":{"deleting":false}}},"id":"9e462384-8998-41b0-9d5d-e636b454174a"},{"cell_type":"markdown","source":["## Workspace and Dataset IDs receiver"],"metadata":{"nteract":{"transient":{"deleting":false}}},"id":"203c6389-a299-436f-b7f5-71b14483ca0f"},{"cell_type":"code","source":["wsList = get_AccessibleWorkspaces()"],"outputs":[],"execution_count":null,"metadata":{"jupyter":{"outputs_hidden":false,"source_hidden":false},"nteract":{"transient":{"deleting":false}}},"id":"5d624671-ef8d-41ff-bb2b-823228d108f8"},{"cell_type":"code","source":["response_list = []\n","dsId = None\n","# wsNum = 0\n","if doesTableUMDExists == 0:\n","    for ws in wsList:\n","        # wsNum += 1\n","        wsId = ws['id']\n","        # print('Workspace ID:' + wsId)\n","        dsId = get_WorkspaceUsageMetricsId(wsId)\n","        # print('DONE WS Number: ' + str(wsNum))\n","        if dsId is not None:\n","            response_list.append({'WorkspaceId': wsId, 'DatasetId': dsId, 'ExtractionState': 0})\n","    usgDatasetIds = spark.createDataFrame(response_list)\n","    writetolake = usgDatasetIds.write.mode(\"overwrite\").format(\"delta\").save(\"Tables/UsageMetricsDatasets\")\n","else:\n","    sqlPrepare = spark.sql(\"SELECT * FROM UsageMetricsDatasets WHERE DatasetId IS NOT NULL\")\n","    # print('SQL Selection Completed')\n","    existingWsIds = sqlPrepare.select(col('WorkspaceId'))\n","    receivedListOfWs = spark.createDataFrame(wsList).select(col('id').alias(\"WorkspaceId\"))\n","    listOfUsableWs = existingWsIds.intersectAll(receivedListOfWs)\n","    # print('Intersection Completed')\n","    listOfNewWs = receivedListOfWs.exceptAll(existingWsIds)\n","    # print('Except Completed')\n","    extractionPrepare = \"('\"+\"','\".join(list(listOfUsableWs.toPandas()['WorkspaceId']))+\"')\"\n","    # print('List of Workspace IDs for SQL Created')\n","    existedRecords = spark.sql(f'SELECT * FROM UsageMetricsDatasets WHERE WorkspaceId IN {extractionPrepare} AND DatasetId IS NOT NULL')\n","    # print('Extraction of prepared dataset ID completed')\n","    collectionOfNewWs = listOfNewWs.collect()\n","    if listOfNewWs.count() != 0:\n","        for ws in collectionOfNewWs:\n","            # wsNum += 1\n","            wsId = ws['WorkspaceId']\n","            # print('Workspace ID:' + wsId)\n","            dsId = get_WorkspaceUsageMetricsId(wsId)\n","            # print('DONE WS Number: ' + str(wsNum))\n","            response_list.append({'WorkspaceId': wsId, 'DatasetId': dsId, 'ExtractionState': 0})\n","            newRecords = spark.createDataFrame(response_list)\n","        writetolake = newRecords.write.mode(\"append\").format(\"delta\").save(\"Tables/UsageMetricsDatasets\")\n","        usgDatasetIds = existedRecords.union(newRecords)\n","    else: \n","        usgDatasetIds = existedRecords"],"outputs":[],"execution_count":null,"metadata":{"collapsed":false,"jupyter":{"outputs_hidden":false,"source_hidden":false},"nteract":{"transient":{"deleting":false}}},"id":"9122f62e-573c-44f9-b742-5ff9aef8fba8"},{"cell_type":"markdown","source":["### Data Extraction"],"metadata":{"nteract":{"transient":{"deleting":false}}},"id":"8ab0505f-3f9b-4c96-815d-810a158940ae"},{"cell_type":"markdown","source":["#### Preparation"],"metadata":{"nteract":{"transient":{"deleting":false}}},"id":"ffe13ca3-906b-4a31-a25c-11806ad8345c"},{"cell_type":"code","source":["wsListUpdated = usgDatasetIds.collect()"],"outputs":[],"execution_count":null,"metadata":{"jupyter":{"outputs_hidden":false,"source_hidden":false},"nteract":{"transient":{"deleting":false}}},"id":"ed64a3a8-fb22-4cbb-9b91-4d095f90425c"},{"cell_type":"markdown","source":["#### Reports"],"metadata":{"nteract":{"transient":{"deleting":false}}},"id":"213568e1-eca8-4fb4-b1ed-a5ba2b3c9bb9"},{"cell_type":"code","source":["basicListOfTables = ['Reports','Users','Report pages','Workspace views'] # List of all basic tables from that need to be extracted\n","schema = None\n","for bsTbl in basicListOfTables:\n","    response_list = []\n","    for ws in wsListUpdated:\n","        dsId = ws[0]\n","        wsId = ws[2]\n","        dsExportState = ws[1]\n","        response = post_ExecuteQuery(wsId,dsId,f\"EVALUATE '{bsTbl}'\")\n","        size = len(response['rows'])\n","        if size != 0:\n","            response_list.append(response)\n","        time.sleep(0.5)\n","    df = extract_DataFrame(response_list,schema)\n","    df = replace_ColumnNames(df,bsTbl)\n","    updatedText = bsTbl.replace(\"'\",\"\").title().replace(\" \",\"\")\n","    if bsTbl == 'Users':\n","        df = df.distinct()    \n","    writetolake = df.write.mode(\"overwrite\").format(\"delta\").save(f\"Tables/{updatedText}\")"],"outputs":[],"execution_count":null,"metadata":{"collapsed":false,"jupyter":{"outputs_hidden":false,"source_hidden":false},"nteract":{"transient":{"deleting":false}}},"id":"f75b84ac-5f92-4c0f-b876-131fed147b2e"},{"cell_type":"code","source":["basicListOfTables = [\"Report page views\",\"Report load times\",\"Report views\"] # List of all basic tables from that need to be extracted\n","md = \"\"\n","dax = \"\"\n","for bsTbl in basicListOfTables:\n","    doesTableExists = 0\n","    findTable = \"\"\n","    updatedText = bsTbl.replace(\"'\",\"\").title().replace(\" \",\"\")\n","    if tblsReady == 0:\n","        doesTableExists = 0\n","    else:\n","        findTable = tblsInLK.filter(col('name') == 'updatedText')\n","        doesTableExists = findTable.count()\n","    response_list = []\n","    if bsTbl == \"Report page views\":\n","        schema = \\\n","            StructType([ \n","                StructField(f\"{bsTbl}[UserKey]\",StringType(),True), \\\n","                StructField(f\"{bsTbl}[UserId]\",StringType(),True), \\\n","                StructField(f\"{bsTbl}[TenantId]\",StringType(),True), \\\n","                StructField(f\"{bsTbl}[Timestamp]\",StringType(),True), \\\n","                StructField(f\"{bsTbl}[AppGuid]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[SectionId]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[AppName]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[Date]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[DeviceBrowserVersion]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[ReportId]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[SessionSource]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[OriginalReportId]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[WorkspaceId]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[Client]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[OriginalWorkspaceId]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[DeviceOSVersion]\", StringType(), True) \\\n","            ])\n","    elif bsTbl == \"Report load times\":\n","        schema = \\\n","            StructType([ \n","                StructField(f\"{bsTbl}[DeviceOSVersion]\",StringType(),True), \\\n","                StructField(f\"{bsTbl}[AppGuid]\",StringType(),True), \\\n","                StructField(f\"{bsTbl}[Timestamp]\",StringType(),True), \\\n","                StructField(f\"{bsTbl}[UserId]\",StringType(),True), \\\n","                StructField(f\"{bsTbl}[loadTime]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[Client]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[DeviceBrowserVersion]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[OriginalGroupId]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[TenantId]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[EndTime]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[SessionSource]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[AppName]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[Browser]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[OriginalReportId]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[GroupId]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[StartTime]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[Date]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[LocationCity]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[Country]\", StringType(), True) \\\n","            ])\n","    elif bsTbl == \"Report views\":\n","        schema = \\\n","            StructType([ \n","                StructField(f\"{bsTbl}[AppName]\",StringType(),True), \\\n","                StructField(f\"{bsTbl}[CapacityId]\",StringType(),True), \\\n","                StructField(f\"{bsTbl}[CapacityName]\",StringType(),True), \\\n","                StructField(f\"{bsTbl}[ConsumptionMethod]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[CreationTime]\",StringType(),True), \\\n","                StructField(f\"{bsTbl}[DatasetName]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[Date]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[DistributionMethod]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[OriginalConsumptionMethod]\",StringType(),True), \\\n","                StructField(f\"{bsTbl}[ReportId]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[ReportName]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[ReportType]\", StringType(), True),  \\\n","                StructField(f\"{bsTbl}[UserAgent]\", StringType(), True), \\\n","                StructField(f\"{bsTbl}[UserId]\",StringType(),True), \\\n","                StructField(f\"{bsTbl}[UserKey]\",StringType(),True) \\\n","            ])\n","    else:\n","        schema = None\n","    for ws in wsListUpdated:\n","        dsId = ws[0]\n","        wsId = ws[2]\n","        dsExportState = ws[1]\n","        if doesTableExists == 0:\n","            dax = f\"EVALUATE '{bsTbl}'\"\n","            md = \"overwrite\"\n","        else: \n","            if dsExportState == 0:\n","                dax = f\"EVALUATE '{bsTbl}'\"\n","                spark.sql(f\"UPDATE '{bsTbl}' SET ExtractionState = 1 WHERE WorkspaceId = '{wsId}' AND DatasetId = '{dsId}'\")\n","            else:\n","                tableLastDate = spark.sql(f\"SELECT MAX(Date) AS date FROM {updatedText}\").collect()[0]['date']\n","                fromDate = datetime.strptime(df, '%Y-%m-%dT%H:%M:%S').date().strftime('DATE(%Y,%m,%d)')\n","                dax = f\"DEFINE VAR _td = TODAY() VAR _from = {fromDate} EVALUATE FILTER('{bsTbl}',[Date]>_from && [Date]<_td)\"\n","            md = \"append\"\n","        response = post_ExecuteQuery(wsId,dsId,dax)\n","        size = len(response['rows'])\n","        if size != 0:\n","            response_list.append(response)\n","        time.sleep(0.5)\n","    # print('Data extracted')\n","    exports = extract_DataFrame(response_list,schema)\n","    exports = replace_ColumnNames(exports,bsTbl)\n","    writetolake = exports.write.mode(md).format(\"delta\").save(f\"Tables/{updatedText}\")\n","    # print('Data written')"],"outputs":[],"execution_count":null,"metadata":{"collapsed":false,"jupyter":{"outputs_hidden":false,"source_hidden":false},"nteract":{"transient":{"deleting":false}}},"id":"4ea53436-f21c-4f83-97ed-34a808d23ffe"}],"metadata":{"kernel_info":{"name":"synapse_pyspark"},"kernelspec":{"name":"synapse_pyspark","display_name":"Synapse PySpark"},"language_info":{"name":"python"},"microsoft":{"host":{"synapse_widget":{"state":{},"token":"4a7187e1-57aa-4d2b-8f17-cfff511e7658"}},"language":"python","ms_spell_check":{"ms_spell_check_language":"en"}},"nteract":{"version":"nteract-front-end@1.0.0"},"widgets":{},"save_output":true,"spark_compute":{"compute_id":"/trident/default","session_options":{"conf":{},"enableDebugMode":false}},"notebook_environment":{},"synapse_widget":{"version":"0.1","state":{}},"trident":{"lakehouse":{}}},"nbformat":4,"nbformat_minor":5}