flume采集kafka报错怎么解决

报错信息:
Source.java:120)] Event #: 0
2018-11-23 17:59:18,995 (PollableSourceRunner-KafkaSource-kaSource) [DEBUG - org.apache.flume.source.kafka.KafkaSource.process(KafkaSource.java:119)] Waited: 965
2018-11-23 17:59:18,995 (PollableSourceRunner-KafkaSource-kaSource) [DEBUG - org.apache.flume.source.kafka.KafkaSource.process(KafkaSource.java:120)] Event #: 0
2018-11-23 17:59:19,005 (PollableSourceRunner-KafkaSource-kaSource) [DEBUG - org.apache.flume.source.kafka.KafkaSource.process(KafkaSource.java:119)] Waited: 975
2018-11-23 17:59:19,005 (PollableSourceRunner-KafkaSource-kaSource) [DEBUG - org.apache.flume.source.kafka.KafkaSource.process(KafkaSource.java:120)] Event #: 0
2018-11-23 17:59:19,015 (PollableSourceRunner-KafkaSource-kaSource) [DEBUG - org.apache.flume.source.kafka.KafkaSource.process(KafkaSource.java:119)] Waited: 985
2018-11-23 17:59:19,015 (PollableSourceRunner-KafkaSource-kaSource) [DEBUG - org.apache.flume.source.kafka.KafkaSource.process(KafkaSource.java:120)] Event #: 0
2018-11-23 17:59:19,025 (PollableSourceRunner-KafkaSource-kaSource) [DEBUG - org.apache.flume.source.kafka.KafkaSource.process(KafkaSource.java:119)] Waited: 995
2018-11-23 17:59:19,025 (PollableSourceRunner-KafkaSource-kaSource) [DEBUG - org.apache.flume.source.kafka.KafkaSource.process(KafkaSource.java:120)] Event #: 0
2018-11-23 17:59:19,036 (PollableSourceRunner-KafkaSource-kaSource) [DEBUG - org.apache.flume.source.kafka.KafkaSource.process(KafkaSource.java:119)] Waited: 1006
2018-11-23 17:59:19,036 (PollableSourceRunner-KafkaSource-kaSource) [DEBUG - org.apache.flume.source.kafka.KafkaSource.process(KafkaSource.java:120)] Event #: 0
2018-11-23 17:59:19,036 (PollableSourceRunner-KafkaSource-kaSource) [ERROR - org.apache.flume.source.kafka.KafkaSource.process(KafkaSource.java:153)] KafkaSource EXCEPTION, {}
java.lang.NullPointerException
at org.apache.flume.instrumentation.MonitoredCounterGroup.increment(MonitoredCounterGroup.java:261)
at org.apache.flume.instrumentation.kafka.KafkaSourceCounter.incrementKafkaEmptyCount(KafkaSourceCounter.java:49)
at org.apache.flume.source.kafka.KafkaSource.process(KafkaSource.java:146)
at org.apache.flume.source.PollableSourceRunner$PollingRunner.run(PollableSourceRunner.java:139)
at java.lang.Thread.run(Thread.java:748)

--------------------------------------------

配置文件
kafkaLogger.sources = kaSource

kafkaLogger.channels = memoryChannel
kafkaLogger.sinks = kaSink

The channel can be defined as follows.

kafkaLogger.sources.kaSource.channels = memoryChannel
kafkaLogger.sources.kaSource.type= org.apache.flume.source.kafka.KafkaSource
kafkaLogger.sources.kaSource.zookeeperConnect=192.168.130.4:2181,192.168.130.5:2181,192.168.130.6:2181
kafkaLogger.sources.kaSource.topic=dwd-topic
kafkaLogger.sources.kaSource.groupId = 0

kafkaLogger.channels.memoryChannel.type=memory
kafkaLogger.channels.memoryChannel.capacity = 1000
kafkaLogger.channels.memoryChannel.keep-alive = 60

kafkaLogger.sinks.kaSink.type = elasticsearch
kafkaLogger.sinks.kaSink.hostNames = 192.168.130.6:9300
kafkaLogger.sinks.kaSink.indexName = flume_mq_es_d
kafkaLogger.sinks.kaSink.indexType = flume_mq_es
kafkaLogger.sinks.kaSink.clusterName = zyuc-elasticsearch
kafkaLogger.sinks.kaSink.batchSize = 100
kafkaLogger.sinks.kaSink.client = transport
kafkaLogger.sinks.kaSink.serializer = com.commons.flume.sink.elasticsearch.CommonElasticSearchIndexRequestBuilderFactory
kafkaLogger.sinks.kaSink.serializer.parse = com.commons.log.parser.LogTextParser
kafkaLogger.sinks.kaSink.serializer.formatPattern = yyyyMMdd
kafkaLogger.sinks.kaSink.serializer.dateFieldName = time
kafkaLogger.sinks.kaSink.channel = memoryChannel

1个回答

