如何使用 Flume 将一组 csv 文件从本地目录复制到 HDFS答案

【问题标题】：How do I copy a set of csv files from my local directory to HDFS using Flume如何使用 Flume 将一组 csv 文件从本地目录复制到 HDFS
【发布时间】：2014-07-20 04:42:38
【问题描述】：

如何使用 Flume 将一组 csv 文件从本地目录复制到 HDFS？我尝试使用假脱机目录作为我的源，但未能复制。然后我使用以下水槽配置来得到我的结果：

agent1.sources = tail 
agent1.channels = MemoryChannel-2 
agent1.sinks = HDFS 
agent1.sources.tail.type = exec 
agent1.sources.tail.command = tail -F /home/cloudera/runs/*  
agent1.sources.tail.channels = MemoryChannel-2 
agent1.sinks.HDFS.channel = MemoryChannel-2 
agent1.sinks.HDFS.type = hdfs 
agent1.sinks.HDFS.hdfs.path = hdfs://localhost:8020/user/cloudera/runs                         
agent1.sinks.HDFS.hdfs.file.Type = DataStream 
agent1.channels.MemoryChannel-2.type = memory

我将文件复制到 hdfs，但它们包含特殊字符，对我没有用处。我的本地目录是 /home/cloudera/runs，我的 HDFS 目标目录是 /user/cloudera/runs。

【问题讨论】：

文件名以 Runs_Opposition_YYYY 格式命名（例如 Runs_Opposition_2000.csv）。他们有什么方法可以定期将这些文件名中附有年份的文件复制到 HDFS？

标签： csv hadoop flume flume-ng

【解决方案1】：

I used the below flume configuration to get the job done.

#Flume Configuration Starts
# Define a file channel called fileChannel on agent_slave_1
agent_slave_1.channels.fileChannel1_1.type = file 
# on linux FS
agent_slave_1.channels.fileChannel1_1.capacity = 200000
agent_slave_1.channels.fileChannel1_1.transactionCapacity = 1000
# Define a source for agent_slave_1
agent_slave_1.sources.source1_1.type = spooldir

# on linux FS
#Spooldir in my case is /home/cloudera/runs
agent_slave_1.sources.source1_1.spoolDir = /home/cloudera/runs/
agent_slave_1.sources.source1_1.fileHeader = false
agent_slave_1.sources.source1_1.fileSuffix = .COMPLETED
agent_slave_1.sinks.hdfs-sink1_1.type = hdfs

#Sink is /user/cloudera/runs_scored under hdfs
agent_slave_1.sinks.hdfs-sink1_1.hdfs.path = hdfs://localhost.localdomain:8020/user/cloudera/runs_scored/
agent_slave_1.sinks.hdfs-sink1_1.hdfs.batchSize = 1000
agent_slave_1.sinks.hdfs-sink1_1.hdfs.rollSize = 268435456
agent_slave_1.sinks.hdfs-sink1_1.hdfs.rollInterval = 0
agent_slave_1.sinks.hdfs-sink1_1.hdfs.rollCount = 50000000
agent_slave_1.sinks.hdfs-sink1_1.hdfs.writeFormat=Text

agent_slave_1.sinks.hdfs-sink1_1.hdfs.fileType = DataStream
agent_slave_1.sources.source1_1.channels = fileChannel1_1
agent_slave_1.sinks.hdfs-sink1_1.channel = fileChannel1_1

agent_slave_1.sinks =  hdfs-sink1_1
agent_slave_1.sources = source1_1
agent_slave_1.channels = fileChannel1_1

【讨论】：

【解决方案2】：

在你的水槽里，你需要使用

agent1.sinks.HDFS.hdfs.fileType = DataStream

而不是

agent1.sinks.HDFS.hdfs.file.Type = DataStream

休息似乎很好。

【讨论】：