Kafka→⚙️→S3 (JSON).ipynb

{
  "metadata": {
    "name": "Kafka→⚙️→S3 (JSON)",
    "kernelspec": {
      "language": "scala",
      "name": "spark2-scala",
      "display_name": "spark2-scala"
    },
    "language_info": {
      "codemirror_mode": "text/x-scala",
      "file_extension": ".scala",
      "mimetype": "text/x-scala",
      "name": "scala",
      "pygments_lexer": "scala"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 2,
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# Streaming data from Kafka to S3 using Amazon MSF\n\n`@rmoff / 2024-06-20`"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Define the Kafka source table"
    },
    {
      "cell_type": "code",
      "execution_count": 1,
      "metadata": {
        "autoscroll": "auto"
      },
      "outputs": [],
      "source": "%flink.ssql\nCREATE TABLE basket01 (\n             customerId STRING,\n              customerName STRING,\n              customerAddress STRING, \n              storeId STRING,\n              storeName STRING,\n              storeLocation STRING,\n              products ARRAY\u003cROW\u003c\n                productName STRING,\n                quantity DOUBLE,\n                unitPrice DOUBLE, \n                category STRING\n              \u003e\u003e,\n              `timestamp` STRING\n ) WITH (\n   \u0027connector\u0027 \u003d \u0027kafka\u0027,\n   \u0027topic\u0027 \u003d \u0027supermarketBaskets\u0027,\n   \u0027properties.bootstrap.servers\u0027 \u003d \u00272.tcp.eu.ngrok.io:19693\u0027,\n   \u0027properties.group.id\u0027 \u003d \u0027msf00\u0027,\n   \u0027scan.startup.mode\u0027 \u003d \u0027earliest-offset\u0027,\n   \u0027format\u0027 \u003d \u0027json\u0027\n );"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "_Use this if you need to change properties on the source, such as the the Kafka broker_"
    },
    {
      "cell_type": "code",
      "execution_count": 3,
      "metadata": {
        "autoscroll": "auto"
      },
      "outputs": [],
      "source": "%flink.ssql\nALTER TABLE basket01 SET ( \u0027properties.bootstrap.servers\u0027 \u003d \u00272.tcp.eu.ngrok.io:19693\u0027)"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Smoke test the connection and sample the data from the Kafka topic"
    },
    {
      "cell_type": "code",
      "execution_count": 5,
      "metadata": {
        "autoscroll": "auto"
      },
      "outputs": [],
      "source": "%flink.ssql\nSELECT * FROM basket01 LIMIT 5;"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Create a `VIEW` to filter the data"
    },
    {
      "cell_type": "code",
      "execution_count": 7,
      "metadata": {
        "autoscroll": "auto"
      },
      "outputs": [],
      "source": "%flink.ssql\nCREATE VIEW customers_b AS \nSELECT customerId, customerName, customerAddress\nFROM basket01\nWHERE LOWER(customerName) LIKE \u0027b%\u0027;"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Check the view is working, sample the data"
    },
    {
      "cell_type": "code",
      "execution_count": 9,
      "metadata": {
        "autoscroll": "auto"
      },
      "outputs": [],
      "source": "%flink.ssql\nSELECT * FROM customers_b LIMIT 5;\n"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Define a sink using JSON files on S3"
    },
    {
      "cell_type": "code",
      "execution_count": 11,
      "metadata": {
        "autoscroll": "auto"
      },
      "outputs": [],
      "source": "%flink.ssql\nCREATE TABLE s3_customers_b_json_01 (\n     customerId STRING,\n     customerName STRING,\n     customerAddress STRING\n) WITH (\n     \u0027connector\u0027 \u003d \u0027filesystem\u0027,\n     \u0027path\u0027      \u003d \u0027s3://rmoff/msf-test/customers_b-json\u0027,\n     \u0027format\u0027    \u003d \u0027json\u0027,\n     \u0027sink.rolling-policy.rollover-interval\u0027 \u003d \u002730sec\u0027\n);"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Stream data from Kafka, via the `VIEW`, to the S3 sink"
    },
    {
      "cell_type": "code",
      "execution_count": 13,
      "metadata": {
        "autoscroll": "auto"
      },
      "outputs": [],
      "source": "%flink.ssql\n\nSET \u0027execution.checkpointing.interval\u0027 \u003d \u002710sec\u0027;\n\nINSERT INTO s3_customers_b_json_01\n    SELECT * FROM customers_b;"
    }
  ]
}