KafkaSource.process(KafkaSource.java:153)] KafkaSource EXCEPTION, {}
java.lang.NullPointerException
at org.apache.flume.instrumentation.MonitoredCounterGroup.increment(MonitoredCounterGroup.java:261)
at org.apache.flume.instrumentation.kafka.KafkaSourceCounter.incrementKafkaEmptyCount(KafkaSourceCounter.java:49)
at org.apache.flume.source.kafka.KafkaSource.process(KafkaSource.java:146)
at org.apache.flume.source.PollableSourceRunner$PollingRunner.run(PollableSourceRunner.java:139)
at java.lang.Thread.run(Thread.java:748)
这里的报错是flume1.6版本的一个bug这个bug在1.7版本得到了修复,如果没有打算升级flume版本的话,可以写给flume1.6的源码重新打jar包
https://issues.apache.org/jira/browse/FLUME-2672 这个官方给出的解决方案
将KafkaSourceCounter该类中的
private static final String[] ATTRIBUTES = {TIMER_KAFKA_COMMIT, TIMER_KAFKA_EVENT_GET};
修改为
private static final String[] ATTRIBUTES = {TIMER_KAFKA_COMMIT, TIMER_KAFKA_EVENT_GET, COUNTER_KAFKA_EMPTY};


dingwd

Csdn user default icon
上传中...
上传图片
插入图片
抄袭、复制答案,以达到刷声望分或其他目的的行为,在CSDN问答是严格禁止的,一经发现立刻封号。是时候展现真正的技术了!
其他相关推荐
用flume读取kafka数据到hdfs,source创建时报错Kafka topic must be specified

