apurvam

/* Instructions on compilation and execution
 * =========================================
 *
 * Compile this program with pthreads:
 *
 *   g++ -Wall -lpthread -o graphdb-simulator graphdb-simulator.cpp
 *
 * Before you run this program, you need to create the following
 * directories:
 *

apurvam

public interface Producer<K,V> extends Closeable {
  
  /**
   * Needs to be called before any of the other transaction methods. Assumes that 
   * the producer.app.id is specified in the producer configuration.
   *
   * This method does the following:
   *   1. Ensures any transactions initiated by previous instances of the producer
   *      are committed or rolled back.
   *   2. Gets the internal producer id and epoch, used in all future transactional

apurvam

public interface Consumer<K,V> extends Closeable {
  
  /**
   * Returns the currently consumed but uncommitted offsets.
   */
  Map<TopicPartition, OffsetAndMetadata> uncommittedOffsets();
  
  /**
   * Returns the consumer group id.
   */

apurvam

public class KafkaTransactionsExample {
 
  public static void main(String args[]) {
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerConfig);


    // Note that the ‘transaction.app.id’ configuration _must_ be specified in the 
    // producer config in order to use transactions.
    KafkaProducer<String, String> producer = new KafkaProducer<>(producerConfig);

apurvam

public class TransactionResponse {
  
  /**
   * All the error codes that may be returned by transactional methods.
   */
  public enum TransactionError {
    PRODUCER_FENCED,
    NO_PID_MAPPING,
    NOT_COORDINATOR_FOR_APP_ID,
    INVALID_TXN_REQUEST,

apurvam

KafkaProducer producer = createKafkaProducer(
  “bootstrap.servers”, “localhost:9092”,
  “transactional.id”, “my-transactional-id”);

producer.initTransactions();

KafkaConsumer consumer = createKafkaConsumer(
  “bootstrap.servers”, “localhost:9092”,
  “group.id”, “my-group-id”,
  "isolation.level", "read_committed");

apurvam

ksql> show topics;

 Kafka Topic                 | Registered | Partitions | Partition Replicas | Consumers | Consumer Groups
----------------------------------------------------------------------------------------------------------
 __confluent.support.metrics | false      | 1          | 1                  | 0         | 0
 _schemas                    | false      | 1          | 1                  | 0         | 0
 connect-configs             | false      | 1          | 1                  | 0         | 0
 connect-offsets             | false      | 25         | 1                  | 0         | 0
 connect-statuses            | false      | 5          | 1                  | 0         | 0
 groupbydevice               | true       | 4          | 1                  | 0         | 0

apurvam

ksql> print 'pageviews_kafka_topic_json';
Format:JSON
{"ROWTIME":1516753301093,"ROWKEY":"1","viewtime":1516753300961,"pageid":"Page_53","userid":"User_9"}
{"ROWTIME":1516753301458,"ROWKEY":"11","viewtime":1516753301458,"pageid":"Page_38","userid":"User_5"}
{"ROWTIME":1516753301660,"ROWKEY":"21","viewtime":1516753301660,"pageid":"Page_25","userid":"User_2"}
^C

apurvam

SELECT orderzip_code, TOPK(order_total, 5) FROM orders WINDOW TUMBLING (SIZE 1 HOUR) GROUP BY order_zipcode;

apurvam

SELECT pageid, TOPKDISTINCT(viewtime, 5) FROM pageviews_users GROUP BY pageid;