GeoEvent使用kafka作为输入

做了一回搬运工和实践者。原教程地址：http://www.arcgis.com/home/ite ... 0d671由GeoEvent团队提供。下载的教程附带数据、脚本等。
一、部署过程
Apache Kafka安装配置
1.安装ArcGIS Server10.5.1 和ArcGIS GeoEvent Server10.5.1
2.下载并安装jdk8 网址:http://www.oracle.com/technetw ... .html
3.下载Apache Kafka version 0.10.1.1 网址:https://www.apache.org/dyn/clo ... 1.tgz
4.解压并移动到c:\kafka目录
5.设置KAFKA_HOME系统变量，值为C:\kafka\kafka_2.11-0.10.1.1
注意:kafka的ZooKeeper端口需要更改，因为和GeoEvent使用的Zookeeper端口冲突。
6.打开C:\kafka\Kafka_2.11-0.10.1.1\config
7.更改zookeeper.properties文件中的clientPort属性从2181变成2183；server.properties文件中的zookeeper.connect属性中的2181改为2183。
8.运行教程中\scripts文件夹下的run-zookeeper命令
9.运行教程中\scprits文件夹下的run-kafka命令
10.运行教程中\scprits文件夹下的create-topics命令，默认创建一个Flights topic
添加kafka传输到GeoEvent Server
1.打开ArcGIS GeoEvent Manager然后打开站点>组件>传输
2.点击添加本地传输，浏览到教程中的\component文件夹中的kafka-transport-10.5.0.jar
3.点击打开
4.点击添加传输到GeoEvent Server，如果添加成功，将出现一个消息。
5.检查列表中是否有Apache Kafka Inbound Transport 和 Apache Kafka Outbound Transport


添加传输
注意:如果kafka传输出现在传输列表中，则转到下一节。如果在导入之后，kafka传输没有出现在传输列表中，请按照以下步骤执行：
1,进入\deploy文件夹，将文件夹中的kafka-transport-10.5.0.jar文件后缀更改为kafka-transport-10.5.0.jar.OLD文件。注意后缀名是.jar.OLD，经测试.OLD不可行，.jar.OLD才可以。2,然后再复制一个kafka-transport-10.5.0.jar文件到\deploy目录下，这时目录中存在一个kafka-transport-10.5.0.jar和一个kafka-transport-10.5.0.jar.OLD
3,在GeoEvent Manager的组件传输页面中https://ip:6143/geoevent/manager/components.html，刷新此网页，就会出现kafka出站和入站的两个传输类型
导入一个GeoEvent Server配置
1.打开ArcGIS GeoEvent Manager然后打开站点>GeoEvent>配置存储 点击导入配置。
2.浏览到教程中\configurations文件夹中的kafka_config.xml。
3.点击导入配置
4.访问站点>GeoEvent>连接器，搜索kafka。可以看到Publish Text to Kafka 和 Receive Text from Kafka两个连接器出现。


导入连接器配置
配置输入连接器来接收kafka的消息
1.访问服务>输入，点击添加输入
2.选择Receive Text from Apache Kafka输入连接器类型

kafka输入连接器
3.现在，如下设定Receive Text from Apache Kafka类型的输入


输入的设置
4.点击保存
5.访问 服务>输出，添加输出
6.选择Write to a JSON File类型
7.注册并使用文件夹c:/shuju，其余参数按默认值。
8.点击保存
9.访问服务>监控页面，确认新的输入和输出已存在。
10.访问 服务>GeoEvent服务，点击添加服务
12.命名为KafkaInDataLogger点击创建。
13.拖拽kafka-text-in 输入和 file-json-out输出，并使前者指向后者。

服务详情
14.点击发布
15.用记事本打开教程中\simulations文件夹中的Flights.csv文件,并复制第一行。
16.运行\scrpits下的run-producer-test.bat命令。
17.在run-producer-test命令行中粘贴复制的数据，敲击回车键。


kafka生产者
18.在c:/shuju文件夹中，将看到此数据。


GeoEvent成功接收并输出
19.在GeoEvent Manager中将看到，count增加了1。


count增加
20.在GeoEvent定义中，找到Flight，点击编辑。


增加了一个topic同名定义
21.修改字段类型和标签，如下：


编辑定义
至此，GeoEvent成功接入kafka。
二、测试过程
主要测试此环境在重启时的稳定性。
重启kafka，发送消息，GeoEvent依然正常收到
1.在kafka命令窗口ctrl+c，Y停止。
2.在zookeeper命令窗口ctrl+c，Y停止。


kafka停止
3.运行教程中\scripts文件夹下的run-zookeeper命令
4.运行教程中\scprits文件夹下的run-kafka命令
5.运行\scrpits下的run-producer-test.bat命令。
6.在run-producer-test命令行中粘贴复制的一条数据，敲击回车键。
7.GeoEvent监控页面此服务输入输出数量+1，c:/shuju下的json文件中增加了相同内容的json数据。
重启GeoEvent，发送消息，GeoEvent依然正常收到
在windows服务中停止，然后开启GeoEvent服务。其余步骤同上。
重启计算机，发送消息，GeoEvent依然正常收到
重启计算机，其余步骤基本一致，需要注意的是，若重启计算机后GeoEvent服务自动启动，后启动zookeeper和kafka。那么有可能会出现输入找不到的情况。
将ArcGIS GeoEvent Server服务设置为手动启动。并且在每次此服务启动前，确保输入端的kafka和zookeeper已运行。即可解决。
只要保证kafka和GeoEvent启动的顺序，kafka输入没有丢失情况。
需要注意的关键步骤如下:
1.一定要在kafka运行状态下并且创建完topic后，再在GeoEvent中导入kafka传输。
2.导入配置后，连接器自动出现，而不是手动创建连接器。
3.注意kafka的版本为Kafka_2.11-0.10.1.1。
4.重启时一定要保证kafka和zookeeper已运行，再启动ArcGIS GeoEvent。
文章来源：http://www.jianshu.com/p/f3c75dc5b3fc