Flink日志采集-ELK可视化实现

这篇具有很好参考价值的文章主要介绍了Flink日志采集-ELK可视化实现。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

一、各组件版本

组件 版本
Flink 1.16.1
kafka 2.0.0
Logstash 6.5.4
Elasticseach 6.3.1
Kibana 6.3.1

  针对按照⽇志⽂件⼤⼩滚动⽣成⽂件的⽅式,可能因为某个错误的问题,需要看好多个⽇志⽂件,还有Flink on Yarn模式提交Flink任务,在任务执行完毕或者任务报错后container会被回收从而导致日志丢失,为了方便排查问题可以把⽇志⽂件通过KafkaAppender写⼊到kafka中,然后通过ELK等进⾏⽇志搜索甚⾄是分析告警。

二、Flink配置将日志写入Kafka

2.1 flink-conf.yaml增加下面两行配置信息

env.java.opts.taskmanager: -DyarnContainerId=$CONTAINER_ID
env.java.opts.jobmanager: -DyarnContainerId=$CONTAINER_ID

2.2 log4j.properties配置案例如下

##################################################################
#  Licensed to the Apache Software Foundation (ASF) under one
#  or more contributor license agreements.  See the NOTICE file
#  distributed with this work for additional information
#  regarding copyright ownership.  The ASF licenses this file
#  to you under the Apache License, Version 2.0 (the
#  "License"); you may not use this file except in compliance
#  with the License.  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
# limitations under the License.
##################################################################
# Allows this configuration to be modified at runtime. The file will be checked every 30 seconds.
monitorInterval=30

# This affects logging for both user code and Flink
#rootLogger.appenderRef.file.ref = MainAppender
rootLogger.level = INFO
rootLogger.appenderRef.kafka.ref = Kafka
rootLogger.appenderRef.file.ref = RollingFileAppender

# Uncomment this if you want to _only_ change Flink's logging
#logger.flink.name = org.apache.flink
#logger.flink.level = INFO

# The following lines keep the log level of common libraries/connectors on
# log level INFO. The root logger does not override this. You have to manually
# change the log levels here.
logger.akka.name = akka
logger.akka.level = INFO
logger.kafka.name= org.apache.kafka
logger.kafka.level = INFO
logger.hadoop.name = org.apache.hadoop
logger.hadoop.level = INFO
logger.zookeeper.name = org.apache.zookeeper
logger.zookeeper.level = INFO
logger.shaded_zookeeper.name = org.apache.flink.shaded.zookeeper3
logger.shaded_zookeeper.level = INFO

# Log all infos in the given file
appender.rolling.name = RollingFileAppender
appender.rolling.type = RollingFile
appender.rolling.append = false
appender.rolling.fileName = ${sys:log.file}
appender.rolling.filePattern = ${sys:log.file}.%i
appender.rolling.layout.type = PatternLayout
appender.rolling.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n
appender.rolling.policies.type = Policies
appender.rolling.policies.size.type = SizeBasedTriggeringPolicy
appender.rolling.policies.size.size = 500MB
appender.rolling.strategy.type = DefaultRolloverStrategy
appender.rolling.strategy.max = 10

#appender.main.name = MainAppender
#appender.main.type = RollingFile
#appender.main.append = true
#appender.main.fileName = ${sys:log.file}
#appender.main.filePattern = ${sys:log.file}.%i
#appender.main.layout.type = PatternLayout
#appender.main.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n
#appender.main.policies.type = Policies
#appender.main.policies.size.type = SizeBasedTriggeringPolicy
#appender.main.policies.size.size = 100MB
#appender.main.policies.startup.type = OnStartupTriggeringPolicy
#appender.main.strategy.type = DefaultRolloverStrategy
#appender.main.strategy.max = ${env:MAX_LOG_FILE_NUMBER:-10}

# kafka
appender.kafka.type = Kafka
appender.kafka.name = Kafka
appender.kafka.syncSend = true
appender.kafka.ignoreExceptions = false
appender.kafka.topic = flink_logs
appender.kafka.property.type = Property
appender.kafka.property.name = bootstrap.servers
appender.kafka.property.value = xxx1:9092,xxx2:9092,xxx3:9092
appender.kafka.layout.type = JSONLayout
apender.kafka.layout.value = net.logstash.log4j.JSONEventLayoutV1
appender.kafka.layout.compact = true
appender.kafka.layout.complete = false

# Suppress the irrelevant (wrong) warnings from the Netty channel handler
#logger.netty.name = org.jboss.netty.channel.DefaultChannelPipeline
logger.netty.name = org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline
logger.netty.level = OFF

