一、场景

将指定kafka的一个topic的数据实时入库到elasticsearch中。

二、环境

kafka:0.10.2

elasticsearch:5.6.1

三、应用

3.1、概述

Origins:Kafka Consumer

Destiations:ElasticSearch

3.2、注意事项

1、若只读取一个topic的数据，origins使用Kafka Consumer；若要同时读取多个topic，则使用Kafka Multitopic。

2、kafka若要实现读取历史数据，要添加auto.offset.reset=earliest

3.3、流程

3.3.1、Origins配置

（1）选择源

Origins选择Kafka Consumer

（2）配置

General

Kafka

 

Data Format

 

3.3.2、Destiations配置

（1）选择目的

Destiations选择ElasticSearch

（2）配置

General

ElasticSearch

3.3.3、Pipeline

(1)

(2)

General

Error Records

 

 

3.4、效果