# bin/kafka-topics.sh --zookeeper $zk:2181/kafka -create --topic test_sync --partitions 2 --replication-factor 2

WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.

Created topic "test_sync".

# bin/kafka-topics.sh --zookeeper $zk:2181/kafka -describe --topic test_sync

Topic:test_sync PartitionCount:2        ReplicationFactor:2     Configs:

        Topic: test_sync        Partition: 0    Leader: 112     Replicas: 112,111       Isr: 112,111

        Topic: test_sync        Partition: 1    Leader: 110     Replicas: 110,112       Isr: 110,112

CREATE TABLE test.test_sync (

id int,

name string,

description string,

create_time timestamp,

update_time timestamp,

primary key (id)

)

PARTITION BY HASH (id) PARTITIONS 4

STORED AS KUDU

TBLPROPERTIES ('kudu.master_addresses'='$kudu_master:7051');

# wget https://repository.cloudera.com/artifactory/cloudera-repos/org/apache/kudu/kudu-flume-sink/1.7.0-cdh5.16.1/kudu-flume-sink-1.7.0-cdh5.16.1.jar

# mv kudu-flume-sink-1.7.0-cdh5.16.1.jar $FLUME_HOME/lib/

# wget http://central.maven.org/maven2/org/json/json/20160810/json-20160810.jar

# mv json-20160810.jar $FLUME_HOME/lib/

  public List<Operation> getOperations(Event event) throws FlumeException {

    try {

      Insert insert = table.newInsert();

      PartialRow row = insert.getRow();

      row.addBinary(payloadColumn, event.getBody());

      return Collections.singletonList((Operation) insert);

    } catch (Exception e) {

      throw new FlumeException("Failed to create Kudu Insert object", e);

    }

  }

package com.cloudera.kudu;

public class JsonKuduOperationsProducer implements KuduOperationsProducer {

a1.sources = r1

a1.sinks = k1

a1.channels = c1

# Describe/configure the source

a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource

a1.sources.r1.batchSize = 5000

a1.sources.r1.batchDurationMillis = 2000

a1.sources.r1.kafka.bootstrap.servers = 192.168.0.1:9092

a1.sources.r1.kafka.topics = test_sync

a1.sources.r1.kafka.consumer.group.id = flume-consumer

# Describe the sink

a1.sinks.k1.type = logger

# Use a channel which buffers events in memory

a1.channels.c1.type = memory

a1.channels.c1.capacity = 10000

a1.channels.c1.transactionCapacity = 10000

# Bind the source and sink to the channel

a1.sources.r1.channels = c1

a1.sinks.k1.channel = c1

a1.sinks.k1.type = org.apache.kudu.flume.sink.KuduSink

a1.sinks.k1.producer = com.cloudera.kudu.JsonKuduOperationsProducer

a1.sinks.k1.masterAddresses = 192.168.0.1:7051

a1.sinks.k1.tableName = impala::test.test_sync

a1.sinks.k1.batchSize = 50

bin/flume-ng agent --conf conf --conf-file conf/order.properties --name a1

select * from test_sync limit 10;