kafka-summit.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# We have a repository with all the examples so you can play along at home:\n",
    "\n",
    "### https://github.com/RedPillAnalytics/kafka-examples"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Starting our Docker Environment\n",
    "### We even use Gradle to manage Docker using `com.avast.gradle.docker-compose` plugin:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!./gradlew --console=plain tasks --group docker -q\n",
    "!./gradlew --console=plain -q clean composeUp"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Using Python to interact with KSQL. Let's see if `CLICKSTREAM`, `CLICKSTREAM_CODES` and `CLIICKSTREAM_USERS` are there:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ksql = \"/usr/local/bin/ksql\"\n",
    "!echo \"LIST TOPICS;\" | \"$ksql\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# The KSQL Developer Experience\n",
    "### Create a *registration* stream `CLICKSTREAM`. This is sort of like DDL in relational DBs:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = (\n",
    "    \"CREATE STREAM clickstream \"\n",
    "    \"(_time bigint, time varchar, ip varchar, request varchar, status int, userid int, bytes bigint, agent varchar) \"\n",
    "    \"with (kafka_topic = 'clickstream', value_format = 'json');\"\n",
    ")\n",
    "!echo \"$sql\" | \"$ksql\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create *persistent query* table `EVENTS_PER_MIN`, which reads data from `CLICKSTREAM` and initializes the streaming process:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = (\n",
    "    \"CREATE TABLE events_per_min \"\n",
    "    \"AS SELECT userid, count(*) AS events \"\n",
    "    \"FROM clickstream window \"\n",
    "    \"TUMBLING (size 60 second) GROUP BY userid;\"\n",
    ")\n",
    "!echo \"$sql\" | \"$ksql\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Let's list our topics:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!echo \"LIST TOPICS;\" | \"$ksql\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Suppose we want to make changes to `CLICKSTREAM`. Just drop it right?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = (\n",
    "    \"DROP STREAM clickstream;\"\n",
    ")\n",
    "!echo \"$sql\" | \"$ksql\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Bummer. First we need to terminate the persistent query... but the *query id* is auto-incrementing, and not constant:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = (\n",
    "    \"TERMINATE QUERY CTAS_EVENTS_PER_MIN_0; \"\n",
    ")\n",
    "!echo \"$sql\" | \"$ksql\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Now drop CLICKSTREAM again"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = (\n",
    "    \"DROP STREAM clickstream; \"\n",
    ")\n",
    "!echo \"$sql\" | \"$ksql\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### We can do something better to increase developer efficiency!\n",
    "\n",
    "# The `gradle-confluent` Plugin\n",
    "\n",
    "### [GitHub Repository](https://github.com/RedPillAnalytics/gradle-confluent)\n",
    "\n",
    "### [Gradle Plugin Portal](https://plugins.gradle.org/plugin/com.redpillanalytics.gradle-confluent)\n",
    "\n",
    "### Our `build.gradle` file:"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "plugins {\n",
    "   id \"com.redpillanalytics.gradle-confluent\" version '1.1.8'\n",
    "}\n",
    "\n",
    "confluent {\n",
    "    enablePipelines true\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Let's take a look at the Gradle tasks available by applying the plugin:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!./gradlew ksql:tasks --group confluent"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Let's see the *help* assocated with the `pipelineExecute` task:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!./gradlew help --task ksql:pipelineExecute"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Let's take a look at our [pipeline source code](https://github.com/RedPillAnalytics/kafka-examples/tree/master/ksql/src/main/pipeline).\n",
    "\n",
    "### We can use `pipelineExecute` to execute a single pipeline. Notice the plugin auto-generates `TERMINATE` and `DROP` statements:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!./gradlew --console=plain ksql:pipelineExecute --pipeline-dir 01-clickstream"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Or multiple pipelines. We'll turn the logging up a bit. The `DROP` statements occur in reverse order:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!./gradlew --console=plain ksql:pipelineExecute -i"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### The `pipelineExecute` task is great for developers. But in a real delivery pipeline, we want to build and publish artifacts. We are using mavenLocal() which defaults to `$HOME/.m2` as our maven repository location:"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "plugins {\n",
    "   id \"com.redpillanalytics.gradle-confluent\" version \"1.1.8\"\n",
    "   id 'maven-publish'\n",
    "}\n",
    "\n",
    "publishing {\n",
    "    repositories {\n",
    "        mavenLocal()\n",
    "    }\n",
    "}\n",
    "group = 'com.redpillanalytics'\n",
    "version = '1.0.0'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Let's build and then peek inside the artifact:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!./gradlew --console=plain ksql:build ksql:publish\n",
    "!zipinfo ~/.m2/repository/com/redpillanalytics/ksql-pipeline/1.0.0/ksql-pipeline-1.0.0.zip"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### We build the `ksql-script.sql` script in case we want to use it as our queries file, with either option `--queries-file` or, with property `ksql.queries.file`. Notice we *normalize* the script, removing comments, line breaks, etc.:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!cat ksql/build/pipeline/ksql-script.sql"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### We chose to deploy our artifacts using the KSQL REST API, which is similar to using the `pipelineExecute` command, except that it extracts and executes scripts in the artifact instead:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!./gradlew --console=plain ksql:deploy"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Again, with `-i` for more info:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!./gradlew --console=plain ksql:deploy -i"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### We realized quickly that we needed to be able to granularly control some of our auto-generated KSQL. For instance... we wanted to control whether `DELETE TOPIC` was added to `DROP` statements. So we introduced *directives*:"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "--@DeleteTopic\n",
    "CREATE TABLE ENRICHED_ERROR_CODES_COUNT AS\n",
    "SELECT code, definition, COUNT(*) AS count\n",
    "FROM ENRICHED_ERROR_CODES WINDOW TUMBLING (size 30 second)\n",
    "GROUP BY code, definition\n",
    "HAVING COUNT(*) > 1;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Directives are *smart comments* beginning with `--@`. So far we have only introduced `--@DeleteTopic`, but others are planned:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!./gradlew --console=plain ksql:pipelineExecute --pipeline-dir 01-clickstream -i"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# KSQL User Defined Functions\n",
    "\n",
    "### What we really needed was a `CASE` statement. KSQL doesn't have the [CASE](https://github.com/confluentinc/ksql/issues/620) yet, but it's coming.\n",
    "\n",
    "### In the meantime, we wrote the `decode()` function, which was inspired by the Oracle `decode()` function that existed before `CASE`.\n",
    "\n",
    "### `Decode` function:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```groovy\n",
    "import groovy.util.logging.Slf4j\n",
    "import io.confluent.ksql.function.udf.Udf\n",
    "import io.confluent.ksql.function.udf.UdfDescription\n",
    "\n",
    "@Slf4j\n",
    "@UdfDescription(\n",
    "        name = \"decode\",\n",
    "        description = \"\"\"Given up to 3 pairs of 'search' and 'text', return the first 'text' value where 'search' matches 'expression'. If no match, return 'defaultvalue'. 'ignorecase' defaults to 'false'.\"\"\")\n",
    "class Decode {\n",
    "\n",
    "   @Udf(description = \"\"\"Given 1 pair of 'search' and 'text', return the first 'text' value where 'search' matches 'expression'. If no match, return 'defaultvalue'. 'ignorecase' defaults to 'false'.\"\"\")\n",
    "   String decode(String expression, String search1, String text1, String defaultvalue, Boolean ignorecase = false) {\n",
    "\n",
    "      // If any of the expected values are null, then just return null\n",
    "      if (expression == null || search1 == null || text1 == null) return null\n",
    "\n",
    "      return Utils.textMatch(expression, search1, ignorecase) ? text1 : defaultvalue\n",
    "   }\n",
    "    \n",
    "   @Udf(description = \"\"\"Given 2 pairs of 'search' and 'text', return the first 'text' value where 'search' matches 'expression'. If no match, return 'defaultvalue'. 'ignorecase' defaults to 'false'.\"\"\")\n",
    "   String decode(String expression, String search1, String text1, String search2, String text2, String defaultvalue, Boolean ignorecase = false) {\n",
    "\n",
    "      // If any of the expected values are null, then just return null\n",
    "      if (expression == null || search1 == null || text1 == null || search2 == null || text2 == null) return null\n",
    "\n",
    "      if (Utils.textMatch(expression, search1, ignorecase)) return text1\n",
    "\n",
    "      else if (Utils.textMatch(expression, search2, ignorecase)) return text2\n",
    "\n",
    "      else return defaultvalue\n",
    "   }\n",
    "    \n",
    "   @Udf(description = \"\"\"Given 3 pairs of 'search' and 'text', return the first 'text' value where 'search' matches 'expression'. If no match, return 'defaultvalue'. 'ignorecase' defaults to 'false'.\"\"\")\n",
    "   String decode(String expression, String search1, String text1, String search2, String text2, String search3, String text3, String defaultvalue, Boolean ignorecase = false) {\n",
    "\n",
    "      // If any of the expected values are null, then just return null\n",
    "      if (expression == null || search1 == null || text1 == null || search2 == null || text2 == null || search3 == null || text3 == null) return null\n",
    "\n",
    "      if (Utils.textMatch(expression, search1, ignorecase)) return text1\n",
    "\n",
    "      else if (Utils.textMatch(expression, search2, ignorecase)) return text2\n",
    "\n",
    "      else if (Utils.textMatch(expression, search3, ignorecase)) return text3\n",
    "\n",
    "      else return defaultvalue\n",
    "   }\n",
    "}\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### `DecodeTest` Spock test specification:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```groovy\n",
    "import groovy.util.logging.Slf4j\n",
    "import spock.lang.Shared\n",
    "import spock.lang.Specification\n",
    "import spock.lang.Unroll\n",
    "\n",
    "@Slf4j\n",
    "class DecodeTest extends Specification {\n",
    "\n",
    "   @Shared\n",
    "   Decode decode = new Decode()\n",
    "\n",
    "   @Unroll\n",
    "   def \"When: #expression, #search, #text, #defaultValue; Expect: #result\"() {\n",
    "\n",
    "      expect:\n",
    "      decode.decode(expression, search, text, defaultValue) == result\n",
    "\n",
    "      where:\n",
    "      expression    | search        | text  | defaultValue || result\n",
    "      'KSQL Rocks!' | 'ksql rocks!' | 'yes' | 'no'         || 'no'\n",
    "      'KSQL Rocks!' | 'KSQL Rocks!' | 'yes' | 'no'         || 'yes'\n",
    "   }\n",
    "\n",
    "   @Unroll\n",
    "   def \"When: #expression, #search, #text, #defaultValue, #ignorecase; Expect: #result\"() {\n",
    "\n",
    "      expect:\n",
    "      decode.decode(expression, search, text, defaultValue, ignorecase) == result\n",
    "\n",
    "      where:\n",
    "      expression    | search        | text  | defaultValue | ignorecase || result\n",
    "      'KSQL Rocks!' | 'KSQL rocks!' | 'yes' | 'no'         | true       || 'yes'\n",
    "      'KSQL Rocks!' | 'KSQL Rocks!' | 'yes' | 'no'         | true       || 'yes'\n",
    "      'KSQL Rocks!' | 'KSQL rocks!' | 'yes' | 'no'         | false      || 'no'\n",
    "   }\n",
    "\n",
    "   @Unroll\n",
    "   def \"When: #expression, #search1, #text1, #search2, #text2, #defaultValue, #ignorecase; Expect: #result\"() {\n",
    "\n",
    "      expect:\n",
    "      decode.decode(expression, search1, text1, search2, text2, defaultValue, ignorecase) == result\n",
    "\n",
    "      where:\n",
    "      expression    | search1       | text1 | search2       | text2 | defaultValue | ignorecase || result\n",
    "      'KSQL Rocks!' | 'KSQL rocks!' | 'yes' | 'KSQL Sucks!' | 'no'  | 'no'         | true       || 'yes'\n",
    "      'KSQL Rocks!' | 'KSQL rocks!' | 'no'  | 'KSQL Rocks!' | 'yes' | 'no'         | false      || 'yes'\n",
    "      'KSQL Rocks!' | 'KSQL rocks!' | 'no'  | 'KSQL Sucks!' | 'no'  | 'yes'        | false      || 'yes'\n",
    "   }\n",
    "\n",
    "   @Unroll\n",
    "   def \"When: #expression, #search1, #text1, #search2, #text2, #search3, #text3, #defaultValue, #ignorecase; Expect: #result\"() {\n",
    "\n",
    "      expect:\n",
    "      decode.decode(expression, search1, text1, search2, text2, search3, text3, defaultValue, ignorecase) == result\n",
    "\n",
    "      where:\n",
    "      expression    | search1       | text1 | search2       | text2 | search3       | text3 | defaultValue | ignorecase || result\n",
    "      'KSQL Rocks!' | 'KSQL rocks!' | 'yes' | 'KSQL Sucks!' | 'no'  | 'KSQL, meh'   | 'no'  | 'no'         | true       || 'yes'\n",
    "      'KSQL Rocks!' | 'KSQL rocks!' | 'no'  | 'KSQL Rocks!' | 'yes' | 'KSQL, meh'   | 'no'  | 'no'         | false      || 'yes'\n",
    "      'KSQL, meh'   | 'KSQL rocks!' | 'no'  | 'KSQL Sucks!' | 'no'  | 'KSQL, meh'   | 'yes' | 'no'         | false      || 'yes'\n",
    "      'KSQL, meh'   | 'KSQL rocks!' | 'no'  | 'KSQL Sucks!' | 'no'  | \"What's KSQL\" | 'no'  | 'yes'        | false      || 'yes'\n",
    "      'KSQL, meh'   | 'ksql rocks!' | 'no'  | 'ksql, meh'   | 'yes' | \"What's KSQL\" | 'no'  | 'no'         | true       || 'yes'\n",
    "\n",
    "   }\n",
    "}\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### The KSQL server does not resolve any of our library dependencies... not even the `io.confluent.ksql.function.udf` library that contains our KSQL annotations. So we need to compile a *fat* or *uber* JAR with all of our dependencies included. For this, we use the `shadow` plugin.\n",
    "\n",
    "### Here is our `build.gradle` file for the `functions` subproject:"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "plugins {\n",
    "   id 'groovy'\n",
    "   id 'maven-publish'\n",
    "   id 'com.adarshr.test-logger' version '1.6.0'\n",
    "   id \"com.github.johnrengelman.shadow\" version \"5.0.0\"\n",
    "}\n",
    "\n",
    "//customize ShadowJar\n",
    "jar.enabled = false\n",
    "shadowJar { classifier = '' }\n",
    "tasks.build.dependsOn tasks.shadowJar\n",
    "\n",
    "// mavenLocal publish\n",
    "publishing {\n",
    "   publications {\n",
    "      shadow(MavenPublication) { publication ->\n",
    "         project.shadow.component(publication)\n",
    "      }\n",
    "   }\n",
    "   repositories {\n",
    "      mavenLocal()\n",
    "   }\n",
    "}\n",
    "group = 'com.redpillanalytics'\n",
    "version = '1.0.0'\n",
    "\n",
    "dependencies {\n",
    "   compile localGroovy()\n",
    "   compile 'org.slf4j:slf4j-simple:+'\n",
    "   compile 'io.confluent.ksql:ksql-udf:+'\n",
    "   testCompile 'org.spockframework:spock-core:1.2-groovy-2.5'\n",
    "}\n",
    "\n",
    "// confluent dependencies\n",
    "repositories {\n",
    "   jcenter()\n",
    "   maven {\n",
    "      url \"http://packages.confluent.io/maven/\"\n",
    "   }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Let's build and publish our UDF artifact, and see how many items are in it:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!./gradlew --console=plain functions:build functions:publish\n",
    "!zipinfo -h ~/.m2/repository/com/redpillanalytics/functions/1.0.0/functions-1.0.0.jar"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### We now copy our full JAR artifact, with all dependencies, to the `/etc/ksql-server/ext` directory, which is where our KSQL Server property `ksql.extension.dir` points to:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!/usr/local/bin/docker cp ~/.m2/repository/com/redpillanalytics/functions/1.0.0/functions-1.0.0.jar ksql-server:/etc/ksql-server/ext"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### To speed up the demo, I copied the artifact to the server ahead of time... so no KSQL Server restart is required:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!echo \"DESCRIBE FUNCTION DECODE;\" | \"$ksql\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = (\n",
    "    \"select definition, decode(definition, \"\n",
    "    \"'Proxy authentication required','Bad', \"\n",
    "    \"'Page not found','Bad', \"\n",
    "    \"'Redirect','Good', \"\n",
    "    \"'Unknown') label \"\n",
    "    \"from enriched_error_codes limit 20;\"\n",
    ")\n",
    "!echo \"$sql\" | \"$ksql\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Kafka Streams Microservices\n",
    "\n",
    "### Each of our Kafka Streams processes was developed as a separate service, and it needs to build and deploy that way. As an example, we've borrowed the `WordCountLambdaExample` class from the `kafka-streams-examples` repository provided by Confluent:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```java\n",
    "import org.apache.kafka.common.serialization.Serde;\n",
    "import org.apache.kafka.common.serialization.Serdes;\n",
    "import org.apache.kafka.streams.KafkaStreams;\n",
    "import org.apache.kafka.streams.StreamsBuilder;\n",
    "import org.apache.kafka.streams.StreamsConfig;\n",
    "import org.apache.kafka.streams.kstream.KStream;\n",
    "import org.apache.kafka.streams.kstream.KTable;\n",
    "import org.apache.kafka.streams.kstream.Produced;\n",
    "\n",
    "import java.util.Arrays;\n",
    "import java.util.Properties;\n",
    "import java.util.regex.Pattern;\n",
    "\n",
    "\n",
    "public class WordCountLambdaExample {\n",
    "\n",
    "  public static void main(final String[] args) throws Exception {\n",
    "    final String bootstrapServers = args.length > 0 ? args[0] : \"localhost:9092\";\n",
    "    final Properties streamsConfiguration = new Properties();\n",
    "    // Give the Streams application a unique name.  The name must be unique in the Kafka cluster\n",
    "    // against which the application is run.\n",
    "    streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, \"wordcount-lambda-example\");\n",
    "    streamsConfiguration.put(StreamsConfig.CLIENT_ID_CONFIG, \"wordcount-lambda-example-client\");\n",
    "    // Where to find Kafka broker(s).\n",
    "    streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);\n",
    "    // Specify default (de)serializers for record keys and for record values.\n",
    "    streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());\n",
    "    streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());\n",
    "    // Records should be flushed every 10 seconds. This is less than the default\n",
    "    // in order to keep this example interactive.\n",
    "    streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000);\n",
    "    // For illustrative purposes we disable record caches\n",
    "    streamsConfiguration.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);\n",
    "\n",
    "    // Set up serializers and deserializers, which we will use for overriding the default serdes\n",
    "    // specified above.\n",
    "    final Serde<String> stringSerde = Serdes.String();\n",
    "    final Serde<Long> longSerde = Serdes.Long();\n",
    "\n",
    "    // In the subsequent lines we define the processing topology of the Streams application.\n",
    "    final StreamsBuilder builder = new StreamsBuilder();\n",
    "\n",
    "    // Construct a `KStream` from the input topic \"streams-plaintext-input\", where message values\n",
    "    // represent lines of text (for the sake of this example, we ignore whatever may be stored\n",
    "    // in the message keys).\n",
    "    //\n",
    "    // Note: We could also just call `builder.stream(\"streams-plaintext-input\")` if we wanted to leverage\n",
    "    // the default serdes specified in the Streams configuration above, because these defaults\n",
    "    // match what's in the actual topic.  However we explicitly set the deserializers in the\n",
    "    // call to `stream()` below in order to show how that's done, too.\n",
    "    final KStream<String, String> textLines = builder.stream(\"clickstream\");\n",
    "\n",
    "    final Pattern pattern = Pattern.compile(\"\\\\W+\", Pattern.UNICODE_CHARACTER_CLASS);\n",
    "\n",
    "    final KTable<String, Long> wordCounts = textLines\n",
    "      // Split each text line, by whitespace, into words.  The text lines are the record\n",
    "      // values, i.e. we can ignore whatever data is in the record keys and thus invoke\n",
    "      // `flatMapValues()` instead of the more generic `flatMap()`.\n",
    "      .flatMapValues(value -> Arrays.asList(pattern.split(value.toLowerCase())))\n",
    "      // Count the occurrences of each word (record key).\n",
    "      //\n",
    "      // This will change the stream type from `KStream<String, String>` to `KTable<String, Long>`\n",
    "      // (word -> count).  In the `count` operation we must provide a name for the resulting KTable,\n",
    "      // which will be used to name e.g. its associated state store and changelog topic.\n",
    "      //\n",
    "      // Note: no need to specify explicit serdes because the resulting key and value types match our default serde settings\n",
    "      .groupBy((key, word) -> word)\n",
    "      .count();\n",
    "\n",
    "    // Write the `KTable<String, Long>` to the output topic.\n",
    "    wordCounts.toStream().to(\"clickstream-wordcount\", Produced.with(stringSerde, longSerde));\n",
    "\n",
    "    // Now that we have finished the definition of the processing topology we can actually run\n",
    "    // it via `start()`.  The Streams application as a whole can be launched just like any\n",
    "    // normal Java application that has a `main()` method.\n",
    "    final KafkaStreams streams = new KafkaStreams(builder.build(), streamsConfiguration);\n",
    "    // Always (and unconditionally) clean local state prior to starting the processing topology.\n",
    "    // We opt for this unconditional call here because this will make it easier for you to play around with the example\n",
    "    // when resetting the application for doing a re-run (via the Application Reset Tool,\n",
    "    // http://docs.confluent.io/current/streams/developer-guide.html#application-reset-tool).\n",
    "    //\n",
    "    // The drawback of cleaning up local state prior is that your app must rebuilt its local state from scratch, which\n",
    "    // will take time and will require reading all the state-relevant data from the Kafka cluster over the network.\n",
    "    // Thus in a production scenario you typically do not want to clean up always as we do here but rather only when it\n",
    "    // is truly needed, i.e., only under certain conditions (e.g., the presence of a command line flag for your app).\n",
    "    // See `ApplicationResetExample.java` for a production-like example.\n",
    "    streams.cleanUp();\n",
    "    streams.start();\n",
    "\n",
    "    // Add shutdown hook to respond to SIGTERM and gracefully close Kafka Streams\n",
    "    Runtime.getRuntime().addShutdownHook(new Thread(streams::close));\n",
    "  }\n",
    "\n",
    "}\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "### We used the Gradle `application` plugin. Here is our `build.gradle` file for building this service:"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "plugins {\n",
    "   id 'application'\n",
    "   id 'maven-publish'\n",
    "}\n",
    "\n",
    "// application entry point\n",
    "mainClassName = 'WordCountLambdaExample'\n",
    "applicationName = 'wordcount-lambda-example'\n",
    "\n",
    "// mavenLocal publish\n",
    "publishing {\n",
    "   publications {\n",
    "      streams(MavenPublication) {\n",
    "         artifact distZip\n",
    "      }\n",
    "   }\n",
    "   repositories {\n",
    "      mavenLocal()\n",
    "   }\n",
    "}\n",
    "// Default artifact naming.\n",
    "group = 'com.redpillanalytics'\n",
    "version = '1.0.0'\n",
    "\n",
    "dependencies {\n",
    "   compile 'org.slf4j:slf4j-simple:+'\n",
    "   compile 'org.apache.kafka:kafka-streams:+'\n",
    "}\n",
    "\n",
    "repositories {\n",
    "   jcenter()\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### The `application` plugin provides us the `run` task, which allows developers to test their application easily:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!./gradlew --console=plain streams:run -q"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Running Streams using Gradle directly from the Git repository is not an ideal solution for a production microservice. Thankfully, the Application plugin can create distribution artifacts complete with start scripts that can be packaged up as part of the build process:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!./gradlew --console=plain streams:build streams:publish"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### We can now get our distribution artifact from the Maven repository, unzip it, and start the application. In our production environment, we used Jenkins processes to manage this side of it:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!rm -rf streams/build/deploy\n",
    "!mkdir -p streams/build/deploy\n",
    "!cp ~/.m2/repository/com/redpillanalytics/streams/1.0.0/streams-1.0.0.zip streams/build/deploy/streams-1.0.0.zip\n",
    "!unzip streams/build/deploy/streams-1.0.0.zip -d streams/build/deploy\n",
    "!./streams/build/deploy/wordcount-lambda-example-1.0.0/bin/wordcount-lambda-example"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# One More Thing... Gradle Analytics\n",
    "\n",
    "### We've also contributed the `gradle-analytics` plugin, which captures everything going on side of Gradle, and makes it available to numerous cloud analytics sources... and of course, Apache Kafka.\n",
    "\n",
    "### https://github.com/RedPillAnalytics/gradle-analytics"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "plugins {\n",
    "    // facilitates publishing build activity to streaming and analytics platforms\n",
    "    // including Kafka\n",
    "    id 'com.redpillanalytics.gradle-analytics' version \"1.2.3\"\n",
    "}\n",
    "\n",
    "analytics.sinks {\n",
    "    kafka {\n",
    "        servers = 'localhost:9092'\n",
    "        acks = 'all'\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### By default, the `producer` task compresses and cleans files once they've been sent. We aren't doing that..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!./gradlew --console=plain producer -Panalytics.compressFiles=false -Panalytics.cleanFiles=false"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Now let's see our new topics:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!echo \"LIST TOPICS;\" | \"$ksql\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!/Users/stewartbryson/bin/kaf consume gradle_build"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!/Users/stewartbryson/bin/kaf consume gradle_task"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!/Users/stewartbryson/bin/kaf consume gradle_test"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!/Users/stewartbryson/bin/kaf consume gradle_ksqlstatements"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Stopping our Docker environment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!./gradlew --console=plain -q composeDown\n",
    "!/usr/local/bin/docker ps"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "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.6.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}