Acceldata
 ODP
Search documentation… ⌘K

Superset Pinot testing - Stream Data

Make sure you have Kafka (used kafka3 here), Pinot (1.4.0), and Superset running.

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"

Set up Kafka

Client.properties for krb auth

[root@newsuper-2 kafka3]# cat client.properties
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/kafka.service.keytab" \
 storeKey=true \
 useTicketCache=false \
 principal="kafka/newsuper-2.newsuper.harshith.svc.cluster.local@ADSRE.COM";

Create a Topic

bin/kafka-topics.sh --create \
 --topic flight-events \
 --bootstrap-server $KAFKA_BROKERS \
 --partitions 3 \
 --replication-factor 3 \
 --command-config client.properties
 
bin/kafka-topics.sh --list \
 --bootstrap-server $KAFKA_BROKERS \
 --command-config client.properties
 
bin/kafka-topics.sh --describe \
 --topic flight-events \
 --bootstrap-server $KAFKA_BROKERS \
 --command-config client.properties

Producer Script to Ingest Data

#!/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

Set up Pinot Injestion

Create Pinot Realtime Table

Create Schema

cat > flightEvents_schema.json << 'EOF'
{
  "schemaName": "flightEvents",
  "dimensionFieldSpecs": [
    {"name": "Carrier", "dataType": "STRING"},
    {"name": "FlightNum", "dataType": "INT"},
    {"name": "Origin", "dataType": "STRING"},
    {"name": "Dest", "dataType": "STRING"}
  ],
  "metricFieldSpecs": [
    {"name": "ArrDelay", "dataType": "INT"},
    {"name": "DepDelay", "dataType": "INT"},
    {"name": "AirTime", "dataType": "INT"},
    {"name": "Distance", "dataType": "INT"}
  ],
  "dateTimeFieldSpecs": [
    {
      "name": "timestamp",
      "dataType": "LONG",
      "format": "1:MILLISECONDS:EPOCH",
      "granularity": "1:MILLISECONDS"
    }
  ]
}
EOF

Create Table 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
  • Now give permissions for Pinot to read Kafka (for that topic or in general, all topics) and add Pinot Jaas to the Pinot process so that Pinot can pass through the Kerberos authentication.
cat > /usr/odp/3.3.6.3-101/pinot/pinot_jaas.conf << 'EOF'
KafkaClient {
  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";
};
 
Client {
  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";
};
EOF

Add this to Pinot JVM args,

-Djava.security.auth.login.config=/usr/odp/3.3.6.3-101/pinot/pinot_jaas.conf

like the following, for all components of Pinot.

Preserved image

Info

If you see failures related to auth, you can try adding -Dsun.security.krb5.debug=true to above and see some logs.

[Kafka topic is still active, and messages are being pushed and consumed by Pinot Live]

And once that's done, as we have already connected Pinot to Kafka from previous testing, we should see a new table in Superset.

Create Charts

Bar Chart - Live Flights by Carrier

| Field | Value | Description |
|-------|-------|-------------|
| X-Axis | `Carrier` | Airline carrier codes on horizontal axis |
| Metrics | `COUNT(*)` | Number of flight events (bar height) |
| Dimensions | *(leave empty)* | Not needed for simple bar chart |
| Time Range | `Last hour` | Filter to recent data |

Preserved image

Line Chart - Arrival Delays Over Time

| Field | Value | Description |
|-------|-------|-------------|
| X-Axis | `timestamp` | Time on horizontal axis |
| Time Grain | `minute` | Aggregate by minute |
| Metrics | `AVG(ArrDelay)` | Average arrival delay (line value) |
| Dimensions | *(leave empty)* | Single line for all carriers |
| Time Range | `Last hour` | Show last hour of data |

Preserved image

Big Number - Total Events

| Field | Value | Description |
|-------|-------|-------------|
| Chart Type | `Big Number` | Single large metric display |
| Metric | `COUNT(*)` | Total event count |

Preserved image

Dashboard

Preserved image

  • Refresh (on 3dot menu, top right)

Preserved image

  • Autorefresh

Video live refresh for the Dashboard

The following is a quick video showing autorefresh.

Preserved image