计划使用flume读取kafka的数据传送到hdfs上,结果错误如下 ![图片说明](https://img-ask.csdn.net/upload/202003/20/1584694939_566000.png) 但是我的kafka里确实有对应的topic,名字为topic_start,下面是我的配置文件 ![图片说明](https://img-ask.csdn.net/upload/202003/20/1584695029_280148.png) 不知道问题出在哪里,目前测试结果为kafka里的topic,flume都无法读取出来,求大佬帮忙解决一下这个问题。目前我毫无头绪,如果不行只能考录重新安装flume等办法了

Flume和kafka连接的问题

本人使用flume1.60版本和kafka0.8.2.2版本进行连接,配置如下: a0.sources.r1.type = xiaomu.flume.source.TailFileSource a0.sources.r1.filePath = /root/access2.txt a0.sources.r1.posiFile = /root/posi2.txt a0.sources.r1.interval = 2000 a0.sources.r1.charset = UTF-8 a0.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel a0.channels.c1.capacity = 1000 a0.channels.c1.transactionCapacity = 1000 a0.channels.c1.brokerList=slave1:9092,slave4:9092,slave3:9092 a0.channels.c1.topic=usertest3 a0.channels.c1.zookeeperConnect=slave2:2181,slave5:2181,slave6:2181 a0.channels.c1.parseAsFlumeEvent = false 但是我echo "xxx" >> access2.txt之后,在kafka那边就不一样了,比如我打xiaomu,就会出来两行,一行是xiaomum,第二行是空白,还有时候是一行但是开头有一个方框,如图所示: ![图片说明](https://img-ask.csdn.net/upload/201807/21/1532184292_313303.png) 这个怎么解决呀?求助各位大神了!

flume+kafka+hdfs 整合问题

本想搭建一个 flume+hdfs+kafka+storm+mysql 的日志实时分析和存储的系统,但是flume日志收集这块一直不通,查看flume的日志也没有报错,不知道该怎么解决了,求大家帮帮忙,贴出集群配置和配置文件如下: 共5台机器:node1~node5,其中node3~node5为日志收集的agent,node1~node2为flume的collector,最终存储两份,一份到kafka,一份到hdfs。 agent的配置文件如下: #def agent.sources = src_spooldir agent.channels = file memory agent.sinks = collector_avro1 collector_avro2 # sources agent.sources.src_spooldir.type = spooldir agent.sources.src_spooldir.channels = file memory agent.sources.src_spooldir.spoolDir = /data/flume/spoolDir agent.sources.src_spooldir.selector.type = multiplexing agent.sources.src_spooldir.fileHeader = true # channels agent.channels.file.type = file agent.channels.file.checkpointDir = /data/flume/checkpoint agent.channels.file.dataDirs = /data/flume/data agent.channels.memory.type = memory agent.channels.memory.capacity = 10000 agent.channels.memory.transactionCapacity = 10000 agent.channels.memory.byteCapacityBufferPercentage = 20 agent.channels.memory.byteCapacity = 800000 # sinks agent.sinks.collector_avro1.type = avro agent.sinks.collector_avro1.channel = file agent.sinks.collector_avro1.hostname = node1 agent.sinks.collector_avro1.port = 45456 agent.sinks.collector_avro2.type = avro agent.sinks.collector_avro2.channel = memory agent.sinks.collector_avro2.hostname = node2 agent.sinks.collector_avro2.port = 4545 collector端的配置文件如下: #def agent.sources = src_avro agent.channels = file memory agent.sinks = hdfs kafka # sources agent.sources.src_avro.type = avro agent.sources.src_avro.channels = file memory agent.sources.src_avro.bind = node1 agent.sources.src_avro.port = 45456 agent.sources.src_avro.selector.type = replicating # channels agent.channels.file.type = file agent.channels.file.checkpointDir = /data/flume/checkpoint agent.channels.file.dataDirs = /data/flume/data agent.channels.memory.type = memory agent.channels.memory.capacity = 10000 agent.channels.memory.transactionCapacity = 10000 agent.channels.memory.byteCapacityBufferPercentage = 20 agent.channels.memory.byteCapacity = 800000 # sinks agent.sinks.hdfs.type = hdfs agent.sinks.hdfs.channel = file agent.sinks.hdfs.hdfs.path = hdfs://node1/flume/events/%y-%m-%d/%H%M/%S agent.sinks.hdfs.hdfs.filePrefix = log_%Y%m%d_%H agent.sinks.hdfs.hdfs.fileSuffix = .txt agent.sinks.hdfs.hdfs.useLocalTimeStamp = true agent.sinks.hdfs.hdfs.writeFormat = Text agent.sinks.hdfs.hdfs.rollCount = 0 agent.sinks.hdfs.hdfs.rollSize = 1024 agent.sinks.hdfs.hdfs.rollInterval = 0 agent.sinks.kafka.type = org.apache.flume.sink.kafka.KafkaSink agent.sinks.kafka.channel = memory agent.sinks.kafka.kafka.topic = test agent.sinks.kafka.kafka.bootstrap.servers = node3:9092,node4:9092,node5:9092 agent.sinks.kafka.kafka.flumeBatchSize = 20 agent.sinks.kafka.kafka.producer.acks = 1 agent.sinks.kafka.kafka.producer.linger.ms = 1 agent.sinks.kafka.kafka.producer.compression.type = snappy 最终 hdfs和kafka都没有接收到数据。

关于flume和kafka结合效率的问题

最近做了个测试。是flume+kafka的。是读取文件夹的。31M的文件读了很长时间。大概20分钟。不知道什么原因。哪位大神知道啊。指导下。 下面是flume的配置 #agent section producer.sources = s producer.channels = c producer.sinks = r #source section #producer.sources.s.type = seq #producer.sources.s.channels = c producer.sources.s.type =spooldir producer.sources.s.spoolDir = /home/lb/data producer.sources.s.channels = c # Each sink's type must be defined producer.sinks.r.type = org.apache.flume.plugins.KafkaSink producer.sinks.r.metadata.broker.list=127.0.0.1:9092 producer.sinks.r.partition.key=1 producer.sinks.r.partitioner.class=org.apache.flume.plugins.SinglePartition producer.sinks.r.serializer.class=kafka.serializer.StringEncoder producer.sinks.r.request.required.acks=0 producer.sinks.r.max.message.size=1000000 producer.sinks.r.producer.type=async producer.sinks.r.custom.encoding=UTF-8 producer.sinks.r.custom.topic.name=lbTestToptic #Specify the channel the sink should use producer.sinks.r.channel = c # Each channel's type is defined. producer.channels.c.type = memory producer.channels.c.capacity = 100000 producer.channels.c.transactionCapacity = 10000 希望有知道原因的大神给个帮助啊。谢谢

能否用spark streaming和flume或kafka对实时网络数据进行检测

目前已经有一个训练好的机器学习分类模型,存在于HDFS上,可以对LibSVMFile格式的数据进行检测。它是对很多的一段时间内的流量数据(比如1s,很多个1s)提取特征训练之后得到的。 我们知道streaming是将输入流分成微切片,微切片能否可以是从pcap文件读取呢?因为提取特征包括训练模型的时候是需要对pcap文件操作的。 flume和kafka都是可以传输txt的,能不能传输pcap文件呢?要将输入的网络数据流像tcpdump一样可以存为pcap文件,又有像kafka一样的缓存功能可以用哪些技术呢? 最后就是能否用spark streaming利用分类模型对网络数据流进行提特征并预测,而且与防火墙联动,这在技术上是否可行?

kafka连接flume因为hostname的配置报错?

flume连接kafka的时候kafka的server.properties的hostname使用localhost可以收集到flume发送过来的数据,但是想远程连接服务器,所以将localhost改成了服务器的ip地址,就连接不上了。 flume中的example.conf中的地址以及kafka消费者的启动语句中的都已经改掉了。

flume增量采集动态日志

请教一个flume增量采集日志问题,谢谢。 遇到一个应用,它生成的日志形式比较特殊,每天产生一个带日期的前缀,并根据文件容量扩展带序号的后缀。 例如2019年11月18日会首先生成20191118.log001,然后当日志文件写满20MB后,会生成20191118.log002,每个文件最大容量20MB,依次类推增加后缀的数字。 我们想用flume实时追加采集日志,如果source里定义: a1.sources.r1.type = exec a1.sources.r1.command = tail -f /tmp/20191118.log001 则可以实时收取,但只能取到一个日志文件,该文件写满20MB后,生成20191118.log2,flume就采集不到了。 如果使用: a1.sources.r1.type = spooldir a1.sources.r1.spoolDir = /tmp a1.sources.r1.includePattern = S2019*.* 则可以采集所有文件,但只能将文件采集一次,无法做到实时增量采集。 请教对于这样的日志,如何使用flume采集增量日志,谢谢。

Flume运行报错,试了网上很多办法都没有效果,求解。(已解决)

![图片说明](https://img-ask.csdn.net/upload/202001/10/1578656542_760185.png) 已解决,是因为环境变量Path中有双引号""导致的报错,删除双引号""即可。

flume采集数据到hdfs性能问题

本人目前遇到flume采集写入hdfs性能等各种问题,大致如下。在10上的xx/xx目录下的数据进行读取 sink到08上的flume 由08上的flume写到07的hdfs上 30多m的文件写了好久。有时候会内存溢出等问题![图片说明](https://img-ask.csdn.net/upload/201503/12/1426162664_624860.jpg) # Name the components on this agent a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/configure the source a1.sources.r1.type = avro a1.sources.r1.bind = r09n08 a1.sources.r1.port = 55555 a1.sources.r1.interceptors = i1 a1.sources.r1.interceptors.i1.type = timestamp #hdfs sink a1.sinks.k1.type = hdfs a1.sinks.k1.hdfs.path = hdfs://r09n07:8020/project/dame/input/%Y%m%d/%H a1.sinks.k1.hdfs.fileType = DataStream a1.sinks.k1.hdfs.filePrefix = hdfs- a1.sinks.k1.hdfs.rollInterval = 0 #a1.sinks.k1.hdfs.fileSuffix = .log #a1.sinks.k1.hdfs.round = true #a1.sinks.k1.hdfs.roundValue = 1 #a1.sinks.k1.hdfs.roundUnit = minute a1.sinks.k1.hdfs.rollSize = 67108864 a1.sinks.k1.hdfs.rollCount = 0 #a1.sinks.k1.hdfs.writeFormat = Text # Use a channel which buffers events in file a1.channels = c1 a1.channels.c1.type = memory #a1.channels.c1.checkpointDir=/home/nids/wg/apache-flume-1.5.2-bin/checkpoint #a1.channels.c1.dataDirs=/home/nids/wg/apache-flume-1.5.2-bin/datadir a1.sinks.k1.hdfs.batchSize = 10000 #a1.sinks.k1.hdfs.callTimeout = 6000 #a1.sinks.k1.hdfs.appendTimeout = 6000 #a1.channels.c1.type = memory a1.channels.c1.capacity = 100000 a1.channels.c1.transactionCapacity = 10000 a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1 上面是08机器上的配置文件 ``` 下面是10机器上的配置文件 # Name the components on this agent a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe the sink a1.sinks.k1.type = logger #### a1.sources.r1.type = spooldir a1.sources.r1.spoolDir = /home/nids/wg/apache-flume-1.5.2-bin/ceshi12 a1.sources.r1.fileHeader =false a1.sources.r1.channels = c1 #### # Describe/configure the source #a1.sources.r1.type = avro a1.sources.r1.bind = localhost a1.sources.r1.port = 44444 # avro sink a1.sinks.k1.type = avro a1.sinks.k1.channel = c1 a1.sinks.k1.hostname = r09n08 a1.sinks.k1.port = 55555 # Use a channel which buffers events in file a1.channels = c1 a1.channels.c1.type = memory #a1.channels.c1.checkpointDir = /home/nids/wg/apache-flume-1.5.2-bin/checkpoint #a1.channels.c1.dataDirs = /home/nids/wg/apache-flume-1.5.2-bin/datadir a1.sinks.k1.hdfs.batchSize = 10000 #a1.channels.c1.type = memory a1.channels.c1.capacity = 100000 a1.channels.c1.transactionCapacity = 10000 # Bind the source and sink to the channel a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1 求各位高手解答。有时候只写了一部分数据就不再继续了,对单个文件执行时没有问题就是对目录扫描 channel是 memory类型时性能极差。不知道问题出在哪里 ```

kafka ArrayIndexOutOfBoundsException: 18

在kafka上出了下面这个问题,上网查了下都说是新版的kafka clien向旧版的kafka发送请求,旧版的kafka(<0.10)不支持ApiVersion(key:18) Request,造成的,但是我所有的produce,consumer,kafka服务器上装的kafka clien都是0.9.0.1,应该不会出现这个问题才对,为什么?求各位大神指点 ``` [2018-10-25 10:03:17,919] INFO [Kafka Server 0], started (kafka.server.KafkaServer) [2018-10-25 10:03:18,080] INFO [ReplicaFetcherManager on broker 0] Removed fetcher for partitions [topic-test,0] (kafka.server.ReplicaFetcherManager) [2018-10-25 10:03:18,099] INFO [ReplicaFetcherManager on broker 0] Removed fetcher for partitions [topic-test,0] (kafka.server.ReplicaFetcherManager) [2018-10-25 10:03:48,864] ERROR Processor got uncaught exception. (kafka.network.Processor) java.lang.ArrayIndexOutOfBoundsException: 18 at org.apache.kafka.common.protocol.ApiKeys.forId(ApiKeys.java:68) at org.apache.kafka.common.requests.AbstractRequest.getRequest(AbstractRequest.java:39) at kafka.network.RequestChannel$Request.<init>(RequestChannel.scala:79) at kafka.network.Processor$$anonfun$run$11.apply(SocketServer.scala:426) at kafka.network.Processor$$anonfun$run$11.apply(SocketServer.scala:421) at scala.collection.Iterator$class.foreach(Iterator.scala:742) at scala.collection.AbstractIterator.foreach(Iterator.scala:1194) at scala.collection.IterableLike$class.foreach(IterableLike.scala:72) at scala.collection.AbstractIterable.foreach(Iterable.scala:54) at kafka.network.Processor.run(SocketServer.scala:421) at java.lang.Thread.run(Thread.java:748) ```

flume配置了kakfaChannel后,启动报错!求大神帮忙

是在虚拟机上配置搭建的服务,flume部署在172.235.10.10上, kafka部署在172.235.10.11上。配置文件如下: flume: a1.sources = r1 a1.sinks = k1 a1.channels = c1 a1.sources.r1.type = http a1.sources.r1.channels = c1 a1.sources.r1.bind = 0.0.0.0 a1.sources.r1.port = 8000 a1.sinks.k1.type = file_roll a1.sinks.k1.channel=c1 a1.sinks.k1.directory=/usr/log/flume a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 a1.channels.c1.brokerList=172.235.10.11:9002 a1.channels.c1.topic=test2 a1.channels.c1.zookeeperConnect=172.235.10.11:2181 zookeeper是使用了kafka自带的,配置文件没做修改 kafka配置server.properties修改: host.name=172.235.10.11 zookeeper.connect=172.235.10.11:2181 kafka启动正常,生产者和消费者也是可以通话的 但在启动flume的时候却报这样的一个错误: Error while getting events from Kafka. This is usually caused by trying to read a non-flume event. Ensure the setting for parseAsFlumeEvent is correct 求大神帮忙

大数据:flume-ng启动报错

flume-ng1.5.0启动报错java.lang.OutOfMemoryError: Direct buffer memory。 flume-env.sh内存配置4G绝对足够了,请求解决方法

log4j向flume发送数据乱码

flume部署在linux上,log4J2目部署在windows上。用log4j的flumeAppender把数据发送到kafka上,但通过logsink和filesink打印出来的都是乱码,log4J在Windows本地打印出来是正常的。请问这个怎么解决。。。。 我在linux上用flume的avro client把一个UTF-8编码的文本发送到flume,依然是乱码。但用GBK的文本就是正常的。 但是我在win上吧log4j的编码都改过,flume打印出来的还是乱码。

运行flume的agent,出现如下错误

我的代码: ``` agent.sources = s1 agent.channels = c1 agent.sinks = k1 agent.sources.s1.type=spooldir agent.sources.s1.spoolDir=/tmp/logs/tomcat2kafka agent.sources.s1.channels=c1 agent.channels.c1.type=memory agent.channels.c1.capacity=10000 agent.channels.c1.transactionCapacity=100 #设置Kafka接收 agent.sinks.k1.type= org.apache.flume.sink.kafka.KafkaSink #设置Kafka的broker地址和端口号 agent.sinks.k1.brokerList=222.30.194.254:9092 #设置Kafka的Topic agent.sinks.k1.topic=kafkatest2 #设置序列化方式 agent.sinks.k1.serializer.class=kafka.serializer.StringEncoder agent.sinks.k1.channel=c1 ``` 错误提示: ``` [ERROR - org.apache.flume.sink.kafka.KafkaSink.process(KafkaSink.java:240)] Failed to publish events org.apache.kafka.common.errors.InterruptException: Flush interrupted. at org.apache.kafka.clients.producer.KafkaProducer.flush(KafkaProducer.java:546) at org.apache.flume.sink.kafka.KafkaSink.process(KafkaSink.java:224) at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:67) at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:145) at java.lang.Thread.run(Thread.java:748) Caused by: java.lang.InterruptedException at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:998) at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304) at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231) at org.apache.kafka.clients.producer.internals.ProduceRequestResult.await(ProduceRequestResult.java:57) at org.apache.kafka.clients.producer.internals.RecordAccumulator.awaitFlushCompletion(RecordAccumulator.java:425) at org.apache.kafka.clients.producer.KafkaProducer.flush(KafkaProducer.java:544) ... 4 more ``` 网上是真没有相应的答案,无奈了,给分求助

spark+flume运行jar包报错

spark+flume运行jar包报错 ![图片说明](https://img-ask.csdn.net/upload/201709/13/1505302384_331459.png) 这个jar包在类中就没有用到啊

kafka 消费端 处理数据比较慢,会不会出现数据积压?

如题,kafka消费端接收到数据后 要进行部分业务逻辑操作,可能会有3秒左右,处理很慢 的话,对程序有什么影响呢?新手提问, 望各位大神不吝赐教!

flume开启报错java.lang.SecurityException: sealing violation: package org.apache.flume.conf is sealed

开启了hadoop,zk,kafka之后,配置了conf,flume消费kafka中产生的信息 开启flume的命令: flume-ng agent --conf /home/hduser/apps/flume/conf --conf-file /home/hduser/apps/flume/conf/applog --name a1 -Dflume.root.logger=INFO,console 第一次开启 报错信息: ``` 2019-04-27 13:20:32,643 (main) [ERROR - org.apache.flume.node.Application.main(Application.java:374)] A fatal error occurred while running. Exception follows. java.lang.SecurityException: sealing violation: package org.apache.flume.conf is sealed at java.net.URLClassLoader.getAndVerifyPackage(URLClassLoader.java:399) at java.net.URLClassLoader.definePackageInternal(URLClassLoader.java:419) at java.net.URLClassLoader.defineClass(URLClassLoader.java:451) 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:335) at java.lang.ClassLoader.loadClass(ClassLoader.java:357) 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:335) at java.lang.ClassLoader.loadClass(ClassLoader.java:357) at org.apache.flume.node.Application.main(Application.java:350) ``` 上网查了老半天,实在没办法了,有大佬知道怎么办吗? 这问题就是说package org.apache.flume.conf is sealed被封装吗? 多次开启后的错误显示: ``` (main) [ERROR - org.apache.flume.node.Application.main(Application.java:374)] A fatal error occurred while running. Exception follows. java.lang.SecurityException: sealing violation: can't seal package org.apache.flume.conf: already loaded at java.net.URLClassLoader.getAndVerifyPackage(URLClassLoader.java:406) at java.net.URLClassLoader.definePackageInternal(URLClassLoader.java:419) at java.net.URLClassLoader.defineClass(URLClassLoader.java:451) 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:335) at java.lang.ClassLoader.loadClass(ClassLoader.java:357) 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:335) at java.lang.ClassLoader.loadClass(ClassLoader.java:357) at org.apache.flume.node.Application.main(Application.java:350) ``` 说那个已经被加载了 更蒙 了 求解决!

Flume运行报错,显示没有配置过滤器和正则表达式无效

Flume的配置文件: a1.sources=s1 a1.channels=c1 a1.sinks=k1 a1.sources.s1.type = spooldir a1.sources.s1.channels = c1 a1.sources.s1.spoolDir =/home/frankyu/serverlogs a1.source.s1.ignorePattern= ^(.)*\\.tmp$ a1.sinks.k1.type = hdfs a1.sinks.k1.hdfs.path = hdfs://hadoop1:9000/flume a1.sinks.k1.hdfs.writeFormat = Text a1.sinks.k1.hdfs.rollInterval = 0 a1.sinks.k1.hdfs.rollSize = 0 a1.sinks.k1.hdfs.rollCount = 10 a1.sinks.k1.channel = c1 a1.channels.c1.type = memory 报错信息: 2019-03-14 02:27:33,799 (conf-file-poller-0) [WARN - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.validateConfigFilterSet(FlumeConfiguration.java:623)] Agent configuration for 'a1' has no configfilters. 2019-03-14 02:27:33,790 (conf-file-poller-0) [WARN - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1161)] Invalid property specified: source.s1.ignorePattern 2019-03-14 02:27:33,796 (conf-file-poller-0) [WARN - org.apache.flume.conf.FlumeConfiguration.<init>(FlumeConfiguration.java:126)] Configuration property ignored: a1.source.s1.ignorePattern = ^(.)*\.tmp$ 感谢帮助!

flume-ng能否自定义数据读取完成标识?

flume读取文件时会为文件添加一个读取完成的标示,例如:文件python_20161027.log, 读取完成后会添加一个.COMPLETED的标示,把文件变成了python_20161027.log.COMPLETED 这样破坏了原有的文档结构,例如一些本来可以直接读取的txt文件,被flume采集后就不能再直接读取了,而且还会出现一些其他的问题。 我在使用flume的过程中发现,如果在上游一个程序在不停的写log日志,下游用flume实时采集日志有可能会报java.lang.IllegalStateException: File name has been re-used with different files. 这是因为我们上游的程序是以重定向的方式来写log日志的,当flume读取日志后,把日志名变成了python_20161027.log.COMPLETED后,上游程序再次生成log日志时,先判断是否存在python_20161027.log文件,如果没有的话就会创建python_20161027.log文件,而flume再次读取python_20161027.log文件时,还要生成python_20161027.log.COMPLETED文件。但是因为文件目录下已经存在此文件了,所以就会报上面的错误 我想问问各位大神,有没有什么办法可以让flume采集日志文件后,不改变原有的文件名,从而避免上面的问题

技术大佬:我去,你写的 switch 语句也太老土了吧

昨天早上通过远程的方式 review 了两名新来同事的代码,大部分代码都写得很漂亮,严谨的同时注释也很到位,这令我非常满意。但当我看到他们当中有一个人写的 switch 语句时,还是忍不住破口大骂:“我擦,小王,你丫写的 switch 语句也太老土了吧!” 来看看小王写的代码吧,看完不要骂我装逼啊。 private static String createPlayer(PlayerTypes p...

Vue + Spring Boot 项目实战(十九):Web 项目优化解决方案

快来一起探索如何打脸我们的破项目,兄弟姐妹们把害怕打在公屏上!

你连存活到JDK8中著名的Bug都不知道,我怎么敢给你加薪

CopyOnWriteArrayList.java和ArrayList.java,这2个类的构造函数,注释中有一句话 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 public ArrayList(Collection&lt;? ...

副业收入是我做程序媛的3倍,工作外的B面人生是怎样的?

提到“程序员”,多数人脑海里首先想到的大约是:为人木讷、薪水超高、工作枯燥…… 然而,当离开工作岗位,撕去层层标签,脱下“程序员”这身外套,有的人生动又有趣,马上展现出了完全不同的A/B面人生! 不论是简单的爱好,还是正经的副业,他们都干得同样出色。偶尔,还能和程序员的特质结合,产生奇妙的“化学反应”。 @Charlotte:平日素颜示人,周末美妆博主 大家都以为程序媛也个个不修边幅,但我们也许...

CSDN:因博主近期注重写专栏文章(已超过150篇),订阅博主专栏人数在突增,近期很有可能提高专栏价格(已订阅的不受影响),提前声明,敬请理解!

CSDN:因博主近期注重写专栏文章(已超过150篇),订阅博主专栏人数在突增,近期很有可能提高专栏价格(已订阅的不受影响),提前声明,敬请理解! 目录 博客声明 大数据了解博主粉丝 博主的粉丝群体画像 粉丝群体性别比例、年龄分布 粉丝群体学历分布、职业分布、行业分布 国内、国外粉丝群体地域分布 博主的近期访问每日增量、粉丝每日增量 博客声明 因近期博主写专栏的文章越来越多,也越来越精细,逐步优化文章。因此,最近一段时间,订阅博主专栏的人数增长也非常快,并且专栏价

一个HashMap跟面试官扯了半个小时

一个HashMap能跟面试官扯上半个小时 关注 安琪拉的博客 1.回复面试领取面试资料 2.回复书籍领取技术电子书 3.回复交流领取技术电子书 前言 HashMap应该算是Java后端工程师面试的必问题,因为其中的知识点太多,很适合用来考察面试者的Java基础。 开场 面试官: 你先自我介绍一下吧! 安琪拉: 我是安琪拉,草丛三婊之一,最强中单(钟馗不服)!哦,不对,串场了,我是**,目...

我说我不会算法,阿里把我挂了。

不说了,字节跳动也反手把我挂了。

记录下入职中软一个月(外包华为)

我在年前从上一家公司离职,没想到过年期间疫情爆发,我也被困在家里,在家呆着的日子让人很焦躁,于是我疯狂的投简历,看面试题,希望可以进大公司去看看。 我也有幸面试了我觉得还挺大的公司的(虽然不是bat之类的大厂,但是作为一名二本计算机专业刚毕业的大学生bat那些大厂我连投简历的勇气都没有),最后选择了中软,我知道这是一家外包公司,待遇各方面甚至不如我的上一家公司,但是对我而言这可是外包华为,能...

面试:第十六章:Java中级开发

HashMap底层实现原理,红黑树,B+树,B树的结构原理 Spring的AOP和IOC是什么?它们常见的使用场景有哪些?Spring事务,事务的属性,传播行为,数据库隔离级别 Spring和SpringMVC,MyBatis以及SpringBoot的注解分别有哪些?SpringMVC的工作原理,SpringBoot框架的优点,MyBatis框架的优点 SpringCould组件有哪些,他们...

培训班出来的人后来都怎么样了?(二)

接着上回说,培训班学习生涯结束了。后面每天就是无休止的背面试题,不是没有头脑的背,培训公司还是有方法的,现在回想当时背的面试题好像都用上了,也被问到了。回头找找面试题,当时都是打印下来天天看,天天背。 不理解呢也要背,面试造飞机,上班拧螺丝。班里的同学开始四处投简历面试了,很快就有面试成功的,刚开始一个,然后越来越多。不知道是什么原因,尝到胜利果实的童鞋,不满足于自己通过的公司,嫌薪水要少了,选择...

面试了一个 31 岁程序员,让我有所触动,30岁以上的程序员该何去何从?

最近面试了一个31岁8年经验的程序猿,让我有点感慨,大龄程序猿该何去何从。

大三实习生,字节跳动面经分享,已拿Offer

说实话,自己的算法,我一个不会,太难了吧

程序员垃圾简历长什么样?

已经连续五年参加大厂校招、社招的技术面试工作,简历看的不下于万份 这篇文章会用实例告诉你,什么是差的程序员简历! 疫情快要结束了,各个公司也都开始春招了,作为即将红遍大江南北的新晋UP主,那当然要为小伙伴们做点事(手动狗头)。 就在公众号里公开征简历,义务帮大家看,并一一点评。《启舰:春招在即,义务帮大家看看简历吧》 一石激起千层浪,三天收到两百多封简历。 花光了两个星期的所有空闲时...

工作八年,月薪60K,裸辞两个月,投简历投到怀疑人生!

近日,有网友在某职场社交平台吐槽,自己裸辞两个月了,但是找工作却让自己的心态都要崩溃了,全部无果,不是已查看无回音,就是已查看不符合。 “工作八年,两年一跳,裸辞两个月了,之前月薪60K,最近找工作找的心态崩了!所有招聘工具都用了,全部无果,不是已查看无回音,就是已查看不符合。进头条,滴滴之类的大厂很难吗???!!!投简历投的开始怀疑人生了!希望 可以收到大厂offer” 先来看看网...

大牛都会用的IDEA调试技巧!!!

导读 前天面试了一个985高校的实习生,问了他平时用什么开发工具,他想也没想的说IDEA,于是我抛砖引玉的问了一下IDEA的调试用过吧,你说说怎么设置断点...

都前后端分离了,咱就别做页面跳转了!统统 JSON 交互

文章目录1. 无状态登录1.1 什么是有状态1.2 什么是无状态1.3 如何实现无状态1.4 各自优缺点2. 登录交互2.1 前后端分离的数据交互2.2 登录成功2.3 登录失败3. 未认证处理方案4. 注销登录 这是本系列的第四篇,有小伙伴找不到之前文章,松哥给大家列一个索引出来: 挖一个大坑,Spring Security 开搞! 松哥手把手带你入门 Spring Security,别再问密...

97年世界黑客编程大赛冠军作品(大小仅为16KB),惊艳世界的编程巨作

这是世界编程大赛第一名作品(97年Mekka ’97 4K Intro比赛)汇编语言所写。 整个文件只有4095个字节, 大小仅仅为16KB! 不仅实现了3D动画的效果!还有一段震撼人心的背景音乐!!! 内容无法以言语形容,实在太强大! 下面是代码,具体操作看最后! @echo off more +1 %~s0|debug e100 33 f6 bf 0 20 b5 10 f3 a5...

程序员是做全栈工程师好?还是专注一个领域好?

昨天,有位大一的同学私信我,说他要做全栈工程师。 我一听,这不害了孩子么,必须制止啊。 谁知,讲到最后,更确定了他做全栈程序员的梦想。 但凡做全栈工程师的,要么很惨,要么很牛! 但凡很牛的,绝不是一开始就是做全栈的! 全栈工程师听起来好听,但绝没有你想象的那么简单。 今天听我来给你唠,记得帮我点赞哦。 一、全栈工程师的职责 如果你学习编程的目的只是玩玩,那随意,想怎么学怎么学。...

不要再到处使用 === 了

我们知道现在的开发人员都使用 === 来代替 ==,为什么呢?我在网上看到的大多数教程都认为,要预测 JavaScript 强制转换是如何工作这太复杂了,因此建议总是使用===。这些都...

什么是a站、b站、c站、d站、e站、f站、g站、h站、i站、j站、k站、l站、m站、n站?00后的世界我不懂!

A站 AcFun弹幕视频网,简称“A站”,成立于2007年6月,取意于Anime Comic Fun,是中国大陆第一家弹幕视频网站。A站以视频为载体,逐步发展出基于原生内容二次创作的完整生态,拥有高质量互动弹幕,是中国弹幕文化的发源地;拥有大量超粘性的用户群体,产生输出了金坷垃、鬼畜全明星、我的滑板鞋、小苹果等大量网络流行文化,也是中国二次元文化的发源地。 B站 全称“哔哩哔哩(bilibili...

十个摸鱼,哦,不对,是炫酷(可以玩一整天)的网站!!!

文章目录前言正文**1、Kaspersky Cyberthreat real-time map****2、Finding Home****3、Silk – Interactive Generative Art****4、Liquid Particles 3D****5、WINDOWS93****6、Staggering Beauty****7、Ostagram图片生成器网址****8、全历史网址*...

终于,月薪过5万了!

来看几个问题想不想月薪超过5万?想不想进入公司架构组?想不想成为项目组的负责人?想不想成为spring的高手,超越99%的对手?那么本文内容是你必须要掌握的。本文主要详解bean的生命...

用了这个 IDE 插件,5分钟解决前后端联调!

点击上方蓝色“程序猿DD”,选择“设为星标”回复“资源”获取独家整理的学习资料!作者 |李海庆我是一个 Web 开发前端工程师,受到疫情影响,今天是我在家办公的第78天。开发了两周,...

大厂的 404 页面都长啥样?最后一个笑了...

每天浏览各大网站,难免会碰到404页面啊。你注意过404页面么?猿妹搜罗来了下面这些知名网站的404页面,以供大家欣赏,看看哪个网站更有创意: 正在上传…重新上传取消 腾讯 正在上传…重新上传取消 网易 淘宝 百度 新浪微博 正在上传…重新上传取消 新浪 京东 优酷 腾讯视频 搜...

自从喜欢上了B站这12个UP主,我越来越觉得自己是个废柴了!

不怕告诉你,我自从喜欢上了这12个UP主,哔哩哔哩成为了我手机上最耗电的软件,几乎每天都会看,可是吧,看的越多,我就越觉得自己是个废柴,唉,老天不公啊,不信你看看…… 间接性踌躇满志,持续性混吃等死,都是因为你们……但是,自己的学习力在慢慢变强,这是不容忽视的,推荐给你们! 都说B站是个宝,可是有人不会挖啊,没事,今天咱挖好的送你一箩筐,首先啊,我在B站上最喜欢看这个家伙的视频了,为啥 ,咱撇...

代码注释如此沙雕,会玩还是你们程序员!

某站后端代码被“开源”,同时刷遍全网的,还有代码里的那些神注释。 我们这才知道,原来程序员个个都是段子手;这么多年来,我们也走过了他们的无数套路… 首先,产品经理,是永远永远吐槽不完的!网友的评论也非常扎心,说看这些代码就像在阅读程序员的日记,每一页都写满了对产品经理的恨。 然后,也要发出直击灵魂的质问:你是尊贵的付费大会员吗? 这不禁让人想起之前某音乐app的穷逼Vip,果然,穷逼在哪里都是...

Java14 新特性解读

Java14 已于 2020 年 3 月 17 号发布,官方特性解读在这里:https://openjdk.java.net/projects/jdk/14/以下是个人对于特性的中文式...

前端还能这么玩?(女朋友生日,用前端写了一个好玩的送给了她,高兴坏了)

前端还能这么玩?(女朋友生日,用前端写了一个好玩的送给了她,高兴坏了)

爬虫(101)爬点重口味的

小弟最近在学校无聊的很哪,浏览网页突然看到一张图片,都快流鼻血。。。然后小弟冥思苦想,得干一点有趣的事情python 爬虫库安装https://s.taobao.com/api?_ks...

工作两年简历写成这样,谁要你呀!

作者:小傅哥 博客:https://bugstack.cn 沉淀、分享、成长,让自己和他人都能有所收获! 一、前言 最近有伙伴问小傅哥,我的简历怎么投递了都没有反应,心里慌的很呀。 工作两年了目前的公司没有什么大项目,整天的维护别人的代码,有坑也不让重构,都烦死了。荒废我一身技能无处施展,投递的简历也没人看。我是不动物园里的猩猩,狒狒了! 我要加班,我要996,我要疯狂编码,求给我个机会… ...

相关热词 c# 解析cad c#数字格式化万 c#int转byte c#格式化日期 c# wpf 表格 c# 实现ad域验证登录 c#心跳包机制 c#使用fmod.dll c#dll vb 调用 c# outlook
立即提问
相关内容推荐