1.概述

目前，隨著大數據的浪潮，Kafka 被越來越多的企業所認可，如今的Kafka已發展到0.10.x，其優秀的特性也帶給我們解決實際業務的方案。對於數據分流來說，既可以分流到離線存儲平台（HDFS），離線計算平台（Hive倉庫），也可以分流實時流水計算（Storm，Spark）等，同樣也可以分流到海量數據查詢（HBase），或是及時查詢（ElasticSearch）。而今天筆者給大家分享的就是Kafka 分流數據到 ElasticSearch。

2.內容

我們知道，ElasticSearch是有其自己的套件的，簡稱ELK，即ElasticSearch，Logstash以及Kibana。ElasticSearch負責存儲，Logstash負責收集數據來源，Kibana負責可視化數據，分工明確。想要分流Kafka中的消息數據，可以使用Logstash的插件直接消費，但是需要我們編寫複雜的過濾條件，和特殊的映射處理，比如系統保留的`_uid`欄位等需要我們額外的轉化。今天我們使用另外一種方式來處理數據，使用Kafka的消費API和ES的存儲API來處理分流數據。通過編寫Kafka消費者，消費對應的業務數據，將消費的數據通過ES存儲API，通過創建對應的索引的，存儲到ES中。其流程如下圖所示：

3.實現

下面，我們開始進行實現細節處理，這裡給大家提供實現的核心代碼部分，實現代碼如下所示：

我們以插件的形式進行消費，從Kafka到ES的數據流向，只需要定義插件格式，如下所示：

{
"job": {
"content": {
"reader": {
"name": "kafka",
"parameter": {
"topic": "kafka_es_client_error",
"groupid": "es2",
"bootstrapServers": "k1:9094,k2:9094,k3:9094"
},
"threads": 6
},
"writer": {
"name": "es",
"parameter": {
"host": [
"es1:9300,es2:9300,es3:9300"
],
"index": "client_error_%s",
"type": "client_error"
}
}
}
}
}

這裡處理消費存儲的方式，將讀和寫的源分開，配置各自屬性即可。

這裡，我們通過每天建立索引進行存儲，便於業務查詢，實現細節如下所示：

public class EsProducer {

private final static Logger LOG = LoggerFactory.getLogger(EsProducer.class);
private final KafkaConsumer consumer;
private ExecutorService executorService;
private Configuration conf = null;
private static int counter = 0;

public EsProducer {
String root = System.getProperty("user.dir") + "/conf/";
String path = SystemConfigUtils.getProperty("kafka.x.plugins.exec.path");
conf = Configuration.from(new File(root + path));
Properties props = new Properties;
props.put("bootstrap.servers", conf.getString("job.content.reader.parameter.bootstrapServers"));
props.put("group.id", conf.getString("job.content.reader.parameter.groupid"));
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
consumer = new KafkaConsumer(props);
consumer.subscribe(Arrays.asList(conf.getString("job.content.reader.parameter.topic")));
}

public void execute {
executorService = Executors.newFixedThreadPool(conf.getInt("job.content.reader.threads"));
while (true) {
ConsumerRecords records = consumer.poll(100);
if (null != records) {
executorService.submit(new KafkaConsumerThread(records, consumer));
}
}
}

public void shutdown {
try {
if (consumer != null) {
consumer.close;
}
if (executorService != null) {
executorService.shutdown;
}
if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {
LOG.error("Shutdown kafka consumer thread timeout.");
}
} catch (InterruptedException ignored) {
Thread.currentThread.interrupt;
}
}

class KafkaConsumerThread implements Runnable {

private ConsumerRecords records;

public KafkaConsumerThread(ConsumerRecords records, KafkaConsumer consumer) {
this.records = records;
}

@Override
public void run {
String index = conf.getString("job.content.writer.parameter.index");
String type = conf.getString("job.content.writer.parameter.type");
for (TopicPartition partition : records.partitions) {
List> partitionRecords = records.records(partition);
for (ConsumerRecord record : partitionRecords) {
JSONObject json = JSON.parseObject(record.value);
List> list = new ArrayList<>;
Map map = new HashMap<>;
index = String.format(index, CalendarUtils.timeSpan2EsDay(json.getLongValue("_tm") * 1000L));

if (counter < 10) { LOG.info("Index : " + index); counter++; } for (String key : json.keySet) { if ("_uid".equals(key)) { map.put("uid", json.get(key)); } else { map.put(key, json.get(key)); } list.add(map); } EsUtils.write2Es(index, type, list); } } } } }

這裡消費的數據源就處理好了，接下來，開始ES的存儲，實現代碼如下所示：

public class EsUtils {

private static TransportClient client = null;

static {
if (client == null) {
client = new PreBuiltTransportClient(Settings.EMPTY);
}
String root = System.getProperty("user.dir") + "/conf/";
String path = SystemConfigUtils.getProperty("kafka.x.plugins.exec.path");
Configuration conf = Configuration.from(new File(root + path));
List