flume传输大文件异常Input length = 2 - 简书
flume传输大文件异常Input length = 2
在使用flume spooling组件传输文件时，经常会出现以下问题：1.文件只能传输一部分。2.第二次将文件放到flume的监控目录下时，不能传输文件，监控目录中的文件也不会被删除。3.flume重启之后，文件才能继续被传输。原因:在Spooling传输文件的时候，出现了同时读写一个文件导致的问题。
异常:Uncaught exception in SpoolDirectorySource thread. Restart or reconfigure Flume to continue processing.java.nio.charset.MalformedInputException: Input length = 2。爆出下面的异常：
解决办法：1.将原始日志cp到安装flume的同级目录下（防止在第二步mv的时候，不是在同一个磁盘会出现上面的异常）。2.使用mv到flume监控目录。
http://geek.csdn.net/news/detail/210469http://www.36dsj.com/archives/87945 本文除了一些常用的大数据工具，还总结汇总了其他大数据工具，几乎是最全的大数据工具的总结。 如果你想入门大数据，可以对他们进行...
主要收集整理的内容主要有日志收集系统、消息系统、分布式服务、集群管理、RPC、基础设施、搜索引擎、Iaas和监控管理等大数据开源工具。 日志收集系统 一、FacebookScribe 贡献者：Facebook 简介：Scribe是Facebook开源的日志收集系统，在Fac...
Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具（例如配置管理，服务发现，断路器，智能路由，微代理，控制总线）。分布式系统的协调导致了样板模式, 使用Spring Cloud开发人员可以快速地支持实现这些模式的服务和应用程序。他们将在任何分布式...
//一共81个，开源大数据处理工具汇总（下），包括日志收集系统/集群管理/RPC等 | 36大数据http://www.36dsj.com/archives/25042 接上一部分：一共81个，开源大数据处理工具汇总（上），第二部分主要收集整理的内容主要有日志收集系统、消息...
用两张图告诉你，为什么你的 App 会卡顿? - Android - 掘金Cover 有什么料？ 从这篇文章中你能获得这些料： 知道setContentView()之后发生了什么？ ... Android 获取 View 宽高的常用正确方式，避免为零 - 掘金相信有很多朋友...
文/志成哥哥 总想写点东西，却总觉得没意思，当自己越来越大的时候，那份感觉也就慢慢变了，是自己内心存在着一种心理——抵触。长大了，改变了。 其实就是自己舍不得，舍不得一切的一切。舍不得过去成长的记忆，舍不得现在的那份快乐，舍不得承认自己的模样。 在小学，也许童真真的可以让人...
夕阳西下，余晖洒落。教室里，留存的三两位学生，在各自座位上做着他们觉得重要的事情。相比于大多数时候，七十多人共同拥有的紧凑空间里，此刻，生出不明自言的空灵。仿佛流水旁，看落叶入水、鱼虾摆纹。身上、心上仿佛都有了未知的力量。一身两用、一心两用，做着习题、背着单词，却仿若浮到半...
题目：大家都知道斐波那契数列，现在要求输入一个整数n，请你输出斐波那契数列的第n项。n&=39 分析： 迭代方法，用两个变量记录fn - 1和fn - 2: 不使用递归的原因是如果是一个超大的n，会造成内存溢出
《罗辑思维》第五季，，《制造业晴雨表》笔记 首先罗胖将这两年比较热的两个词“实体经济”和“虚拟经济”转化成了更接地气的两个词“制造业”和“服务业”。 制造业和服务业，对社会发展的贡献，哪个更大?格力的董明珠和哇哈哈的宗庆都在公众场合，大声疾呼制造业的重要性。...
做梦的，醒来的 沉默着，躁动着 世界太大，人会迷失 要么庸俗，要么孤独 家里连续下了半个月的雨了，这样的雨夜貌似很适合一个人听着音乐捧着一本惬意的书读一读，但是我却捧起了那本曾让我郁闷的《一个人的朝圣》，更是一遍一遍退了敲打的文字，尘封在心底的情感不知该怎样表达，也许就这样...Flume相关参数介绍
Flume配置文件(flume-site.conf)
1、 watchdog
watchdog.restarts.max
watchdog每分钟重启的最大数???
2、 common node
flume.config.heartbeat.period
node发送心跳周期,默认5000(毫秒)
flume.node.status.port
node web端口
flume.node.heartbeat.backoff.ceiling
node向master发送心跳之后等待反馈的最大时长,默认60000(毫秒)
flume.node.http.autofindport
如果已有node启动，允许第二个node自动选择一个未使用的端口做web服务。多个node的界面端口从3向后延续
flume.agent.logdir
agent日志路径
flume.agent.logdir.maxage
当前处于打开状态agent日志文件收集信息的时长,在这之后该日志文件将会被关闭，并将数据发送到网络，默认10000(毫秒)
flume.agent.logdir.retransmit
在end-to-end模式下agent向collector发送数据失败后再次发送的间隔时长，默认60000(毫秒)，建议至少是flume.collector.roll.millis的两倍
flume.agent.failover.backoff.initial
当primary sink(可以认为是第一collector)故障后，重启primary sink的一个延迟时间，在此期间，agent将把数据发送到secondary sink(可能是第二collector)
flume.agent.failover.backoff.max
在一定时限内尝试链接故障节点失败后，agent将把数据转发向备用节点
4、collector
flume.collector.event.host
默认collector地址
flume.collector.port
默认collector端口
flume.collector.dfs.dir
最终数据发向目录(默认)，可以是本地，可以是hdfs，默认是/tmp
flume.collector.dfs.compress.codec
压缩格式GzipCodec, DefaultCodec (deflate), BZip2Codec，默认是None
flume.collector.roll.millis
hdfs文件切换(关闭后新建)的时长
flume.collector.output.format
collector发送数据格式avro, avrojson(默认), avrodata…
flume.master.servers
用逗号分隔多个master地址列表
flume.master.store
master配置存储方式(zookeeper/memory) zookeeper保证master的配置在多master节点之间同步，memory则保存在内存中，其配置随着master宕机而丢失
flume.master.serverid
master的唯一标识
flume.master.http.port
flume.master.heartbeat.missed.max
判断节点失效的最大未达心跳数
flume.master.savefile
当前flume配置文件的路径，默认conf/current.flume
flume.master.savefile.autoload
启动时是否加载current.flume，默认false
flume.master.gossip.period
master通信周期(毫秒)
flume.master.heartbeat.rpc
THRIFT/AVRO
flume.event.rpc
THRIFT/AVRO
flume.report.server.rpc.type
THRIFT/AVRO
6、zookeeper
flume.master.zk.logdir
zookeeper日志路径
flume.thrift.socket.timeout.ms
thrift网络连接超时时间(毫秒)
command shell(flume command)
[root@axing-centOS1 paas_log]# flume shell
08:14:46,047 [main] INFO conf.FlumeConfiguration: Loading configurations from /home/opt/flume-0.9.4/conf
==================================================
FlumeShell v0.9.4-cdh3u6
Copyright (c) Cloudera 2010, All Rights Reserved
==================================================
Type a command to execute (hint: many commands
only work when you are connected to a master node)
You may connect to a master node by typing:
connect host[:adminport=35873[:reportport=45678]]
[flume (disconnected)] connect localhost
Using default admin port: 35873
Using default report port: 45678
Connecting to Flume master localhost:...
08:14:50,420 [main] INFO util.AdminRPCThrift: Connected to master at localhost:35873
[flume localhost:] getnodestatus
Master knows about 7 nodes
agent130 --& LOST
collector_192_168_124_130 --& ACTIVE
collect130 --& LOST
collector_129 --& ACTIVE
agent_1_168_124_130 --& ACTIVE
agent_130 --& ACTIVE
agent_129 --& ACTIVE
connect master:port
登录master
config logicalnode source sink
为逻辑节点配置一个source到sink的映射
getnodestatus
获得节点状态(HELLO, CONFIGURING, ACTIVE, IDLE, ERROR, DECOMMISSIONED, LOST )
HELLO, node启动时
CONFIGURING, node被配置后
ACTIVE, 一个event从source送达到sink
IDLE, source中所有evnet发送完毕后
ERROR, 节点故障退出，数据没有flush
DECOMMISSIONED, node被master移除
LOST, master长时间未收到node心跳
getconfigs
getmappings [physical node]
如果physical node参数被省略，将显示所有logical node到physical node的映射关系
同步执行命令
Source file
异步执行命令
wait ms [cmdid]
设定一个时间，周期检查命令进程的状态(success or failure)
waitForNodesActive ms node1 [node2 […]]
设定一个时间，检查node是否处于使用(configuring, active)状态
waitForNodesDone ms node1 [node2 […]]
设定一个时间，检查node是否处于未用(IDLE, ERROR, LOST)状态
command shell(exec & submit command)
包含转义字符的java string
能引住除单引号之外的所有字符
touch master, 不做操作
config logicalnode source sink
为逻辑节点配置source到sink的映射
multiconfig flumespec
unconfig logicalnode
取消逻辑节点的配置，影响master调整故障转移列表(failover list)
refreshAll logicalnode
save filename
保存current configuration到master硬盘
load filename
从master中加载current configuration
map physicalnode logicalnode
配置物理节点到逻辑节点的映射关系，master的配置将被同步到logicalnode
spawn physicalnode logicalnode
decommission logicalnode
unmap physicalnode logicalnode
purge logicalnode
清除状态，类似重启一个logical node, 适用于(DECOMMISSIONED、 LOST)状态
清除所有logical node的状态
Flume Source
1、Flume’s Tiered Event Sources
collectorSource[(port)]
Collector source，监听端口汇聚数据
autoCollectorSource
通过master协调物理节点自动汇聚数据
logicalSource
逻辑source，由master分配端口并监听rpcSink
2、Flume’s Basic Sources
监听用户编辑历史和快捷键输入，只在node_nowatch模式下可用
监听标准输入，只在node_nowatch模式下可用，每行将作为一个event source
rpcSource(port)
由rpc框架(thrift/avro)监听tcp端口
text("filename")
一次性读取一个文本，每行为一个event
tail("filename"[, startFromEnd=false])
每行为一个event。监听文件尾部的追加行，如果startFromEnd为true，tail将从文件尾读取，如果为false，tail将从文件开始读取全部数据
multitail("filename"[, file2 [,file3… ] ])
同上，同时监听多个文件的末尾
tailDir("dirname"[, fileregex=".*"[, startFromEnd=false[, recurseDepth=0]]])
监听目录中的文件末尾，使用正则去选定需要监听的文件（不包含目录），recurseDepth为递归监听其下子目录的深度
seqfile("filename")
监听hdfs的sequencefile，全路径
syslogUdp(port)
监听Udp端口
syslogTcp(port)
监听Tcp端口
syslogTcp1(port)
只监听Tcp端口的一个链接
execPeriodic("cmdline", ms)
周期执行指令，监听指令的输出，整个输出都被作为一个event
execStream("cmdline")
执行指令，监听指令的输出，输出的每一行被作为一个event
exec("cmdline"[, aggregate=false[,restart=false[,period=0]]])
执行指令，监听指令的输出，aggregate如果为true，整个输出作为一个event如果为false，则每行作为一个event。如果restart为true，则按period为周期重新运行
synth(msgCount,msgSize)
随即产生字符串event,msgCount为产生数量，msgSize为串长度
synthrndsize(msgCount,minSize,maxSize)
同上，minSize – maxSize
nonlsynth(msgCount,msgSize)
asciisynth(msgCount,msgSize)
Ascii码字符
twitter("username","pw"[,"url"])
尼玛twitter的插件啊
irc("server",port, "nick","chan")
scribe[(+port)]
Scribe插件
report[(periodMillis)]
生成所有physical node报告为事件源
Flume Sinks
1、Flume’s Collector Tier Event Sinks
collectorSink( "fsdir","fsfileprefix",rollmillis)
collectorSink，数据通过collector汇聚之后发送到hdfs, fsdir 是hdfs目录，fsfileprefix为文件前缀码
2、Flume’s Agent Tier Event Sinks
agentSink[("machine"[,port])]
Defaults to agentE2ESink，如果省略，machine参数，默认使用flume.collector.event.host与flume.collector.event.port作为默认collecotr（以下同此）
agentE2ESink[("machine"[,port])]
执着的agent，如果agent发送event没有收到collector成功写入的状态码，该event将被agent重复发送，直到接到成功写入的状态码
agentDFOSink[("machine" [,port])]
本地热备agent，agent发现collector节点故障后，不断检查collector的存活状态以便重新发送event，在此间产生的数据将缓存到本地磁盘中
agentBESink[("machine"[,port])]
不负责的agent，如果collector故障，将不做任何处理，它发送的数据也将被直接丢弃
agentE2EChain("m1[:_p1_]" [,"m2[:_p2_]"[,…]])
指定多个collector提高可用性。 当向主collector发送event失效后，转向第二个collector发送，当所有的collector失败后，它会非常执着的再来一遍...
agentDFOChain("m1[:_p1_]"[, "m2[:_p2_]"[,…]])
同上，当向所有的collector发送事件失效后，他会将event缓存到本地磁盘，并检查collector状态，尝试重新发送
agentBEChain("m1[:_p1_]"[, "m2[:_p2_]"[,…]])
同上，当向所有的collector发送事件失效后，他会将event丢弃
autoE2EChain
无需指定collector, 由master协调管理event的流向
autoDFOChain
autoBEChain
3、Flume’s Logical Sinks
logicalSink("logicalnode")
4、Flume’s Basic Sinks
在不使用collector收集event的情况下，可将source直接发向basic sinks
console[("formatter")]
转发到控制台
text("txtfile" [,"formatter"])
转发到文本文件
seqfile("filename")
转发到seqfile
dfs("hdfspath")
转发到hdfs
customdfs("hdfspath"[, "format"])
自定义格式dfs
+escapedCustomDfs("hdfspath", "file", "format")
rpcSink("host"[, port])
syslogTcp("host"[,port])
发向网络地址
irc("host",port, "nick", "chan")
没有更多推荐了，
(window.slotbydup=window.slotbydup || []).push({
id: '4770930',
container: s,
size: '300,250',
display: 'inlay-fix'Apache Flume 大数据ETL工具Apache Flume 大数据ETL工具社交乐趣大新闻百家号由于今日头条的限制，每天只能上传一集视频，该套视频早都已经录制结束，一下是内容的大纲，有需要的朋友可以关注一下汪文君Flume深入浅出实战视频-01-ApacheFlume的介绍.wmv汪文君Flume深入浅出实战视频-02-环境搭建，快速入门，快速启动一个Agent.wmv汪文君Flume深入浅出实战视频-03-Apache Flume的配置，Avro Agent的详细介绍.wmv汪文君Flume深入浅出实战视频-04-Apache Flume Agent的四种架构介绍.wmv汪文君Flume深入浅出实战视频-05-Apache Flume如何保证数据不丢失，源码级别的分析.wmv汪文君Flume深入浅出实战视频-06-Apache Flume Avro Source详细介绍以及IpFilter功能介绍.wmv汪文君Flume深入浅出实战视频-07-Apache Flume ExecSource和SeqSource实战讲解.wmv汪文君Flume深入浅出实战视频-08-Apache Flume SpoolingSource实战讲解.wmv汪文君Flume深入浅出实战视频-09-Apache Flume TailDirSource，NetCatSource实战讲解.wmv汪文君Flume深入浅出实战视频-10-Apache Flume SyslogSource(TCP,UDP,multiport)实战讲解.wmv汪文君Flume深入浅出实战视频-11-Apache Flume HttpSource以及SSL加密source实战讲解.wmv汪文君Flume深入浅出实战视频-12-Apache Flume MemoryChannel,PseudoChannel详细介绍.wmv汪文君Flume深入浅出实战视频-13-Apache Flume JdbcChannel的详细介绍，以及解决包冲突的问题.wmv汪文君Flume深入浅出实战视频-14-Apache Flume FileChannel和SpillableMemoryChannel实战讲解.wmv汪文君Flume深入浅出实战视频-15-Apache Flume Selector之Replicating多路复制实战详解 .wmv汪文君Flume深入浅出实战视频-16-Apache Flume Selector之Multiplexing多路复用实战详解 .wmv汪文君Flume深入浅出实战视频-17-Apache Flume NullSink，LoggerSink，AvroSink回顾.wmv汪文君Flume深入浅出实战视频-18-Apache Flume FileRollSink实战讲解.wmv汪文君Flume深入浅出实战视频-19-Apache Flume HdfsSink实战讲解.wmv汪文君Flume深入浅出实战视频-20-Apache Flume Agent使用sinkgroup实现高可用failover功能实战讲解.wmv汪文君Flume深入浅出实战视频-21-Apache Flume Agent使用sinkgroup实现负载均衡loadbalance功能实战讲解.wmv汪文君Flume深入浅出实战视频-22-Apache Flume HiveSink实战讲解（期间踩了很多坑）.wmv汪文君Flume深入浅出实战视频-23-Apache Flume Interceptors介绍以及Timestamp Interceptor实战讲解.wmv汪文君Flume深入浅出实战视频-24-Apache Flume HostInterceptors，StaticInterceptors，UUIDInterceptors实战详解.wmv汪文君Flume深入浅出实战视频-25-Apache Flume SearchAndReplace Interceptors实战详解.wmv汪文君Flume深入浅出实战视频-26-Apache Flume Regex Filtering Interceptor实战详解（非常重要）.wmv汪文君Flume深入浅出实战视频-27-Apache Flume Regex Extractor Interceptor实战详解.wmv汪文君Flume深入浅出实战视频-28-Apache Flume Monitoring by JMX实战详解.wmv汪文君Flume深入浅出实战视频-29-Apache Flume Monitoring by HTTP实战详解.wmv汪文君Flume深入浅出实战视频-30-Apache Flume自定义Source实战练习，自定义Flume插件.wmv汪文君Flume深入浅出实战视频-31-Apache Flume自定义Sink实战练习，自定义Flume插件.wmv汪文君Flume深入浅出实战视频-32-Apache Flume自定义Interceptors实战练习，自定义Flume插件.wmv汪文君Flume深入浅出实战视频-33-InfluxDB介绍，安装，Console操作，Http操作，Admin Panel.wmv汪文君Flume深入浅出实战视频-34-Grafana的介绍以及创建一个Panel.wmv汪文君Flume深入浅出实战视频-35-Apache Flume综合实战练习之功能需求描述.wmv汪文君Flume深入浅出实战视频-36-Apache Flume综合实战练习之InfluxDbSink插件开发.wmv汪文君Flume深入浅出实战视频-37-Apache Flume综合实战练习之配置调试InfluxDbSink插件.wmv汪文君Flume深入浅出实战视频-38-Apache Flume综合实战练习之通过JMX获取程序CPU，内存的实时使用情况.wmv汪文君Flume深入浅出实战视频-39-Apache Flume综合实战练习之JmxPerformanceSource插件开发及调试.wmv汪文君Flume深入浅出实战视频-40-Apache Flume综合实战练习之配置Granfana显示Flume采集过来的性能数据.wmv汪文君Flume深入浅出实战视频-41-Apache Flume综合实战练习之统计分析业务受理数据.wmv汪文君Flume深入浅出实战视频-42-Apache Flume总结.wmv汪文君 Apache Flume本文由百家号作者上传并发布，百家号仅提供信息发布平台。文章仅代表作者个人观点，不代表百度立场。未经作者许可，不得转载。社交乐趣大新闻百家号最近更新：简介:社会是一本书，一起慢慢读作者最新文章相关文章You are a piece of shit
windows下flume配置与遇到的问题
最近学习大数据，听到数据采集需要flume这一个东西。于是开始捣鼓环境，可能是linux上没有hadoop环境没成功。后来尝试了在windows环境下使用flume，虽然其中遇到种种问题，但总算解决了。
1、解压flume
2、在conf目录下新增a.conf
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = D:/work/logs/a #源日志文件位置
a1.sources.r1.deserializer = org.apache.flume.sink.solr.morphline.BlobDeserializer$Builder
a1.sources.r1.batchsize = 1
# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://132.122.237.125:9000/input/flume #需要放的hadoop input目标文件位置
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.writeFormat = Text
a1.sinks.k1.hdfs.batchSize = 1
a1.sinks.k1.hdfs.rollInterval = 0
a1.sinks.k1.hdfs.rollcount = 1
a1.sinks.k1.hdfs.rollsize = 0
#a1.sinks.k1.hdfs.filePrefix = logFile.%Y-%m-%d
a1.sinks.k1.hdfs.fileSuffix = .log #需要采集的数据后缀名
a1.sinks.k1.hdfs.useLocalTimeStamp = true
a1.sinks.k1.hdfs.idleTimeout = 60000
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c13、打开CMD，进入到flume bin目录下，输入命令：flume-ng.cmd agent --conf ..\conf --conf-file ..\conf\a.conf --name a1其中a.conf 就是自己写的配置文件
这就启动成功了
在指定的源文件地址D:/work/logs/a放入a.log文件，就能自动同步到HDFS上的flume文件夹内，同步后a.log文件名会改变，标识为已同步状态
但发现同步到flume的文件会有乱码问题，以下是解决方案：
a1.sinks.k1.hdfs.fileType = DataStream 一开始配置文件写的是filetype，但配置文件识别大小写，改成 fileType之后就可以了。
没有更多推荐了，
(window.slotbydup=window.slotbydup || []).push({
id: '4770930',
container: s,
size: '300,250',
display: 'inlay-fix'123 条评论分享收藏感谢收起