一、各組件版本
| 組件 | 版本 |
|---|---|
| Flink | 1.16.1 |
| kafka | 2.0.0 |
| Logstash | 6.5.4 |
| Elasticseach | 6.3.1 |
| Kibana | 6.3.1 |
??針對(duì)按照?志?件??滾動(dòng)?成?件的?式，可能因?yàn)槟硞€(gè)錯(cuò)誤的問題，需要看好多個(gè)?志?件，還有Flink on Yarn模式提交Flink任務(wù)，在任務(wù)執(zhí)行完畢或者任務(wù)報(bào)錯(cuò)后container會(huì)被回收從而導(dǎo)致日志丟失，為了方便排查問題可以把?志?件通過KafkaAppender寫?到kafka中，然后通過ELK等進(jìn)??志搜索甚?是分析告警。
二、Flink配置將日志寫入Kafka
2.1 flink-conf.yaml增加下面兩行配置信息
env.java.opts.taskmanager: -DyarnContainerId=$CONTAINER_IDenv.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}
# 提交到y(tǒng)arn的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模式提交任務(wù)前期準(zhǔn)備
2.3.1 需要根據(jù)kafka的版本在flink/lib?錄下放?kafka-clients的jar包
2.3.2 kafka處于啟動(dòng)狀態(tài)
2.3.3 Flink Standalone集群
# 根據(jù)kafka的版本放?kafka-clients
kafka-clients-3.1.0.jar
# jackson對(duì)應(yīng)的jar包
jackson-annotations-2.13.3.jar
jackson-core-2.13.3.jar
jackson-databind-2.13.3.jar
2.4 Flink on yarn任務(wù)提交案例
/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òng)腳本需要加入這個(gè)參數(shù)，日志才能采集到任務(wù)名稱(-D env.java.opts="-DflinkJobName=xxx")
??消費(fèi)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進(jìn)?查詢或搜索
三、LogStash部署
??部署過程略，網(wǎng)上都有
??需要注意Logstash內(nèi)部kafka-clients和Kafka版本兼容問題,需要根據(jù)Kafka版本選擇合適的Logstash版本
??將以下內(nèi)容寫?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啟動(dòng)：
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部署
??部署過程略，網(wǎng)上都有
??注意需要用root用戶以外的用戶啟動(dòng)Elasticsearch
??啟動(dòng)腳本：
Su elasticsearchlogtest
elasticsearch-6.3.1/bin/elasticsearch
??Windows訪問ES客戶端推薦使用ElasticHD，本地運(yùn)行后可以直連ES
五、Kibana部署
??部署過程略，網(wǎng)上都有
??啟動(dòng)腳本:
??kibana-6.3.1-linux-x86_64/bin/kibana
5.1 配置規(guī)則
文章來源：http://www.zghlxwxcb.cn/news/detail-743592.html
5.2 日志分析
文章來源地址http://www.zghlxwxcb.cn/news/detail-743592.html
到了這里，關(guān)于Flink日志采集-ELK可視化實(shí)現(xiàn)的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
