DennisFederico

Event Deduplication in Kafka Stream processing using ksqlDB (SIMPLIFIED)

Introduction

Event deduplication emiting the very first message and filtering the rest of the duplicates within a tumbling window or a session window, using ksqlDB.

The ID to deduplicate is the eventId field inside the record payload, thus the first step is to re-key the stream by this field.

The deduplication is done by counting the number of times the eventId appears in the stream within a window, and only emitting the first event that has a count of 1.

DennisFederico

using Selection.Components;
using Unity.Burst;
using Unity.Entities;
using Unity.Mathematics;
using Unity.Physics;
using Unity.Physics.Systems;
using Unity.Transforms;
using UnityEngine;
using BoxCollider = Unity.Physics.BoxCollider;

DennisFederico

---

Using Openssl

Certificate Authority

CA Key

DennisFederico

Install Prometheus-Grafana pack

https://www.confluent.io/blog/monitor-kafka-clusters-with-prometheus-grafana-and-confluent/

Note on Ansible Playbooks and Binaries

This document is prepared to set up the monitoring stack using Ansible as much as possible, Prometheus and Grafana have been tested to work on an Air-Gap (no internet access) environment providing the binaries from the official site, using the playbooks from GitHub user (0x0I)[https://github.com/O1ahmad]. Prometheus Node exporter playbook, provided by (Clud Alchemy)[https://github.com/cloudalchemy] has not bee tested on an Air-Gap environment, it doesn't mean it couldn't work.

DennisFederico

---
all:
  vars:
    ## ANSIBLE CONNECTIVITY PARAMS AND FLAGS
    ansible_connection: ssh
    ansible_user: dfederico
    ansible_become: true
    ansible_ssh_private_key_file: ~/.ssh/id_rsa
    ansible_python_interpreter: /usr/bin/python3
    ansible_ssh_common_args: '-o StrictHostKeyChecking=no'

DennisFederico

Self-signed Certificates

This were prepared using cfssl (Cloud Flare SSL) tools, a wrapper over openssl to simplify its usage, thus opensslmust be installed too.

Prepare CA - Certificate Authority

Create CA Key

$ openssl genrsa -out generated/CAkey.pem 2048

DennisFederico

tags: #ksqldb/tutorial

---

Provisioning ksqlDB Cluster

Cluster Data

environment=env-0x3135

DennisFederico

CREATE TYPE BALANCE_TYPE AS STRUCT<account_id STRING, balance DOUBLE>;

CREATE STREAM CUSTOMER_STATS (
  CUSTOMER_ID STRING KEY,
  CUSTOMER_NAME STRING,
  BALANCES ARRAY<BALANCE_TYPE>
) WITH (
  KAFKA_TOPIC='CUSTOMER_STATS',
  PARTITIONS=1,  
  VALUE_FORMAT='JSON_SR');

DennisFederico

export BOOTSTRAP_SERVERS=<CONFLUENT_CLOUD_CLUSTER_BOOTSTRAP:9092>
export SASL_JAAS_CONFIG="org.apache.kafka.common.security.plain.PlainLoginModule required username='<API-KEY>' password='<API_SECRET>';"
export SCHEMA_REGISTRY_URL=<CONFLUENT_CLOUD_SCHEMA_REGISTRY_URL:443>
export BASIC_AUTH_CREDENTIALS_SOURCE=USER_INFO
export SCHEMA_REGISTRY_BASIC_AUTH_USER_INFO=<SR_API_KEY>:<SR_API_SECRET>

DennisFederico

---
version: '2'

services:
  connect_1:
    image: confluentinc/cp-server-connect:7.2.1
    hostname: connect-1
    container_name: connect-1
    ports:
      - "8083:8083"