flume+hdfs构建大数据

作者:Chris

hadoop #

http://hadoop.apache.org/docs/r1.0.4/cn/

hdfs dfs -mkdir /ebuyposp
hdfs dfs -mkdir /ebuyposp2
hdfs dfs -ls /

hdfs dfs -cat /ebuyposp1/k1_.1462357028642
hdfs dfs -rm -r -f -skipTrash /ebuyposp/*
hdfs dfs -rm -r -f -skipTrash /ebuyposp2/*
hdfs dfs -rm -r -f -skipTrash /ebuyposp2
hdfs dfs -rm -r -f -skipTrash /test20160428

file_roll格式 #

# clientAgent
clientAgent.channels = c1
clientAgent.sources = s1
clientAgent.sinks = k1 k2


# clientAgent Spooling Directory Source
clientAgent.sources.s1.type = avro
clientAgent.sources.s1.channels =c1
clientAgent.sources.s1.bind = 127.0.0.1
clientAgent.sources.s1.port = 44444

# clientAgent FileChannel
clientAgent.channels.c1.type = memory
clientAgent.channels.c1.capacity = 200000000
clientAgent.channels.c1.keep-alive = 30
clientAgent.channels.c1.write-timeout = 30
clientAgent.channels.c1.checkpoint-timeout=600
clientAgent.channels.c1.transactionCapacity = 1000


# clientAgent Sinks
# k1 sink
clientAgent.sinks.k1.channel = c1
clientAgent.sinks.k1.type = file_roll

clientAgent.sinks.k1.sink.directory = ./tmp1
clientAgent.sinks.k1.sink.rollInterval=86400
clientAgent.sinks.k1.sink.batchSize=100
clientAgent.sinks.k1.sink.serializer=text
clientAgent.sinks.k1.sink.serializer.appendNewline = false

# connect to serverAgent
clientAgent.sinks.k1.hostname = 127.0.0.1
clientAgent.sinks.k1.port = 41415

# k2 sink
clientAgent.sinks.k2.channel = c1
clientAgent.sinks.k2.type = file_roll

clientAgent.sinks.k2.sink.directory = ./tmp2
clientAgent.sinks.k2.sink.rollInterval=86400
clientAgent.sinks.k2.sink.batchSize=100
clientAgent.sinks.k2.sink.serializer=text
clientAgent.sinks.k2.sink.serializer.appendNewline = false

# connect to CollectorBackupAgent
clientAgent.sinks.k2.hostname = 127.0.0.1
clientAgent.sinks.k2.port = 41415

# clientAgent sinks group
clientAgent.sinkgroups = g1
clientAgent.sinkgroups.g1.sinks = k1 k2
# load_balance type
clientAgent.sinkgroups.g1.processor.type = load_balance
clientAgent.sinkgroups.g1.processor.backoff = true
clientAgent.sinkgroups.g1.processor.selector = random
clientAgent.sinkgroups.g1.processor.priority.k1 = 10
clientAgent.sinkgroups.g1.processor.priority.k2 = 9

spooldir监听目录格式 #

# clientAgent
clientAgent.channels = c1
clientAgent.sources = s1
clientAgent.sinks = k1 k2

# clientAgent Spooling Directory Source
clientAgent.sources.s1.type = spooldir

# 监听目录
clientAgent.sources.s1.spoolDir =/app/test/ebuy3.0/ebuyposp/logs/

clientAgent.sources.s1.fileHeader = true
#clientAgent.sources.s1.deletePolicy =immediate
clientAgent.sources.s1.deletePolicy =never
clientAgent.sources.s1.batchSize =1000
clientAgent.sources.s1.channels =c1
clientAgent.sources.s1.deserializer.maxLineLength =1048576

# clientAgent FileChannel
clientAgent.channels.c1.type = file

# 检查传输及断点续传等验证文件,可原理可参考MR合并小文件的方式
clientAgent.channels.c1.checkpointDir = ./spool/checkpoint

# 传输数据
clientAgent.channels.c1.dataDirs = ./spool/data

clientAgent.channels.c1.capacity = 200000000
clientAgent.channels.c1.keep-alive = 30
clientAgent.channels.c1.write-timeout = 30
clientAgent.channels.c1.checkpoint-timeout=600

# clientAgent Sinks
# k1 sink
clientAgent.sinks.k1.channel = c1
clientAgent.sinks.k1.type = avro

# connect to serverAgent
clientAgent.sinks.k1.hostname = 127.0.0.1
clientAgent.sinks.k1.port = 41415

# k2 sink
clientAgent.sinks.k2.channel = c1
clientAgent.sinks.k2.type = avro

# connect to CollectorBackupAgent
clientAgent.sinks.k2.hostname = 127.0.0.1
clientAgent.sinks.k2.port = 41415

# clientAgent sinks group
clientAgent.sinkgroups = g1
clientAgent.sinkgroups.g1.sinks = k1 k2

# load_balance type
clientAgent.sinkgroups.g1.processor.type = load_balance
clientAgent.sinkgroups.g1.processor.backoff = true
clientAgent.sinkgroups.g1.processor.selector = random
clientAgent.sinkgroups.g1.processor.priority.k1 = 10
clientAgent.sinkgroups.g1.processor.priority.k2 = 9

client传输到server #

# clientAgent
clientAgent.channels = c1
clientAgent.sources = s1
clientAgent.sinks = k1 k2


# clientAgent Spooling Directory Source
clientAgent.sources.s1.type = avro
clientAgent.sources.s1.channels =c1
clientAgent.sources.s1.bind = 127.0.0.1
clientAgent.sources.s1.port = 44444


# clientAgent FileChannel
clientAgent.channels.c1.type = memory
clientAgent.channels.c1.capacity = 200000000
clientAgent.channels.c1.keep-alive = 30
clientAgent.channels.c1.write-timeout = 30
clientAgent.channels.c1.checkpoint-timeout=600
clientAgent.channels.c1.transactionCapacity = 1000


# clientAgent Sinks
# k1 sink
clientAgent.sinks.k1.channel = c1
clientAgent.sinks.k1.type = avro

# connect to serverAgent
clientAgent.sinks.k1.hostname = 127.0.0.1
clientAgent.sinks.k1.port = 41415

# k2 sink
clientAgent.sinks.k2.channel = c1
clientAgent.sinks.k2.type = avro

# connect to CollectorBackupAgent
clientAgent.sinks.k2.hostname = 127.0.0.1
clientAgent.sinks.k2.port = 41415

# clientAgent sinks group
clientAgent.sinkgroups = g1
clientAgent.sinkgroups.g1.sinks = k1 k2

# load_balance type
clientAgent.sinkgroups.g1.processor.type = load_balance
clientAgent.sinkgroups.g1.processor.backoff = true
clientAgent.sinkgroups.g1.processor.selector = random
clientAgent.sinkgroups.g1.processor.priority.k1 = 10
clientAgent.sinkgroups.g1.processor.priority.k2 = 9

server配置 #

# serverAgent
serverAgent.channels = c2
serverAgent.sources = s2
serverAgent.sinks =k1 k2

# serverAgent AvroSource
serverAgent.sources.s2.type = avro
serverAgent.sources.s2.bind = 0.0.0.0
serverAgent.sources.s2.port = 41415
serverAgent.sources.s2.channels = c2


# serverAgent FileChannel
serverAgent.channels.c2.type = file
serverAgent.channels.c2.checkpointDir =./spool/checkpoint
serverAgent.channels.c2.dataDirs = ./spool/data
serverAgent.channels.c2.capacity = 200000000
serverAgent.channels.c2.transactionCapacity=6000
serverAgent.channels.c2.checkpointInterval=60000

# serverAgent hdfsSink
serverAgent.sinks.k1.type = hdfs
serverAgent.sinks.k1.channel = c2
serverAgent.sinks.k1.hdfs.path = hdfs://localhost:9000/ebuyposp1
serverAgent.sinks.k1.hdfs.filePrefix =k1_%{file}
serverAgent.sinks.k1.hdfs.inUsePrefix =_
serverAgent.sinks.k1.hdfs.inUseSuffix =.tmp
serverAgent.sinks.k1.hdfs.rollSize = 0
serverAgent.sinks.k1.hdfs.rollCount = 0
serverAgent.sinks.k1.hdfs.rollInterval = 240
serverAgent.sinks.k1.hdfs.writeFormat = Text
serverAgent.sinks.k1.hdfs.fileType = DataStream
serverAgent.sinks.k1.hdfs.batchSize = 6000
serverAgent.sinks.k1.hdfs.callTimeout = 60000


serverAgent.sinks.k2.type = hdfs
serverAgent.sinks.k2.channel = c2
serverAgent.sinks.k2.hdfs.path = hdfs://localhost:9000/ebuyposp2
serverAgent.sinks.k2.hdfs.filePrefix =k2_%{file}
serverAgent.sinks.k2.hdfs.inUsePrefix =_
serverAgent.sinks.k2.hdfs.inUseSuffix =.tmp
serverAgent.sinks.k2.hdfs.rollSize = 0
serverAgent.sinks.k2.hdfs.rollCount = 0
serverAgent.sinks.k2.hdfs.rollInterval = 240
serverAgent.sinks.k2.hdfs.writeFormat = Text
serverAgent.sinks.k2.hdfs.fileType = DataStream
serverAgent.sinks.k2.hdfs.batchSize = 6000
serverAgent.sinks.k2.hdfs.callTimeout = 60000

HUE #

$ git clone https://github.com/cloudera/hue.git
$ cd hue
$ make apps
$ build/env/bin/hue runserver
$ build/env/bin/supervisor

您的评论

Build by Loppo 0.6.14