flume常见问题

编程问答 2024-04-20

问题描述

  • 报错1：执行到如下就不执行了，没有显示Successfully registered new MBean. 
[root@slave1 bin]# /usr/local/software/flume-1.9.0/bin/flume-ng agent -n a1 -c /usr/local/software/flume-1.9.0/conf/ -f file-flume-kafka.conf -Dflume.root.logger=INFO，console
Info: Including Hadoop libraries found via (/usr/local/software/hadoop-2.9.2/bin/hadoop) for HDFS access
Info: Including HBASE libraries found via (/usr/local/software/hbase-2.1.0/bin/hbase) for HBASE access
Info: Including Hive libraries found via (/usr/local/software/hive-2.3.9) for Hive access
+ exec /usr/local/software/jdk1.8.0_181/bin/java -Xmx20m -Dflume.root.logger=INFO，console -cp '/usr/local/software/flume-2.9.2/contrib/capacity-scheduler/*.jar:/usr/local/software/hbase-2.1.0/conf:/usr/local/software/hive-2.3.9/lib/*' -Djava.library.path=:/usr/local/software/hadoop-2.9.2/lib/native:/usr/local/software/hadoop-2.9.2/lib/native org.apache.flume.node.Application -n a1 -f file-flume-kafka.conf
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/usr/local/software/flume-1.9.0/lib/slf4j-log4j12-1.7.25.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/usr/local/software/hadoop-2.9.2/share/hadoop/common/lib/slf4j-log4j12-1.7.25.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/usr/local/software/hive-2.3.9/lib/log4j-slf4j-impl-2.6.2.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
log4j:WARN No appenders could be found for logger (org.apache.flume.util.SSLUtil).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
  • 解决方案
# 执行的命令中包含中文逗号，改为英文逗号后尝试
/usr/local/software/flume-1.9.0/bin/flume-ng agent -n a1 -c /usr/local/software/flume-1.9.0/conf/ -f file-flume-kafka.conf -Dflume.root.logger=INFO,console

# 成功，显示如下
2023-12-20 05:20:23,694 (lifecycleSupervisor-1-1) [INFO - org.apache.flume.instrumentation.MonitoredCounterGroup.register(MonitoredCounterGroup.java:119)] Monitored counter group for type: SOURCE, name: r1: Successfully registered new MBean.
2023-12-20 05:20:23,694 (lifecycleSupervisor-1-1) [INFO - org.apache.flume.instrumentation.MonitoredCounterGroup.start(MonitoredCounterGroup.java:95)] Component type: SOURCE, name: r1 started
  • 报错2：Unable to load channel type: org.apache.flume.channel.kafka.Kafkachannel 
org.apache.flume.FlumeException: Unable to load channel type: org.apache.flume.channel.kafka.Kafkachannel, class: org.apache.flume.channel.kafka.Kafkachannel
	at org.apache.flume.channel.DefaultChannelFactory.getClass(DefaultChannelFactory.java:68)
	at org.apache.flume.node.AbstractConfigurationProvider.getOrCreateChannel(AbstractConfigurationProvider.java:236)
	at org.apache.flume.node.AbstractConfigurationProvider.loadChannels(AbstractConfigurationProvider.java:200)
	at org.apache.flume.node.AbstractConfigurationProvider.getConfiguration(AbstractConfigurationProvider.java:104)
	at org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable.run(PollingPropertiesFileConfigurationProvider.java:145)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.ClassNotFoundException: org.apache.flume.channel.kafka.Kafkachannel
	at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
	at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
	at java.lang.Class.forName0(Native Method)
	at java.lang.Class.forName(Class.java:264)
	at org.apache.flume.channel.DefaultChannelFactory.getClass(DefaultChannelFactory.java:66)
	... 11 more
  • 解决方案
# 没有找到
a1.channels.c1.type = org.apache.flume.channel.kafka.Kafkachannel
# 仔细对比后，发现应该改为如下
a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
  • 错误3：Interceptor.Type not specified for i1 
2023-12-20 05:15:14,436 (conf-file-poller-0) [ERROR - org.apache.flume.channel.ChannelProcessor.configureInterceptors(ChannelProcessor.java:106)] Type not specified for interceptor i1
2023-12-20 05:15:14,436 (conf-file-poller-0) [ERROR - org.apache.flume.node.AbstractConfigurationProvider.loadSources(AbstractConfigurationProvider.java:355)] Source r1 has been removed due to an error during configuration
org.apache.flume.FlumeException: Interceptor.Type not specified for i1
	at org.apache.flume.channel.ChannelProcessor.configureInterceptors(ChannelProcessor.java:107)
	at org.apache.flume.channel.ChannelProcessor.configure(ChannelProcessor.java:82)
	at org.apache.flume.conf.Configurables.configure(Configurables.java:41)
	at org.apache.flume.node.AbstractConfigurationProvider.loadSources(AbstractConfigurationProvider.java:342)
	at org.apache.flume.node.AbstractConfigurationProvider.getConfiguration(AbstractConfigurationProvider.java:105)
	at
  • 解决方案：这2行要么同时放开，要么同时注释
