import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import java.util.Properties;
public class KafkaProducer {
private static Producer<Integer, String> producer;
private final Properties properties = new Properties();
public KafkaProducer() {
properties.put("metadata.broker.list", "buckland:6667,laverne:6667,mahoney:6667");
properties.put("serializer.class", "kafka.serializer.StringEncoder");
properties.put("request.required.acks", "1");
properties.put("security.protocol", "PLAINTEXTSASL");
properties.put("producer.type", "async");
ProducerConfig config = new ProducerConfig(properties);
producer = new Producer<Integer, String>(config);
}
public static void main(String[] args) {
System.out.println("in Kakfa Producer ****");
new KafkaProducer();
String topic = "dev.raw.mce.energy.gas";
String msg = "***** message from KafkaProducer class *****";
KeyedMessage<Integer, String> data = new KeyedMessage<>(topic, msg);
producer.send(data);
producer.close();
System.out.println(" Kakfa Producer is over ****");
}
}
-- From Program Producer
spark-submit \
--master local[2] \
--num-executors 2 \
--driver-memory 1g \
--executor-memory 2g \
--conf "spark.driver.extraJavaOptions= -Djava.security.auth.login.config=/tmp/rchamaku/kafka/kafka_client_jaas.conf" \
--class KafkaProducer \
--name "Sample KafkaProducer by Ram" \
/tmp/rchamaku/kafka/TestKafka-0.0.1-SNAPSHOT-driver.jar
... View more
