Superset Pinot testing - Stream Data

export PINOT_HOST=$(hostname -f)

export KAFKA_BROKERS="newsuper-0.newsuper.harshith.svc.cluster.local:6669,newsuper-1.newsuper.harshith.svc.cluster.local:6669,newsuper-2.newsuper.harshith.svc.cluster.local:6669"

#!/bin/bash

PINOT_HOST=$(hostname -f)

KAFKA_BROKERS="newsuper-0.newsuper.harshith.svc.cluster.local:6669,newsuper-1.newsuper.harshith.svc.cluster.local:6669,newsuper-2.newsuper.harshith.svc.cluster.local:6669"

KAFKA_HOME="${KAFKA_HOME:-.}"

KAFKA_BROKERS="${KAFKA_BROKERS:-localhost:9092}"

TOPIC="flight-events"

CLIENT_CONFIG="client.properties"

CARRIERS=("AA" "UA" "DL" "WN" "AS" "B6" "NK" "F9" "G4" "HA")

ORIGINS=("SFO" "LAX" "JFK" "ORD" "DFW" "ATL" "SEA" "BOS" "MIA" "DEN")

DESTS=("SFO" "LAX" "JFK" "ORD" "DFW" "ATL" "SEA" "BOS" "MIA" "DEN")

echo "Publishing flight events to $TOPIC..."

echo "Press Ctrl+C to stop"

# Generate messages and pipe to a single producer instance

while true; do

  TIMESTAMP=$(date +%s000)

  CARRIER=${CARRIERS[$RANDOM % ${#CARRIERS[@]}]}

  ORIGIN=${ORIGINS[$RANDOM % ${#ORIGINS[@]}]}

  DEST=${DESTS[$RANDOM % ${#DESTS[@]}]}

  while [ "$ORIGIN" == "$DEST" ]; do

    DEST=${DESTS[$RANDOM % ${#DESTS[@]}]}

  done

  FLIGHT_NUM=$((RANDOM % 9000 + 1000))

  ARR_DELAY=$((RANDOM % 120 - 30))

  DEP_DELAY=$((RANDOM % 60 - 15))

  AIR_TIME=$((RANDOM % 300 + 60))

  DISTANCE=$((RANDOM % 2500 + 200))

  echo "{\"timestamp\":${TIMESTAMP},\"Carrier\":\"${CARRIER}\",\"FlightNum\":${FLIGHT_NUM},\"Origin\":\"${ORIGIN}\",\"Dest\":\"${DEST}\",\"ArrDelay\":${ARR_DELAY},\"DepDelay\":${DEP_DELAY},\"AirTime\":${AIR_TIME},\"Distance\":${DISTANCE}}"

  sleep 1

done | ${KAFKA_HOME}/bin/kafka-console-producer.sh \

  --bootstrap-server $KAFKA_BROKERS \

  --topic $TOPIC \

  --producer.config $CLIENT_CONFIG

cat > flightEvents_realtime_table_config.json << 'EOF'

{

  "tableName": "flightEvents",

  "tableType": "REALTIME",

  "segmentsConfig": {

    "timeColumnName": "timestamp",

    "timeType": "MILLISECONDS",

    "schemaName": "flightEvents",

    "replicasPerPartition": "1"

  },

  "tenants": {},

  "tableIndexConfig": {

    "loadMode": "MMAP",

    "streamConfigs": {

      "streamType": "kafka",

      "stream.kafka.consumer.type": "lowlevel",

      "stream.kafka.topic.name": "flight-events",

      "stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",

      "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",

      "stream.kafka.broker.list": "newsuper-0.newsuper.harshith.svc.cluster.local:6669,newsuper-1.newsuper.harshith.svc.cluster.local:6669,newsuper-2.newsuper.harshith.svc.cluster.local:6669",

      "security.protocol": "SASL_PLAINTEXT",

      "sasl.mechanism": "GSSAPI",

      "sasl.kerberos.service.name": "kafka",

      "sasl.jaas.config": "com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true keyTab=\"/etc/security/keytabs/pinot.headless.keytab\" storeKey=true useTicketCache=false principal=\"pinot-odp_quantum@ADSRE.COM\";",

      "realtime.segment.flush.threshold.rows": "10000",

      "realtime.segment.flush.threshold.time": "1h"

    }

  },

  "metadata": {

    "customConfigs": {}

  }

}

EOF