使用Apache Flink和Apache Ignit进行数据流分析

在本文中，我们将讨论如何使用Apache Flink和Apache Ignite构建数据流应用程序。构建数据流应用程序可以以优化和容错的方式将大量有限和无限量的数据提取到Ignite集群中。数据摄取率非常高，每秒可以扩展到数百万个事件。

Apache IgniteSink提供了一个流连接器，用于将Flink数据注入Ignite缓存，连接器将其输入数据发送到Ignite缓存。

需要注意的关键特性是Apache Flink和Apache Ignite提供的性能和扩展。Apache Flink可以处理无界和有界数据集，并且可以大规模运行有状态流应用程序。应用程序计算以集群形式分布和同时执行。Apache Flink还针对任务的本地状态访问进行了优化，并确定了本地状态的持久性。Apache Ignite提供流式处理功能，允许在内存数据网格中大规模地提取数据。

本文应用 github

设置：下载并启动Flink

从下载页面下载二进制文件，可以选择任何您喜欢的Hadoop / Scala组合。如果您打算只使用本地文件系统，任何Hadoop版本都可以正常工作。转到下载目录。

解压缩下载的档案

$ cd ~ / Downloads ＃转到下载目录

$ tar xzf flink - * .tgz ＃解压缩下载的档案

$ cd flink-1.5.0

启动本地Flink群集：

$ ./bin/start-cluster.sh# 启动Flink

检查Dispatcher的Web前端， http：// localhost：8081 ，确保一切正常运行。Web前端应输出单个可用的TaskManager实例。还可以通过检查logs目录中的日志文件来验证系统是否正在运行：$ tail log / flink - * - standalonesession - * .log

下载Kafka 从下载页面（[url]https://kafka.apache.org/downloads[/url]）下载二进制文件。

您可以选择Apache Kafka 

0.10.2.2

版本

scala 2.11

。

1. 启动Zookeeper Server

/bin/zookeeper-server-start.sh ./config/zookeeper.properties

2. 启动经纪人

./bin/kafka-server-start.sh ./config/server.properties

3.创建主题“mytopic”

$ ./bin/kafka-topics.sh --create --topic mytopic --zookeeper localhost:2181 --partitions 1 --replication-factor 1

4.查看主题“mytopic”

$ ./bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic mytopic

5. 在主题中产生一些东西（写点东西然后输入）

$ ./bin/kafka-console-producer.sh --topic mytopic --broker -list localhost:9092

6. 使用Console Producer消费主题：

$ ./bin/kafka-console-consumer.sh --topic mytopic --zookeeper localhost:2181

7. 下载Apache Ignite

在编写本文档时，

master

分支中提供了对Flink集群中的数据流应用程序的IgniteSink支持  。

$ git clone https://github.com/apache/ignite

8. 构建Apache Ignite

$ mvn clean package install -DskipTests

9. 构建Flink程序  本文应用 github

$ mvn clean package

10. 提交Flink程序

$ ./bin/flink run streamers-1.0-SNAPSHOT.jar

11. 在主题中产生一些东西（写点东西并点击回车）

$ ./bin/kafka-console-producer.sh --topic mytopic --broker -list localhost:9092

只要文字不断输入，out文件就会在每个时间窗口的末尾打印计数，例如：

$ tail -f log/flink-*-taskexecutor-*.outlorem : 1bye : 1ipsum : 4

12. Ignite REST服务要检查缓存键值，可以使用Ignite REST服务

$ curl -X GET http://localhost:8080/ignite\?cmd\=getall\&k1\=jam\&cacheName\=testCache

13. 扫描缓存要检查Ignite缓存中的所有密钥，可以使用以下REST服务

$ curl -X GET http://localhost:8080/ignite?cmd=qryscanexe&pageSize=10&cacheName=testCache

总结

我们介绍了如何使用Apache Flink和Apache Ignite构建简单的数据流应用程序，并创建流处理拓扑，允许以分布式，可扩展和容错的方式进行数据流，以便处理由数百万个事件组成的无界数据集。

[该贴被banq于2018-10-14 19:58修改过]