docs-samples/real-time-analytics/NYC_GreenTaxi_KQL_notebook.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "source": [
        "# Use a notebook with Apache Spark to query a KQL database\n",
        "\n",
        "Documentation for use of this notebook can be found here: [Use a notebook with Apache Spark to query a KQL database](https://learn.microsoft.com/fabric/real-time-analytics/spark-connector)"
      ],
      "metadata": {
        "id": "YErMT7GYtf-6"
      }
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "cellStatus": "{\"Ramachandran G\":{\"queued_time\":\"2023-03-23T08:37:28.0477136Z\",\"session_start_time\":\"2023-03-23T08:37:28.3245362Z\",\"execution_start_time\":\"2023-03-23T08:37:41.7858145Z\",\"execution_finish_time\":\"2023-03-23T08:38:02.2179974Z\",\"state\":\"finished\",\"livy_statement_state\":\"available\"}}",
        "id": "rCaON4Aus0ox"
      },
      "outputs": [],
      "source": [
        "# Azure storage access info\n",
        "blob_account_name = \"azureopendatastorage\"\n",
        "blob_container_name = \"nyctlc\"\n",
        "blob_relative_path = \"green\"\n",
        "blob_sas_token = r\"\"\n",
        "\n",
        "# Allow SPARK to read from Blob remotely\n",
        "wasbs_path = 'wasbs://%s@%s.blob.core.windows.net/%s' % (blob_container_name, blob_account_name, blob_relative_path)\n",
        "spark.conf.set(\n",
        "  'fs.azure.sas.%s.%s.blob.core.windows.net' % (blob_container_name, blob_account_name),\n",
        "  blob_sas_token)\n",
        "print('Remote blob path: ' + wasbs_path)\n",
        "\n",
        "# SPARK read parquet, note that it won't load any data yet by now\n",
        "df = spark.read.parquet(wasbs_path)\n",
        "# Display top 10 rows\n",
        "print('Displaying top 10 rows: ')\n",
        "df.printSchema"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "cellStatus": "{\"Ramachandran G\":{\"queued_time\":\"2023-03-23T08:40:42.2507992Z\",\"session_start_time\":null,\"execution_start_time\":\"2023-03-23T08:40:42.7470834Z\",\"execution_finish_time\":\"2023-03-23T08:40:43.1656264Z\",\"state\":\"finished\",\"livy_statement_state\":\"available\"}}",
        "jupyter": {
          "outputs_hidden": false,
          "source_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        },
        "id": "8GE17J68s0o0"
      },
      "outputs": [],
      "source": [
        "#The target where this data will be written to\n",
        "kustoUri = \"https://<enter-your-URI-here>.kusto.data.microsoft.com\"\n",
        "database=\"<enter-your-database-name>\"\n",
        "table=\"<enter-your-table-name>\" #for example, GreenTaxiData"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "cellStatus": "{\"Ramachandran G\":{\"queued_time\":\"2023-03-23T08:40:46.5885086Z\",\"session_start_time\":null,\"execution_start_time\":\"2023-03-23T08:40:46.965185Z\",\"execution_finish_time\":\"2023-03-23T08:49:19.0524624Z\",\"state\":\"finished\",\"livy_statement_state\":\"available\"}}",
        "jupyter": {
          "outputs_hidden": false,
          "source_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        },
        "id": "AOE-XRA-s0o1"
      },
      "outputs": [],
      "source": [
        "#This is an example of writing data to Kusto. The source data is read as a blob into a dataframe from Azure Open Data for GreenTaxi / Limousines in NYC.\n",
        "#The access token is created using the user's credential and will be used to write the data to the Kusto table GreenTaxiData, therefore the user is required \n",
        "#for 'user' privileges or above on the target database and table 'admin' privileges if the table already exists. If the table does not exist, \n",
        "#it will be created with the DataFrame schema.\n",
        "df.write.format(\"com.microsoft.kusto.spark.synapse.datasource\").\\\n",
        "option(\"kustoCluster\",kustoUri).\\\n",
        "option(\"kustoDatabase\",database).\\\n",
        "option(\"kustoTable\", table).\\\n",
        "option(\"accessToken\", mssparkutils.credentials.getToken(kustoUri)).\\\n",
        "option(\"tableCreateOptions\", \"CreateIfNotExist\").mode(\"Append\").save()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "cellStatus": "{\"Ramachandran G\":{\"queued_time\":\"2023-03-23T08:49:29.7202163Z\",\"session_start_time\":null,\"execution_start_time\":\"2023-03-23T08:49:30.1748177Z\",\"execution_finish_time\":\"2023-03-23T08:49:30.6119991Z\",\"state\":\"finished\",\"livy_statement_state\":\"available\"}}",
        "jupyter": {
          "outputs_hidden": false,
          "source_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        },
        "id": "nWyGf7rzs0o2"
      },
      "outputs": [],
      "source": [
        "#This is an example of Reading data from the KQL Database. Here the query retrieves the max,min fares and distances that the taxi recorded every month from the years 2014 to 2020\n",
        "kustoQuery = \"GreenTaxiData |  where puYear between (2014 .. 2020 ) | summarize  MaxDistance=max(tripDistance) , MaxFare = max(fareAmount) ,MinDistance=min(tripDistance) , MinFare = min(fareAmount) by puYear,puMonth | order by puYear,puMonth desc\"\n",
        "kustoDf  = spark.read\\\n",
        "            .format(\"com.microsoft.kusto.spark.synapse.datasource\")\\\n",
        "            .option(\"accessToken\", mssparkutils.credentials.getToken(kustoUri))\\\n",
        "            .option(\"kustoCluster\", kustoUri)\\\n",
        "            .option(\"kustoDatabase\", database) \\\n",
        "            .option(\"kustoQuery\", kustoQuery).load()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "cellStatus": "{\"Ramachandran G\":{\"queued_time\":\"2023-03-23T08:49:36.5443089Z\",\"session_start_time\":null,\"execution_start_time\":\"2023-03-23T08:49:37.0371176Z\",\"execution_finish_time\":\"2023-03-23T08:49:42.6677216Z\",\"state\":\"finished\",\"livy_statement_state\":\"available\"}}",
        "jupyter": {
          "outputs_hidden": false,
          "source_hidden": false
        },
        "nteract": {
          "transient": {
            "deleting": false
          }
        },
        "id": "F77v__INs0o3"
      },
      "outputs": [],
      "source": [
        "kustoDf.show()"
      ]
    }
  ],
  "metadata": {
    "description": "",
    "kernel_info": {
      "name": "synapse_pyspark"
    },
    "kernelspec": {
      "display_name": "Synapse PySpark",
      "language": "Python",
      "name": "synapse_pyspark"
    },
    "language_info": {
      "name": "python"
    },
    "notebook_environment": {},
    "save_output": true,
    "spark_compute": {
      "compute_id": "/trident/default",
      "session_options": {
        "conf": {
          "spark.livy.synapse.ipythonInterpreter.enabled": "true"
        },
        "enableDebugMode": false,
        "keepAliveTimeout": 30
      }
    },
    "synapse_widget": {
      "state": {},
      "version": "0.1"
    },
    "trident": {
      "lakehouse": {}
    },
    "colab": {
      "provenance": []
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}