forked from TIBCOSoftware/snappy-examples
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathKafkaAdImpressionGenerator.scala
75 lines (66 loc) · 2.63 KB
/
KafkaAdImpressionGenerator.scala
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
/*
* Copyright (c) 2016 SnappyData, Inc. All rights reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you
* may not use this file except in compliance with the License. You
* may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
* implied. See the License for the specific language governing
* permissions and limitations under the License. See accompanying
* LICENSE file.
*/
package io.snappydata.adanalytics.aggregator
import java.util.Properties
import io.snappydata.adanalytics.aggregator.Constants._
import io.snappydata.adanalytics.aggregator.KafkaAdImpressionGenerator._
import kafka.producer.{KeyedMessage, Producer, ProducerConfig}
/**
* A simple Kafka Producer program which randomly generates
* ad impression log messages and sends it to Kafka broker.
* This program generates and sends 10 million messages.
*/
object KafkaAdImpressionGenerator extends AdImpressionGenerator {
val props = new Properties()
props.put("producer.type", "async")
props.put("request.required.acks", "0")
props.put("serializer.class", "io.snappydata.adanalytics.aggregator.AdImpressionLogAvroEncoder")
props.put("queue.buffering.max.messages", "1000000") // 10000
props.put("metadata.broker.list", brokerList)
props.put("partitioner.class", "kafka.producer.DefaultPartitioner")
props.put("key.serializer.class", "kafka.serializer.StringEncoder")
props.put("batch.size", "9000000") // bytes
props.put("linger.ms", "50")
val config = new ProducerConfig(props)
val producer = new Producer[String, AdImpressionLog](config)
def main(args: Array[String]) {
println("Sending Kafka messages of topic " + kafkaTopic + " to brokers " + brokerList)
val threads = new Array[Thread](numProducerThreads)
for (i <- 0 until numProducerThreads) {
val thread = new Thread(new Worker())
thread.start()
threads(i) = thread
}
threads.foreach(_.join())
println(s"Done sending $totalNumLogs Kafka messages of topic $kafkaTopic")
System.exit(0)
}
def sendToKafka(log: AdImpressionLog) = {
producer.send(new KeyedMessage[String, AdImpressionLog](
Constants.kafkaTopic, log.getTimestamp.toString, log))
}
}
final class Worker extends Runnable {
def run() {
for (i <- 1 to totalNumLogs) {
sendToKafka(generateAdImpression())
if ((i % 1000000) == 0) {
println(s"Sent $i Kafka messages of topic $kafkaTopic")
}
}
}
}