01_standalone_examples/aws-glue-trino/trino-glue-demo-notebook.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "c0941866-5ab1-46c6-ae4e-3ae228311623",
   "metadata": {},
   "source": [
    "<img src=\"https://lakefs.io/wp-content/uploads/2022/09/lakeFS-Logo.svg\" alt=\"lakeFS logo\" width=200/>\n",
    "\n",
    "# Integration of lakeFS with Trino and Glue Catalog\n",
    "\n",
    "[📚 Docs](https://docs.lakefs.io/integrations/presto_trino.html)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "29105bda-b8b6-4bff-aaa8-7eadbacc7923",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Use Case: Isolated Dev/Test/ETL Environments"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "531dde95-2a5b-4625-becf-55aaf9c496ec",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Config"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cc03351c-5072-4b43-9dc1-28dd46a74d6c",
   "metadata": {},
   "source": [
    "### Spark Configuration: Change lakeFS endpoint, Access and Secret Key"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d7be4548-2642-4d54-a48a-729a32c9a609",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%configure -f\n",
    "{\n",
    "    \"conf\": {\n",
    "        \"spark.hadoop.fs.s3a.endpoint\": \"<lakeFS Endpoint URL>\",\n",
    "        \"spark.hadoop.fs.s3a.access.key\": \"<lakeFS Access Key>\",\n",
    "        \"spark.hadoop.fs.s3a.secret.key\": \"<lakeFS Secret Key>\",\n",
    "        \"spark.hadoop.fs.s3.impl\": \"org.apache.hadoop.fs.s3a.S3AFileSystem\",\n",
    "        \"spark.hadoop.fs.s3a.path.style.access\": \"true\",\n",
    "        \"spark.pyspark.virtualenv.enabled\": \"true\",\n",
    "        \"spark.pyspark.virtualenv.type\":\"native\",\n",
    "        \"spark.pyspark.virtualenv.bin.path\":\"/usr/bin/virtualenv\"\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "61c5aea5-5195-4caa-a535-95b9bab8a1be",
   "metadata": {},
   "source": [
    "### lakeFS endpoint and credentials"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "79a78d9a-10b0-4ffa-b501-213876abe1ad",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "lakefsEndPoint = '<lakeFS Endpoint URL>' # e.g. 'https://username.aws_region_name.lakefscloud.io' \n",
    "lakefsAccessKey = '<lakeFS Access Key>'\n",
    "lakefsSecretKey = '<lakeFS Secret Key>'"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "31ca7f50-bd3d-4a86-a4ed-30066b42c971",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Storage Information\n",
    "\n",
    "If you're not using sample repo lakeFS, then change the Storage Namespace to a location in the bucket you’ve configured. The storage namespace is a location in the underlying storage where data for this repository will be stored."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f6a5b298-1879-4441-a607-0e64289bf90e",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "storageNamespace = 's3://<Bucket Name>' # e.g. 's3://bucket'"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "98c3b1bb-ee40-4147-b1a9-15ad25610ed2",
   "metadata": {},
   "source": [
    "## Install and configure lakectl on your computer (lakeFS command-line tool): https://docs.lakefs.io/reference/cli.html"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "853dd30d-ec0e-4302-9dd2-97c4fa21257e",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7e8191ce-2e9a-4eda-a107-ae99659a5397",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Setup\n",
    "\n",
    "**(you shouldn't need to change anything in this section, just run it)**"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c8ae71c4-6ce8-461a-ae83-e5a99556e673",
   "metadata": {},
   "source": [
    "### Glue database name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "75d47b7d-5d60-4942-9496-d97cf92311b5",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "glueDatabaseName = \"trino_glue_demo\" # This notebook will create this database"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "df42fc28-ea54-4775-9772-d965748f3fd8",
   "metadata": {},
   "source": [
    "### lakeFS repository name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a45fbab3-5f98-46ad-8e12-9021ed2bf81a",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "repo_name = \"trino-glue-demo\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9e11b9bb-aee4-4dfd-aa33-55c9ec6a2939",
   "metadata": {},
   "source": [
    "### Versioning Information"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "96f32e07-19b3-4107-a24a-90761eda18d2",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "mainBranch = \"main\"\n",
    "etlBranch = \"etl_branch\"\n",
    "customersTable = \"customers\"\n",
    "ordersTable = \"orders\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4f5f1d89-a7a0-409c-8241-593370cf8d42",
   "metadata": {},
   "source": [
    "### Install Python libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5ebb1630-7640-4ed2-b5f9-b33107a9067e",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "sc.install_pypi_package(\"urllib3==1.25.3\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2dfb3609-8add-4e7c-b775-89871d204728",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "sc.install_pypi_package(\"lakefs==0.6.0\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "66b4e853-e770-4625-ac3f-94ac96cb1558",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "sc.install_pypi_package(\"pyhive\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b2e753cb-a9a5-472c-b3b8-0f311bf28783",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "sc.install_pypi_package(\"requests\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "947f6158-6eb1-4842-925c-62e791e4fd4e",
   "metadata": {},
   "source": [
    "### Import Python libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "748aea5b-b33e-4c97-8db0-e1cee20a8bc3",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import lakefs\n",
    "import os\n",
    "from pyspark.sql.types import ByteType, IntegerType, LongType, StringType, StructType, StructField\n",
    "from pyspark.sql.functions import *\n",
    "from pyhive import trino\n",
    "import requests"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "da23fe23-042a-4c86-95e2-3aa92ac92d79",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "os.environ[\"LAKECTL_SERVER_ENDPOINT_URL\"] = lakefsEndPoint\n",
    "os.environ[\"LAKECTL_CREDENTIALS_ACCESS_KEY_ID\"] = lakefsAccessKey\n",
    "os.environ[\"LAKECTL_CREDENTIALS_SECRET_ACCESS_KEY\"] = lakefsSecretKey"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "eaa5bb3f-7432-4637-bc8e-68159bdcb969",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Create lakeFSClient"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ed8c6074-b9d9-41d3-b27a-767148d50393",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "print(\"Verifying lakeFS credentials…\")\n",
    "try:\n",
    "    v=lakefs.client.Client().version\n",
    "except:\n",
    "    print(\"Failed to get lakeFS version\")\n",
    "else:\n",
    "    print(f\"lakeFS credentials verified\\n\\nlakeFS version {v}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e1c3ca40-a690-4885-90f9-fc7c069c66c5",
   "metadata": {},
   "source": [
    "### Define lakeFS Repository"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "163ecc3f-eaa9-478a-b929-9fd5ffac385b",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "repo = lakefs.Repository(repo_name).create(storage_namespace=f\"{storageNamespace}/{repo_name}\", default_branch=mainBranch, exist_ok=True)\n",
    "branchMain = repo.branch(mainBranch)\n",
    "print(repo)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "721e4d01-6d8e-4e8b-a4e5-242a5df1957a",
   "metadata": {},
   "source": [
    "### Connect to Trino using `PyHive`\n",
    "\n",
    "You will now connect to Trino using `PyHive` library. You might have to replace the values for `hostName, userName, schemaName and catalogName` as applicable to your environment. The port is set to EMR default of 8889."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "65514242-d74e-4d9a-bd49-3e4b13a9a0c4",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "hostName = \"127.0.0.1\"\n",
    "userName = \"hadoop\"\n",
    "schemaName = \"default\"\n",
    "catalogName = \"glue\"\n",
    "trinoPort = 8889\n",
    "\n",
    "headers = {\n",
    "    'X-Trino-User': userName,\n",
    "    'X-Trino-Schema': schemaName,\n",
    "    'X-Trino-Catalog': catalogName\n",
    "}\n",
    "\n",
    "trinoSession = requests.Session()\n",
    "trinoSession.headers.update(headers)\n",
    "\n",
    "conn = trino.connect(requests_session=trinoSession,\n",
    "                     host=hostName,\n",
    "                     port=trinoPort\n",
    "                    )"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "29febd3f-bfd1-419a-b871-d3430c353b1a",
   "metadata": {},
   "source": [
    "### Define some helper functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c9859b09-24d3-43ff-92ec-65c4604dccf9",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def execute_trino_query(query):\n",
    "    cur = conn.cursor()\n",
    "    cur.execute(query)\n",
    "    result = cur.fetchall()\n",
    "\n",
    "    return result\n",
    "\n",
    "def print_commit(log):\n",
    "    from datetime import datetime\n",
    "    from pprint import pprint\n",
    "\n",
    "    print('Message:', log.message)\n",
    "    print('ID:', log.id)\n",
    "    print('Committer:', log.committer)\n",
    "    print('Creation Date:', datetime.utcfromtimestamp(log.creation_date).strftime('%Y-%m-%d %H:%M:%S'))\n",
    "    print('Parents:', log.parents)\n",
    "    print('Metadata:')\n",
    "    pprint(log.metadata)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5b8149f3-e506-44e5-9a4d-424c3167f838",
   "metadata": {},
   "source": [
    "### Create Glue Database"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d84f3603-7888-41ff-8ad3-ce4b86d7176a",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "execute_trino_query(f\"CREATE SCHEMA {glueDatabaseName} WITH (location = 's3a://{repo_name}/{mainBranch}')\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2377bae1-0cc2-4334-b36c-41fd17c93eba",
   "metadata": {},
   "source": [
    "### Define CUSTOMER.csv data file schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0e42f399-19b7-4def-9988-916a7f8ba067",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "customersSchema = StructType([\n",
    "  StructField(\"Customer_ID\", IntegerType(), False),\n",
    "  StructField(\"Country\", StringType(), False),\n",
    "  StructField(\"Gender\", StringType(), False),\n",
    "  StructField(\"Personal_ID\", IntegerType(), True),\n",
    "  StructField(\"Customer_Name\", StringType(), False),\n",
    "  StructField(\"Customer_FirstName\", StringType(), False),\n",
    "  StructField(\"Customer_LastName\", StringType(), False),\n",
    "  StructField(\"Birth_Date\", StringType(), False),\n",
    "  StructField(\"Customer_Address\", StringType(), False),\n",
    "  StructField(\"Street_ID\", LongType(), False),\n",
    "  StructField(\"Street_Number\", IntegerType(), False),\n",
    "  StructField(\"Customer_Type_ID\", IntegerType(), False)\n",
    "])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "881f8291-2243-4dd3-b907-f2c66ea20563",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "customersSchemaForGlue = \"Customer_ID int, \\\n",
    "  Country varchar, \\\n",
    "  Gender varchar, \\\n",
    "  Personal_ID int, \\\n",
    "  Customer_Name varchar, \\\n",
    "  Customer_FirstName varchar, \\\n",
    "  Customer_LastName varchar, \\\n",
    "  Birth_Date varchar, \\\n",
    "  Customer_Address varchar, \\\n",
    "  Street_ID bigint, \\\n",
    "  Street_Number int, \\\n",
    "  Customer_Type_ID int\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cd6c69ac-32a2-4efa-bcc5-8bf5aa8fe867",
   "metadata": {},
   "source": [
    "### Define ORDER_FACT.csv data file schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6076bbcf-3bb2-4261-944a-0796304dea89",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ordersSchema = StructType([\n",
    "  StructField(\"Customer_ID\", IntegerType(), False),\n",
    "  StructField(\"Employee_ID\", IntegerType(), False),\n",
    "  StructField(\"Street_ID\", LongType(), False),\n",
    "  StructField(\"Order_Date\", StringType(), False),\n",
    "  StructField(\"Delivery_Date\", StringType(), False),\n",
    "  StructField(\"Order_ID\", LongType(), False),\n",
    "  StructField(\"Order_Type\", ByteType(), False),\n",
    "  StructField(\"Product_ID\", LongType(), False),\n",
    "  StructField(\"Quantity\", ByteType(), False),\n",
    "  StructField(\"Total_Retail_Price\", StringType(), False),\n",
    "  StructField(\"CostPrice_Per_Unit\", StringType(), False),\n",
    "  StructField(\"Discount\", LongType(), True)\n",
    "])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "38cefab1-6b14-4a5f-9695-4c54ed13280a",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ordersSchemaForGlue = \"Customer_ID int, \\\n",
    "  Employee_ID int, \\\n",
    "  Street_ID bigint, \\\n",
    "  Order_Date varchar, \\\n",
    "  Delivery_Date varchar, \\\n",
    "  Order_ID bigint, \\\n",
    "  Order_Type int, \\\n",
    "  Product_ID bigint, \\\n",
    "  Quantity int, \\\n",
    "  Total_Retail_Price varchar, \\\n",
    "  CostPrice_Per_Unit varchar, \\\n",
    "  Discount bigint\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "25ea2edb-6c7e-46b9-9aea-fa15952cf3cb",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "406d9862-c34a-4281-80dc-32d15dc446b3",
   "metadata": {},
   "source": [
    "# Main demo starts here 🚦 👇🏻"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cbc87fa4-3d84-4c13-a612-b2f14d15d4e6",
   "metadata": {},
   "source": [
    "For this demo - we'll be utilizing a dataset - [Orion Star - Sports and outdoors RDBMS dataset](https://www.kaggle.com/datasets/chethanp11/orion-star-sports-and-outdoors-rdbms-dataset) from [Kaggle](https://www.kaggle.com/)."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d70fb771-6b77-4905-924e-439329b03716",
   "metadata": {},
   "source": [
    "## Run following command on your computer to clone lakeFS samples repo along with sample data used by this notebook:"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "93deb103-a2df-4a0e-85c7-9c6570d84b26",
   "metadata": {},
   "source": [
    "### git clone https://github.com/treeverse/lakeFS-samples.git"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d3979d6e-de31-43d0-b12d-b9b33580cb64",
   "metadata": {},
   "source": [
    "## Print the command and run it on your computer to upload sample data to lakeFS repository"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fffbd19b-38c6-44e3-ae7e-e28fcd86bdeb",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "print(f\"cd lakeFS-samples && lakectl fs upload -s ./data/OrionStar lakefs://{repo_name}/main/data/OrionStar --recursive && lakectl commit lakefs://{repo_name}/main -m 'Uploaded sample data'\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b1577257-81b2-4971-bafb-5aabfd1e603e",
   "metadata": {},
   "source": [
    "## Create Customers table in the main branch (using [CUSTOMER.csv](https://github.com/treeverse/lakeFS-samples/blob/040ce6fd2a2f45bd991dd17c8e9ad1d88887cdae/data/OrionStar/CUSTOMER.csv) file)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c944ce0b-10ea-4990-9f02-a92354c9849c",
   "metadata": {},
   "source": [
    "#### Register table in Glue catalog"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "be4d39f9-4ef8-4d25-8885-d2a661a027e8",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "customersTablePath = f\"s3a://{repo_name}/{mainBranch}/{customersTable}\"\n",
    "\n",
    "execute_trino_query(f\" \\\n",
    "          CREATE TABLE IF NOT EXISTS {glueDatabaseName}.{customersTable}( \\\n",
    "              {customersSchemaForGlue} \\\n",
    "          ) \\\n",
    "          WITH ( \\\n",
    "              format = 'Parquet', \\\n",
    "              external_location = '{customersTablePath}' \\\n",
    "              ) \\\n",
    "          \")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "be593469-8b0a-4a66-ac45-0d77c74808e4",
   "metadata": {},
   "source": [
    "#### Read CSV file and write data to Customers table in the main branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b3eecdfe-ecf0-4947-9feb-acfae738842f",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "df = spark.read.csv(f\"s3a://{repo_name}/{mainBranch}/data/OrionStar/CUSTOMER.csv\",header=True,schema=customersSchema)\n",
    "df.write.format(\"parquet\").mode(\"append\").save(f\"{customersTablePath}\")\n",
    "df.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "68e404d3-982f-48d3-a96c-b8bf566dab68",
   "metadata": {},
   "source": [
    "## Create Orders table in the main branch (using [ORDER_FACT.csv](https://github.com/treeverse/lakeFS-samples/blob/040ce6fd2a2f45bd991dd17c8e9ad1d88887cdae/data/OrionStar/ORDER_FACT.csv) file)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "413169c4-9335-45b8-912f-6ec050fb541c",
   "metadata": {},
   "source": [
    "#### Register table in Glue catalog"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b37c0a12-b00b-4f1f-9114-eed7500840ca",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ordersTablePath = f\"s3a://{repo_name}/{mainBranch}/{ordersTable}\"\n",
    "\n",
    "execute_trino_query(f\" \\\n",
    "          CREATE TABLE IF NOT EXISTS {glueDatabaseName}.{ordersTable}( \\\n",
    "              {ordersSchemaForGlue} \\\n",
    "          ) \\\n",
    "          WITH ( \\\n",
    "              format = 'Parquet', \\\n",
    "              external_location = '{ordersTablePath}' \\\n",
    "              ) \\\n",
    "          \")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f29db171-c720-47b8-9f85-dfde73318df1",
   "metadata": {},
   "source": [
    "#### Read CSV file and write to Orders table in the main branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "172e822c-efb9-4879-b8a6-dd6879f6cf23",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "df = spark.read.csv(f\"s3a://{repo_name}/{mainBranch}/data/OrionStar/ORDER_FACT.csv\",header=True,schema=ordersSchema)\n",
    "df.write.format(\"parquet\").mode(\"append\").save(f\"{ordersTablePath}\")\n",
    "df.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b763c2a9-561c-447c-ae2b-0b8fa13fedb0",
   "metadata": {},
   "source": [
    "## Commit changes and attach some metadata"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ef6c6bef-9159-4dec-a515-00820317d07d",
   "metadata": {},
   "outputs": [],
   "source": [
    "ref = branchMain.commit(message='Added customers and orders tables!', \n",
    "        metadata={'using': 'python_api'})\n",
    "print_commit(ref.get_commit())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f1ad229b-51da-4c1b-82d7-d30916a287ce",
   "metadata": {},
   "source": [
    "## Execute Trino queries to read the data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "04333d5e-ffa6-4b01-bd16-8fa98cff06be",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "spark.createDataFrame(execute_trino_query(f'SELECT * FROM \"{glueDatabaseName}\".\"{customersTable}\"'), schema=customersSchema).show(10) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "26cbe1d9-de18-4742-b177-dea6854eca7b",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "spark.createDataFrame(execute_trino_query(f'SELECT * FROM \"{glueDatabaseName}\".\"{ordersTable}\"'), schema=ordersSchema).show(10) "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "63ec287f-b340-43be-983f-fad43b024ac5",
   "metadata": {},
   "source": [
    "# 🟢 ETL Job Starts"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5fa6a1a9-70c2-4416-b860-97aa1c2a6dfa",
   "metadata": {},
   "source": [
    "## Create an ETL branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b955a4f9-df67-49c0-b7a9-d84d04bd9b8d",
   "metadata": {},
   "outputs": [],
   "source": [
    "branchETL = repo.branch(etlBranch).create(source_reference=mainBranch, exist_ok=True)\n",
    "print(f\"{etlBranch} ref:\", branchETL.get_commit().id)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fcf3ef86-c103-4de9-b04c-b6ac61751e4e",
   "metadata": {},
   "source": [
    "### Create Glue Database for the ETL branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "93cd2529-05dc-48f9-83c4-60a3ede080e5",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "execute_trino_query(f\"CREATE SCHEMA {glueDatabaseName}_{etlBranch} WITH (location = 's3a://{repo_name}/{etlBranch}')\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4a03cb9f-a949-460a-b47d-ebd2ff10f068",
   "metadata": {},
   "source": [
    "### Register tables in Glue catalog for the ETL branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "90609f1b-a14c-4e3a-ae9b-c27214a92179",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "customersTablePathETLBranch = f\"s3a://{repo_name}/{etlBranch}/{customersTable}\"\n",
    "\n",
    "execute_trino_query(f\" \\\n",
    "          CREATE TABLE IF NOT EXISTS {glueDatabaseName}_{etlBranch}.{customersTable}( \\\n",
    "              {customersSchemaForGlue} \\\n",
    "          ) \\\n",
    "          WITH ( \\\n",
    "              format = 'Parquet', \\\n",
    "              external_location = '{customersTablePathETLBranch}' \\\n",
    "              ) \\\n",
    "          \")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "71ea99a0-aa0d-4ed2-951d-e06a07e40b72",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ordersTablePathETLBranch = f\"s3a://{repo_name}/{etlBranch}/{ordersTable}\"\n",
    "\n",
    "execute_trino_query(f\" \\\n",
    "          CREATE TABLE IF NOT EXISTS {glueDatabaseName}_{etlBranch}.{ordersTable}( \\\n",
    "              {ordersSchemaForGlue} \\\n",
    "          ) \\\n",
    "          WITH ( \\\n",
    "              format = 'Parquet', \\\n",
    "              external_location = '{ordersTablePathETLBranch}' \\\n",
    "              ) \\\n",
    "          \")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "45794ce5-9cdd-49c5-b766-69819638100d",
   "metadata": {},
   "source": [
    "## Execute Trino query to insert Customers data in the ETL branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "439e3363-b60f-4ee4-b9b9-744e77fc2a03",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "execute_trino_query(f\"INSERT INTO {glueDatabaseName}_{etlBranch}.{customersTable} VALUES (1,'US','M',2,'Scott Gibbs','Scott','Gibbs','12APR1970','556 Greywood Rd',9260103713,1068,1030)\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f7985acd-b71f-478d-899d-365452296727",
   "metadata": {},
   "source": [
    "## Execute Trino query to read Customers data from the ETL branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c5f82320-6820-45e3-b226-6a596969e681",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "spark.createDataFrame(execute_trino_query(f'SELECT * FROM \"{glueDatabaseName}_{etlBranch}\".\"{customersTable}\" ORDER BY Customer_ID'), schema=customersSchema).show(10) "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "640a64bd-7900-4498-a20e-af9c297b9801",
   "metadata": {},
   "source": [
    "## Execute Trino query to read Customers data from the main branch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8ce9b1f1-259c-4e55-9351-53d1ff4ddd2a",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "spark.createDataFrame(execute_trino_query(f'SELECT * FROM \"{glueDatabaseName}\".\"{customersTable}\" ORDER BY Customer_ID'), schema=customersSchema).show(10) "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "557472bd-39ed-43d3-a00a-ed69ed8ab280",
   "metadata": {},
   "source": [
    "## More Questions?\n",
    "\n",
    "###### Join the lakeFS Slack group - https://lakefs.io/slack"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "PySpark",
   "language": "python",
   "name": "pysparkkernel"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "python",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "pyspark",
   "pygments_lexer": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}