#通过 flink on yarn 模式还可以添加⾃定义字段
# 日志路径
appender.kafka.layout.additionalField1.type = KeyValuePair
appender.kafka.layout.additionalField1.key = logdir
appender.kafka.layout.additionalField1.value = ${sys:log.file}
# flink-job-name
appender.kafka.layout.additionalField2.type = KeyValuePair
appender.kafka.layout.additionalField2.key = flinkJobName
appender.kafka.layout.additionalField2.value = ${sys:flinkJobName}
# 提交到yarn的containerId
appender.kafka.layout.additionalField3.type = KeyValuePair
appender.kafka.layout.additionalField3.key = yarnContainerId
appender.kafka.layout.additionalField3.value = ${sys:yarnContainerId}

  上⾯的 appender.kafka.layout.type 可以使⽤ JSONLayout ,也可以⾃定义。

  ⾃定义需要将上⾯的appender.kafka.layout.type 和 appender.kafka.layout.value 修改成如下:

appender.kafka.layout.type = PatternLayout
appender.kafka.layout.pattern ={"log_level":"%p","log_timestamp":"%d{ISO8601}","log_thread":"%t","log_file":"%F","l
og_line":"%L","log_message":"'%m'","log_path":"%X{log_path}","job_name":"${sys:flink
_job_name}"}%n

2.3 基于Flink on yarn模式提交任务前期准备

2.3.1 需要根据kafka的版本在flink/lib⽬录下放⼊kafka-clients的jar包

Flink日志采集-ELK可视化实现,Flink,ElasticSearch,flink,elk,大数据

2.3.2 kafka处于启动状态

2.3.3 Flink Standalone集群

# 根据kafka的版本放⼊kafka-clients
kafka-clients-3.1.0.jar
# jackson对应的jar包
jackson-annotations-2.13.3.jar
jackson-core-2.13.3.jar
jackson-databind-2.13.3.jar

2.4 Flink on yarn任务提交案例

/root/software/flink-1.16.1/bin/flink run-application \
-t yarn-application \
-D yarn.application.name=TopSpeedWindowing \
-D parallelism.default=3 \
-D jobmanager.memory.process.size=2g \
-D taskmanager.memory.process.size=2g \
-D env.java.opts="-DflinkJobName=TopSpeedWindowing" \
/root/software/flink-1.16.1/examples/streaming/TopSpeedWindowing.jar

【注意】启动脚本需要加入这个参数,日志才能采集到任务名称(-D env.java.opts="-DflinkJobName=xxx")

  消费flink_logs案例

{
    instant: {
        epochSecond: 1698723428,
        nanoOfSecond: 544000000,
    },
    thread: 'flink-akka.actor.default-dispatcher-17',
    level: 'INFO',
    loggerName: 'org.apache.flink.runtime.rpc.akka.AkkaRpcService',
    message: 'Stopped Akka RPC service.',
    endOfBatch: false,
    loggerFqcn: 'org.apache.logging.slf4j.Log4jLogger',
    threadId: 68,
    threadPriority: 5,
    logdir: '/yarn/container-logs/application_1697779774806_0046/container_1697779774806_0046_01_000002/taskmanager.log',
    flinkJobName: 'flink-log-collect-test',
    yarnContainerId: 'container_1697779774806_0046_01_000002',
}

  ⽇志写⼊Kafka之后可以通过Logstash接⼊elasticsearch,然后通过kibana进⾏查询或搜索

三、LogStash部署

  部署过程略,网上都有

  需要注意Logstash内部kafka-clients和Kafka版本兼容问题,需要根据Kafka版本选择合适的Logstash版本

  将以下内容写⼊config/logstash-sample.conf ⽂件中

input {
	kafka {
		bootstrap_servers => ["xxx1:9092,xxx2:9092,xxx3:9092"] 
		group_id => "logstash-group"
		topics => ["flink_logs"] 
		consumer_threads => 3 
		type => "flink-logs" 
		codec => "json"
		auto_offset_reset => "latest"
	}
}

output {
	elasticsearch {
		hosts => ["xxx:9200"] 
		index => "flink-log-%{+YYYY-MM-dd}"
	}
}

  Logstash启动:

logstash-6.5.4/bin/logstash -f logstash-6.5.4/config/logstash-sample.conf 2>&1 >logstash-6.5.4/logs/logstash.log &

四、Elasticsearch部署

  部署过程略,网上都有

  注意需要用root用户以外的用户启动Elasticsearch

  启动脚本:

Su elasticsearchlogtest

elasticsearch-6.3.1/bin/elasticsearch

Flink日志采集-ELK可视化实现,Flink,ElasticSearch,flink,elk,大数据

  Windows访问ES客户端推荐使用ElasticHD,本地运行后可以直连ES
Flink日志采集-ELK可视化实现,Flink,ElasticSearch,flink,elk,大数据

五、Kibana部署

  部署过程略,网上都有

  启动脚本:

  kibana-6.3.1-linux-x86_64/bin/kibana

5.1 配置规则

Flink日志采集-ELK可视化实现,Flink,ElasticSearch,flink,elk,大数据
Flink日志采集-ELK可视化实现,Flink,ElasticSearch,flink,elk,大数据

5.2 日志分析

Flink日志采集-ELK可视化实现,Flink,ElasticSearch,flink,elk,大数据文章来源地址https://www.toymoban.com/news/detail-743592.html

到了这里,关于Flink日志采集-ELK可视化实现的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 基于Elasticsearch+Logstash+Kibana+Filebeat的日志收集分析及可视化

    sudo rm /var/lib/dpkg/lock* sudo dpkg --configure -a apt update tail -f /var/log/car.log 海量的业务应用,也带来了海量的日志数据,给业务应用的运维带来了新的挑战。例如,我们常用的网约车应用,单个平台的日订单数量可以达到上亿条,产生各种类型的日志数据,像用户业务方面会产生交

    2024年01月21日
    浏览(57)
  • ⼤数据采集和可视化系统实现

    前言 本篇博客用来记录 2022年春季大数据课程结课大作业(大三下,前8周,小组作业) 非常感谢小组同学的付出和积极配合! 了解Kafka、HDFS、MapReduce、Spark、HBase、Hive等组件在大数据体系结构中的角色,并通过本次综合实验对大数据技术在实际应用中的主要流程有初步的认

    2024年02月11日
    浏览(34)
  • ELK(Elasticsearch、Kibana、Logstash)以及向ES导入mysql数据库数据或CSV文件数据,创建索引和可视化数据

    地址:Past Releases of Elastic Stack Software | Elastic 在Products和version处分别选择需要下载的产品和版本,E(elasticsearch)L(logstash)K(kibana)三者版本必须相同 将下载好的elk分别解压到相同路径下 本文中elasticsearch=E=ES=es;L=logstash;K=kibana 一般情况下使用默认配置即可,下面对我的

    2024年02月15日
    浏览(59)
  • 史上最全从零搭建ELKB(Elasticsearch、Logstash、Kibana、Beat)分布式日志管理可视化平台之一

    ELKB(Elasticsearch、Logstash、Kibana、Beat的组合)是一套开源的分布式日志管理方案。凭借其闭环的日志处理流程、高效的检索性能、线性的扩展能力、较低的运维成本等特点,ELKB在最近几年迅速崛起,成为实时日志处理开源领域的首要选择。(https://cloud.tencent.com/developer/article/1143

    2024年01月19日
    浏览(82)
  • Filebeat+Kafka+ELK日志采集(五)——Elasticsearch

    1、下载 2、解压: 3、配置 启动Elasticsearch,进入/bin目录下 ./elasticsearch ,不出意外的外会报以下错误: 报错1:能打开的文件或进程数太低。 解决方法: 修改/etc/security/limits.conf 配置文件,添加配置如下: 报错2: 解决方法: 修改/etc/sysctl.conf 配置文件,添加配置如下: 修

    2024年02月05日
    浏览(59)
  • Suricata+ELK 8.4.3(docker)可视化

    主机1安装elk 主机2安装suricata+filebeat 主机1、主机2均为Ubuntu18.04 准备docker镜像 创建文件夹用于docker映射 编辑文件 /etc/sysctl.conf ,在文件最后一行添加 执行命令令配置生效。该配置限制了一个进程可以拥有的虚拟内存区域的数量,如果太小会导致容器无法启动,所以提前进行

    2024年01月25日
    浏览(37)
  • 温度采集数据可视化

    实现功能: 实时显示近3分钟温度随时间的变化曲线 实时显示当前温度、温度均值、温度波动 根据温度数据实时调整纵坐标轴范围 使用的工具: 硬件使用树莓派4B,烧录的是Raspberry Pi OS系统 使用Qt Creator软件中UI Designer工具设计界面 使用python语言编写逻辑,主要使用python中的

    2024年02月07日
    浏览(40)
  • 【Python】Pyside2 可视化实现:每秒复制源文件一行到目标文件并打印日志

    博主在某个项目中,需要模拟每秒钟生成一行数据,所以有了该博客的想法,其中有线程的内容,为了防止主界面卡住

    2024年01月18日
    浏览(91)
  • 大数据毕业设计:基于python商品数据采集分析可视化系统 淘宝数据采集 大数据 大屏可视化(附源码+论文)✅

    博主介绍:✌全网粉丝10W+,前互联网大厂软件研发、集结硕博英豪成立工作室。专注于计算机相关专业毕业设计项目实战6年之久,选择我们就是选择放心、选择安心毕业✌感兴趣的可以先收藏起来,点赞、关注不迷路✌ 毕业设计:2023-2024年计算机专业毕业设计选题汇总(建议

    2024年02月03日
    浏览(54)
  • 基于python商品数据采集分析可视化系统 淘宝数据采集 大数据 大屏可视化(附源码+论文)大数据毕业设计✅

    毕业设计:2023-2024年计算机专业毕业设计选题汇总(建议收藏) 毕业设计:2023-2024年最新最全计算机专业毕设选题推荐汇总 🍅 感兴趣的可以先收藏起来,点赞、关注不迷路,大家在毕设选题,项目以及论文编写等相关问题都可以给我留言咨询,希望帮助同学们顺利毕业 。

    2024年02月02日
    浏览(58)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包