kafka-＞Flink-＞ElasticSearch(Java形式)

2024-05-26 10:55:06

概述

本文主要是对[1]中内容的复现

环境

开源组件	版本
KAFKA	2.5.0
Flink	1.6.0
Zookeeper	3.6.0
ElasticSearch	7.10.1
KIBANA	7.10.1

因为ElasticSearch7的写法似乎不兼容 ES6的，所以代码中依然是ES6的写法。

代码中依然是ElasticSearch6的驱动
这个实验注意，不要追求太新的版本，最新版本Flink1.12的kafka驱动依赖包都还没有开发出来。

流程图

lateLog用来保存侧边流输出的迟到的数据

ElasticSearch准备工作

KIBANA操作	讲人话	具体命令
删除原有的index索引	删除原有的数据库	curl -XDELETE 'Desktop:9201/auditindex'
新建index	新建数据库	curl -XPUT 'http://Desktop:9201/auditindex?pretty'
创建type的mapping信息	新建表格的字段信息	curl -H "Content-Type: application/json" -XPOST 'http://Desktop:9201/auditindex/audittype/_mapping?include_type_name=true' -d ' { "audittype":{ "properties":{ "area":{"type":"keyword"}, "type":{"type":"keyword"}, "count":{"type":"long"}, "time":{"type":"date","format": "yyyy-MM-dd HH:mm:ss"} } } } '

上述命令尤其是最后一个，不要直接粘贴到terminal中运行，而要写入一个bash脚本中再运行

代码与运行流程

https://gitee.com/appleyuchi/Flink_Code/tree/master/flink清洗数据案例/FlinkProj

ElasticSearch查看接收到的数据

http://desktop:9201/auditindex/_search?pretty=true

打开后可以看到一个大大的JSON

KIBANA设置时区

进入http://desktop:5601/app/management/kibana/settings

然后下面的dateFormat设置为

Etc/UTC

然后点击上图右下角的Save Changes

KIBANA可视化

可视化效果如下:

注意

[1]中的设置是area,这里复现的用的是_id

因为KIBANA操作的时候没有找到area,只有area.keyword

版本差异，暂时无法解决。

本实验相关的JPS进程

106851 TaskManagerRunner
85543 NailgunRunner
84330 ZooKeeperMain
81133 NameNode
87055 Kafka
106575 StandaloneSessionClusterEntrypoint
82193 NodeManager
81617 SecondaryNameNode
38320 RemoteMavenServer
81968 ResourceManager
36945 Main
83639 Elasticsearch
81078 QuorumPeerMain
89016 Launcher
89019 DataReport
126490 Jps
81341 DataNode
87391 kafkaProducerDataReport

Reference:

[1]【20】Flink 实战案例开发（二）：数据报表

[2]kibana7.10.1基本操作(饼图+直方图)

kafka-＞Flink-＞ElasticSearch(Java形式)相关推荐

基于Java、Kafka、ElasticSearch的搜索框架的设计与实现
Jkes是一个基于Java.Kafka.ElasticSearch的搜索框架.Jkes提供了注解驱动的JPA风格的对象/文档映射,使用rest api用于文档搜索. 项目主页:https://gith ...
【转】Filebeat+Kafka+Logstash+ElasticSearch+Kibana 日志采集方案
前言 Elastic Stack 提供 Beats 和 Logstash 套件来采集任何来源.任何格式的数据.其实Beats 和 Logstash的功能差不多,都能够与 Elasticsearch 产 ...
flink的java api_Flink 流处理API之二
1.Transform 1.1 map val streamMap = stream.map { x => x * 2 } 1.2 flatmap flatMap的函数签名:def flatMa ...
ElasticSearch Java High level Rest Client 官方文档中文翻译(一)
ElasticSearch Java High level Rest Client 官方文档中文翻译一纯粹记录自己在看官网的es rest high level api 时的翻译笔记,可以对照着官 ...
Elasticsearch Java API 分组、聚合、嵌套相关查询
Elasticsearch Java API 分组.聚合.嵌套相关查询翼支付监控系统正使用es做后端存储,这边我们是将日志计算处理过后的数据通过kafka储存到es.选择用es作为数据储存端是考虑到 ...
Elasticsearch8.0版本中Elasticsearch Java API Client客户端的基本使用方法
关于Elasticsearch Java API Client客户端如何连接以及如何对索引和文档进行基本的增删改查操作请查看我的上一篇博文:Elasticsearch RestHighLevelCli ...
基于Filebeat+Kafka+Flink仿天猫双11实时交易额
基于Filebeat+Kafka+Flink仿天猫双11实时交易额 1. 写在前面在大数据实时计算方向,天猫双11的实时交易额是最具权威性的,当然技术架构也是相当复杂的,不是本篇博客的简单实现,因为 ...
Elasticsearch Java API 很全的整理以及架构剖析
Elasticsearch 的API 分为 REST Client API(http请求形式)以及 transportClient API两种.相比来说transportClient API效率更高, ...
kafka+flink集成实例
kafka+flink集成 1.目的 1.1 Flink简介 Apache Flink是一个面向数据流处理和批量数据处理的可分布式的开源计算框架,它基于同一个Flink流式执行模型(streaming ...
Elasticsearch Java API 6.2（java client）
前言本节描述了Elasticsearch提供的Java API,所有的Elasticsearch操作都使用客户端对象执行,所有操作本质上都是完全异步的(要么接收监听器,要么未来返回). 此外,客户端 ...

最新文章

热门文章