#a1.sources.r1.interceptors = i1
#a1.sources.r1.interceptors.i1.type = com.test.flume.interceptors.TimestampAndTableNameInterceptor$Builder
  • 报错4：例如需要使用flume从kafka导入到hdfs，执行到一半不执行了 
[root@slave1 flume-1.9.0]# bin/flume-ng agent -n a1 -c conf/ -f /opt/software/bin/kafka_to_hdfs_db.conf -Dflume.root.logger=INFO,console
Info: Including Hadoop libraries found via (/usr/local/software/hadoop-2.9.2/bin/hadoop) for HDFS access
Info: Including HBASE libraries found via (/usr/local/software/hbase-2.1.0/bin/hbase) for HBASE access
Info: Including Hive libraries found via (/usr/local/software/hive-2.3.9) for Hive access
+ exec /usr/local/software/jdk1.8.0_181/bin/java -Xmx20m -Dflume.root.logger=INFO,console -cp '/usr/local/software/flume-1.9.0/conf:/usr/local/software/flume-
SLF4J: Class path contains multiple SLF4J bindings.
2023-12-20 00:05:24,448 (lifecycleSupervisor-1-0) [INFO - org.apache.flume.node.PollingPropertiesFileConfigurationProvider.start(PollingPropertiesFileConfigurationProvider.java:62)] Configuration provider starting
2023-12-20 00:05:25,324 (lifecycleSupervisor-1-4) [INFO - org.apache.kafka.common.config.AbstractConfig.logAll(AbstractConfig.java:279)] ConsumerConfig values: 
	auto.commit.interval.ms = 5000
	auto.offset.reset = latest
	bootstrap.servers = [slave2:9092, slave3:9092]
	check.crcs = true
	client.id = 
	connections.max.idle.ms = 540000
	default.api.timeout.ms = 60000
	enable.auto.commit = false
	exclude.internal.topics = true
	fetch.max.bytes = 52428800
	fetch.max.wait.ms = 500
	fetch.min.bytes = 1
	group.id = flume
	heartbeat.interval.ms = 3000
	interceptor.classes = []
	internal.leave.group.on.close = true
	isolation.level = read_uncommitted
	key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
	max.partition.fetch.bytes = 1048576
	max.poll.interval.ms = 300000
	max.poll.records = 500
	metadata.max.age.ms = 300000
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
	receive.buffer.bytes = 65536
	reconnect.backoff.max.ms = 1000
	reconnect.backoff.ms = 50
	request.timeout.ms = 30000
	retry.backoff.ms = 100
	sasl.client.callback.handler.class = null
	sasl.jaas.config = null
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	sasl.kerberos.service.name = null
	sasl.kerberos.ticket.renew.jitter = 0.05
	sasl.kerberos.ticket.renew.window.factor = 0.8
	sasl.login.callback.handler.class = null
	sasl.login.class = null
	sasl.login.refresh.buffer.seconds = 300
	sasl.login.refresh.min.period.seconds = 60
	sasl.login.refresh.window.factor = 0.8
	sasl.login.refresh.window.jitter = 0.05
	sasl.mechanism = GSSAPI
	security.protocol = PLAINTEXT
	send.buffer.bytes = 131072
	session.timeout.ms = 10000
	ssl.cipher.suites = null
	ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
	ssl.endpoint.identification.algorithm = https
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLS
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS
	value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer

2023-12-20 00:05:25,391 (lifecycleSupervisor-1-4) [INFO - org.apache.kafka.common.utils.AppInfoParser$AppInfo.<init>(AppInfoParser.java:109)] Kafka version : 2.0.1
2023-12-20 00:05:25,660 (PollableSourceRunner-KafkaSource-r1) [INFO - org.apache.flume.source.kafka.SourceRebalanceListener.onPartitionsAssigned(KafkaSource.java:648)] topic topic_db - partition 0 assigned.
2023-12-20 00:05:25,687 (PollableSourceRunner-KafkaSource-r1) [INFO - org.apache.kafka.clients.consumer.internals.Fetcher.resetOffsetIfNeeded(Fetcher.java:601)] [Consumer clientId=consumer-1, groupId=flume] Resetting offset for partition topic_db-0 to offset 35368.

  • 解决方案：ctrl+c退出时会显示具体的错误

  • 报错5：使用flume执行conf配置文件时报错如下

