标签: flume

我有一个分布在7台服务器上的移动服务,每台服务器都在执行特定的任务.我想记录他们的信息,然后从他们那里获取商业智能.我把它四舍五入到Flume.我如何使用它来收集信息？我的系统是用PHP编写的.flume是否适用于PHP？

我正在尝试使用Hive配置水槽以使用Hive Sink类型将水槽输出保存到蜂巢表.我有单节点集群.我使用mapr hadoop发行版.

这是我的flume.conf

agent1.sources = source1
agent1.channels = channel1
agent1.sinks = sink1

agent1.sources.source1.type = exec
agent1.sources.source1.command = cat /home/andrey/flume_test.data

agent1.sinks.sink1.type = hive
agent1.sinks.sink1.channel = channel1
agent1.sinks.sink1.hive.metastore = thrift://127.0.0.1:9083
agent1.sinks.sink1.hive.database = default
agent1.sinks.sink1.hive.table = flume_test
agent1.sinks.sink1.useLocalTimeStamp = false
agent1.sinks.sink1.round = true
agent1.sinks.sink1.roundValue = 10
agent1.sinks.sink1.roundUnit = minute
agent1.sinks.sink1.serializer = DELIMITED
agent1.sinks.sink1.serializer.delimiter = "," 
agent1.sinks.sink1.serializer.serdeSeparator = ','
agent1.sinks.sink1.serializer.fieldnames = id,message

agent1.channels.channel1.type = FILE
agent1.channels.channel1.transactionCapacity = 1000000
agent1.channels.channel1.checkpointInterval 30000
agent1.channels.channel1.maxFileSize = 2146435071
agent1.channels.channel1.capacity 10000000
agent1.sources.source1.channels = channel1

我的数据flume_test.data

1,AAAAAAAA
2,BBBBBBB
3,CCCCCCCC
4,DDDDDD …

这是我第一次来这里,很抱歉,如果我不发布罚款,抱歉我的英语不好.

我正在尝试配置Apache Flume和Elasticsearch接收器.一切都很好,似乎它运行正常,但是当我启动代理时有2个警告; 以下是:

2015-11-16 09:11:22,122 (lifecycleSupervisor-1-3) [ERROR - org.apache.flume.lifecycle.LifecycleSupervisor$MonitorRunnable.run(LifecycleSupervisor.java:253)] Unable to start SinkRunner: { policy:org.apache.flume.sink.DefaultSinkProcessor@ce359aa counterGroup:{ name:null counters:{} } } - Exception follows.
java.lang.NoSuchMethodError: org.elasticsearch.common.transport.InetSocketTransportAddress.<init>(Ljava/lang/String;I)V
    at org.apache.flume.sink.elasticsearch.client.ElasticSearchTransportClient.configureHostnames(ElasticSearchTransportClient.java:143)
    at org.apache.flume.sink.elasticsearch.client.ElasticSearchTransportClient.<init>(ElasticSearchTransportClient.java:77)
    at org.apache.flume.sink.elasticsearch.client.ElasticSearchClientFactory.getClient(ElasticSearchClientFactory.java:48)
    at org.apache.flume.sink.elasticsearch.ElasticSearchSink.start(ElasticSearchSink.java:357)
    at org.apache.flume.sink.DefaultSinkProcessor.start(DefaultSinkProcessor.java:46)
    at org.apache.flume.SinkRunner.start(SinkRunner.java:79)
    at org.apache.flume.lifecycle.LifecycleSupervisor$MonitorRunnable.run(LifecycleSupervisor.java:251)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
2015-11-16 09:11:22,137 (lifecycleSupervisor-1-3) [WARN - org.apache.flume.lifecycle.LifecycleSupervisor$MonitorRunnable.run(LifecycleSupervisor.java:260)] Component SinkRunner: { policy:org.apache.flume.sink.DefaultSinkProcessor@ce359aa counterGroup:{ name:null counters:{} } } stopped, since it could not besuccessfully started due to missing dependencies …

我正在用Flume收集日志到HDFS.对于测试用例,我有小文件(~300kB),因为日志收集过程是根据实际用途进行缩放的.

有没有简单的方法将这些小文件组合成更接近HDFS块大小(64MB)的较大文件？

使用Hbase API(Get/Put)或HBQL API,是否可以检索特定列的时间戳？

我正在测试Flume NG(1.2.0)来收集日志.这是一个简单的测试,Flume收集日志文件flume_test.log并将收集的日志作为sysout输出到控制台.conf/flume.conf是:

agent.sources = tail
agent.channels = memoryChannel
agent.sinks = loggerSink

agent.sources.tail.type = exec
agent.sources.tail.command = tail -f /Users/pj/work/flume_test.log
agent.sources.tail.channels = memoryChannel

agent.sinks.loggerSink.channel = memoryChannel
agent.sinks.loggerSink.type = logger

agent.channels.memoryChannel.type = memory
agent.channels.memoryChannel.capacity = 100

