Flume按文件的方式导入hdfs的问题

sstutu · 发表于 2014-10-31 21:57:09

wjhdtx 发表于 2014-10-31 10:50
大概就是这样，楼主说的最后一个上传到hdfs的.tmp文件大小为0我这没有出现，但我配置的是一个文件一个事件 ...

flume二次开发可以实现，文件名自定义。对于最后一个文件，它是用来做标记的，如果不想这么做，还是自己定义吧。
这些个人功能，都需要自己编程实现。

可以参考这个
flume 插件开发介绍

只能帮到这里了，希望开发完之后，能给大家共享下。相互学习和帮助

wjhdtx · 发表于 2014-11-3 08:33:08

sstutu 发表于 2014-10-31 21:57
flume二次开发可以实现，文件名自定义。对于最后一个文件，它是用来做标记的，如果不想这么做，还是自己定 ...

谢谢回复，完成后分享我的成果。

wjhdtx · 发表于 2014-11-3 11:09:03

继续分享：
-----------------------------------------
1. 如果上传到hdfs中的文件想使用上传前的文件名，使用

a1.sources.r1.basenameHeader = true
a1.sources.r1.basenameHeaderKey = basename

a1.sinks.k1.hdfs.filePrefix = %{basename}
复制代码

2. .tmp文件一直在，可以通过缩短hdfs.idleTimeout来实现，比如设置60，就是60秒没有事件写入该文件，则关闭该文件，在hdfs中重命名去掉.tmp。

wjhdtx · 发表于 2014-11-4 09:39:01

本帖最后由 pig2 于 2014-11-4 10:06 编辑

sstutu 发表于 2014-10-31 21:57
flume二次开发可以实现，文件名自定义。对于最后一个文件，它是用来做标记的，如果不想这么做，还是自己定 ...

再请教下，需求是从本地上传hdfs过程中，文件是一个整体，一个文件对应一个文件。

目前小文件从其他目录拷贝到spool目录会在hdfs中产生该文件，但是当文件较大时，比如有2G的文件，文件在hdfs中的大小为90多M时，又新开了一个文件进行写入，为什么不是追加到原有文件？？

pig2 · 发表于 2014-11-4 10:49:25

wjhdtx 发表于 2014-11-4 09:39
再请教下，需求是从本地上传hdfs过程中，文件是一个整体，一个文件对应一个文件。

试试把这个参数设置为0
hdfs.rollSize

wjhdtx · 发表于 2014-11-4 11:33:54

pig2 发表于 2014-11-4 10:49
试试把这个参数设置为0
hdfs.rollSize

已经设置为0了

ss

日志文件已经通过生成，然后拷贝到spool目录，fume监控该目录，然后上传到hdfs。我是一个文件对应一个事件，就是文件一对一的上传到hdfs中。
我测试小文件没问题，可是一个大文件比如2G，在hdfs中生成一个95.4M的文件，然后又产生一个文件，循环写到95.4M，直到把2G写完，最后只有两个文件，前一个文件固定是95.4M，后面的一个是临时的工作文件。

看了源码，调了好多参数，也没用，进行不下去了：（

muyannian · 发表于 2014-11-4 11:46:42

本帖最后由 muyannian 于 2014-11-4 11:48 编辑

wjhdtx 发表于 2014-11-4 11:33
已经设置为0了

找到个资料，但是对flume不熟悉，你看看是否对你有帮助

使用说明输出到hdfs的数据，首先在hdfs上创建文件.tmp,然后文件关闭时，将tmp后缀去掉，存储方案与file输出类似，可以设定时间间隔、文件大小、接受事件条数作为滚动生成新文件的依据，默认30s

可配置项

Name	Default	Description
channel	–
type	–	The component type name, needs to be hdfs
hdfs.path	–	HDFS directory path (eg hdfs://namenode/flume/webdata/)
hdfs.filePrefix	FlumeData	Name prefixed to files created by Flume in hdfs directory
hdfs.rollInterval	30	Number of seconds to wait before rolling current file (0 = never roll based on time interval)
hdfs.rollSize	1024	File size to trigger roll, in bytes (0: never roll based on file size)
hdfs.rollCount	10	Number of events written to file before it rolled (0 = never roll based on number of events)
hdfs.batchSize	1	number of events written to file before it flushed to HDFS
hdfs.txnEventMax	100
hdfs.codeC	–	Compression codec. one of following : gzip, bzip2, lzo, snappy
hdfs.fileType	SequenceFile	File format: currently SequenceFile,DataStream orCompressedStream(1)DataStream will not compress output file and please don’t set codeC (2)CompressedStream requires set hdfs.codeC with an available codeC
hdfs.maxOpenFiles	5000
hdfs.writeFormat	–	“Text” or “Writable”
hdfs.appendTimeout	1000
hdfs.callTimeout	10000
hdfs.threadsPoolSize	10	Number of threads per HDFS sink for HDFS IO ops (open, write, etc.)
hdfs.rollTimerPoolSize	1	Number of threads per HDFS sink for scheduling timed file rolling
hdfs.kerberosPrincipal	–	Kerberos user principal for accessing secure HDFS
hdfs.kerberosKeytab	–	Kerberos keytab for accessing secure HDFS
hdfs.round	false	Should the timestamp be rounded down (if true, affects all time based escape sequences except %t)
hdfs.roundValue	1	Rounded down to the highest multiple of this (in the unit configured usinghdfs.roundUnit), less than current time.
hdfs.roundUnit	second	The unit of the round down value - second,minute orhour.
serializer	TEXT	Other possible options include AVRO_EVENT or the fully-qualified class name of an implementation of theEventSerializer.Builder interface.
serializer.*

howtodown · 发表于 2014-11-4 12:06:40

wjhdtx 发表于 2014-11-4 11:33
已经设置为0了

这里面肯定是有参数配置的，flume不会把它写死，分界点就是文件的大小。被分割成两个文件，有三种可能
1.这个文件是被hdfs分割的，也就是说是hdfs方面做了限制
验证方式：
可以通过上传一个2G的文件，看是否会产生两个文件

2.flume自身分割的
这个肯定有参数配置，需要特别注意文件大小方面的参数

3.可能是自己疏忽了，是否程序没有传递完毕，或则，最后还会合并。

我在找找资料，你可以先试试这些方法。