2023-12-21 05:45:47,247 (PollableSourceRunner-KafkaSource-r1) [INFO - org.apache.kafka.clients.consumer.internals.Fetcher.resetOffsetIfNeeded(Fetcher.java:601)] [Consumer clientId=consumer-1, groupId=flume] Resetting offset for partition topic_db-0 to offset 117032.
Exception in thread "PollableSourceRunner-KafkaSource-r1" java.lang.NoClassDefFoundError: com/alibaba/fastjson/JSONObject
	at com.test.flume.interceptors.TimestampAndTableNameInterceptor.intercept(TimestampAndTableNameInterceptor.java:22)
	at com.test.flume.interceptors.TimestampAndTableNameInterceptor.intercept(TimestampAndTableNameInterceptor.java:35)
	at org.apache.flume.interceptor.InterceptorChain.intercept(InterceptorChain.java:62)
	at org.apache.flume.channel.ChannelProcessor.processEventBatch(ChannelProcessor.java:148)
	at org.apache.flume.source.kafka.KafkaSource.doProcess(KafkaSource.java:311)
	at org.apache.flume.source.AbstractPollableSource.process(AbstractPollableSource.java:60)
	at org.apache.flume.source.PollableSourceRunner$PollingRunner.run(PollableSourceRunner.java:133)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.ClassNotFoundException: com.alibaba.fastjson.JSONObject
	at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
	at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
	... 8 more

Exception in thread "PollableSourceRunner-KafkaSource-r1" java.lang.NoClassDefFoundError: com/alibaba/fastjson2/util/Wrapper
	at java.lang.ClassLoader.defineClass1(Native Method)
	at java.lang.ClassLoader.defineClass(ClassLoader.java:763)
	at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
	at java.net.URLClassLoader.defineClass(URLClassLoader.java:467)
	at java.net.URLClassLoader.access$100(URLClassLoader.java:73)
	at java.net.URLClassLoader$1.run(URLClassLoader.java:368)
	at java.net.URLClassLoader$1.run(URLClassLoader.java:362)
	at java.security.AccessController.doPrivileged(Native Method)
	at java.net.URLClassLoader.findClass(URLClassLoader.java:361)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
	at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
	at com.test.flume.interceptors.TimestampAndTableNameInterceptor.intercept(TimestampAndTableNameInterceptor.java:22)
	at com.test.flume.interceptors.TimestampAndTableNameInterceptor.intercept(TimestampAndTableNameInterceptor.java:35)
	at org.apache.flume.interceptor.InterceptorChain.intercept(InterceptorChain.java:62)
	at org.apache.flume.channel.ChannelProcessor.processEventBatch(ChannelProcessor.java:148)
	at org.apache.flume.source.kafka.KafkaSource.doProcess(KafkaSource.java:311)
	at org.apache.flume.source.AbstractPollableSource.process(AbstractPollableSource.java:60)
	at org.apache.flume.source.PollableSourceRunner$PollingRunner.run(PollableSourceRunner.java:133)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.ClassNotFoundException: com.alibaba.fastjson2.util.Wrapper
	at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
	at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
	... 20 more

  • 解决方案：去maven中央仓库下载jar包上传到服务器的/usr/local/software/flume-1.9.0/lib目录

  • 报错6：PollableSourceRunner-KafkaSource-r1

2023-12-21 05:57:25,244 (PollableSourceRunner-KafkaSource-r1) [ERROR - org.apache.flume.source.kafka.KafkaSource.doProcess(KafkaSource.java:330)] KafkaSource EXCEPTION, {}
org.apache.kafka.common.errors.InterruptException: java.lang.InterruptedException
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeThrowInterruptException(ConsumerNetworkClient.java:504)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:287)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:242)
	at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1247)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1187)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1154)
	at org.apache.flume.source.kafka.KafkaSource.doProcess(KafkaSource.java:216)
	at org.apache.flume.source.AbstractPollableSource.process(AbstractPollableSource.java:60)
	at org.apache.flume.source.PollableSourceRunner$PollingRunner.run(PollableSourceRunner.java:133)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.InterruptedException
	... 10 more
  • 解决方案
# 进入flume安装目录
cd /usr/local/software/flume-1.9.0/bin/
# 编辑
vim flume-ng
# 设置为如下
JAVA_OPTS="-Xmx1024m"

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）

相关问答

导入项目后报错问题
依赖报错 idea导入项目后依赖报错，解决方案：https://blog....
idea不能识别yaml文件
使用mybatis plus常见错误
错误1：代码生成器依赖和mybatis依赖冲突 启动项目时报错如下...
gradle常见问题与错误
错误1：gradle项目控制台输出为乱码 # 解决方案：https://bl...
Mybatis Plus传入参数0不起作用
错误还原：在查询的过程中，传入的workType为0时，该条件不起...
linux中make编译源码包失败
报错如下，gcc版本太低 ^ server.c:5346:31: 错误：‘struct...