我按照以下方式运行Flume:

$ $FLUME_HOME/bin/flume-ng agent --conf $FLUME_HOME/conf --conf-file $FLUME_HOME/conf/flume.conf --name agent1 -Dflume.root.logger=DEBUG,console

在控制台上运行Flume日志后:

Info: Sourcing environment configuration script /usr/local/lib/flume-ng/conf/flume-env.sh
+ exec /Library/Java/JavaVirtualMachines/jdk1.7.0_07.jdk/Contents/Home/bin/java -Xmx20m -Dflume.root.logger=DEBUG,console -cp '/usr/local/lib/flume-ng/conf:/usr/local/lib/flume-ng/lib/*' -Djava.library.path= org.apache.flume.node.Application --conf-file /usr/local/lib/flume-ng/conf/flume.conf --name agent1
2012-09-12 18:23:52,049 (main) [INFO - org.apache.flume.lifecycle.LifecycleSupervisor.start(LifecycleSupervisor.java:67)] Starting lifecycle supervisor 1
2012-09-12 …

我正在尝试一些非常简单的方法,我希望将内容从我的日志文件推送到S3.我能够创建一个水槽代理,它将从apache访问日志文件中读取内容并使用记录器接收器.现在我试图找到一个解决方案,我可以用"S3 sink"替换记录器接收器.(我知道默认情况下不存在)

我正在寻找指引我走正确道路的一些指示.下面是我目前使用的测试属性文件.

a1.sources=src1
a1.sinks=sink1
a1.channels=ch1

#source configuration
a1.sources.src1.type=exec
a1.sources.src1.command=tail -f /var/log/apache2/access.log

#sink configuration
a1.sinks.sink1.type=logger

#channel configuration
a1.channels.ch1.type=memory
a1.channels.ch1.capacity=1000
a1.channels.ch1.transactionCapacity=100

#links
a1.sources.src1.channels=ch1
a1.sinks.sink1.channel=ch1

首先,我正在考虑使用什么来将事件放入Hadoop,在那里存储它们并定期对它们进行分析(可能使用Ooozie来安排定期分析)Kafka或Flume,并决定Kafka可能是一个更好的解决方案,因为我们还有一个用于执行事件处理的组件,因此以这种方式,批处理和事件处理组件以相同的方式获取数据.

但是知道我正在寻找具体的建议如何从经纪人那里获取数据到Hadoop.

我在这里发现Flume可以与Kafka结合使用

• Flume - 包含Kafka Source(消费者)和Sink(制作人)

并且在同一页面和Kafka文档中也发现了一些名为Camus的东西

• Camus - LinkedIn的Kafka => HDFS管道.这个用于LinkedIn的所有数据,效果很好.

我对能做到这一点的更好(更简单,记录更好的解决方案)感兴趣吗？此外,有任何示例或教程如何做到这一点？

我应该何时使用这种变体而不是更简单的高级消费者？

如果有另外一个/更好的解决方案而不是这两个,我会打开建议.

谢谢

我使用HTTPSourcein Flume来接收 格式的POST事件,json如下所示:

{"username":"xyz","password":"123"}

我的问题是:我是否必须修改事件的来源(我的意思是发送JSON到Flume 的那个)所以JSON具有以下格式:

[{
  "headers" : {
             "timestamp" : "434324343",
             "host" : "random_host.example.com"
             },
  "body" : "{"username":"xyz","password":"123"}"
}]

这是最好的方法吗？或者我可以在其他地方修改它？

我的conf文件flume agent是:

## Componentes
SomeAgent.sources = SomeHTTP
SomeAgent.channels = MemChannel
SomeAgent.sinks = SomeHDFS

## Fuente e Interceptores
SomeAgent.sources.SomeHTTP.type = http
SomeAgent.sources.SomeHTTP.port = 5140
SomeAgent.sources.SomeHTTP.handler = org.apache.flume.source.http.JSONHandler
SomeAgent.sources.SomeHTTP.channels = MemChannel
SomeAgent.sources.SomeHTTP.interceptors = i1 i2

## Interceptores
SomeAgent.sources.SomeHTTP.interceptors.i1.type = timestamp
SomeAgent.sources.SomeHTTP.interceptors.i2.type = host
SomeAgent.sources.SomeHTTP.interceptors.i2.hostHeader = hostname

## …

从Kafka“导入”流数据到HBase的最佳实践是什么？

用例如下：车辆传感器数据流式传输到 Kafka。之后，必须转换这些传感器数据（即，从人类可读数据中的 protobuf 反序列化）并存储在 HBase 中。

1）您推荐哪种工具集（例如，Kafka --> Flume --> HBase，Kafka --> Storm --> HBase，Kafka --> Spark Streaming --> HBase，Kafka --> HBase）

2) 进行 protobuf 去盐化的最佳位置是什么（例如，在 Flume 中使用拦截器）？

感谢您的支持。

最好的，托马斯