flume连接metaQ传数据到spark进行实时计算

以前我们在使用hadoop的时候，都是使用shell脚本来从线上服务器拉取日志（今天凌晨把昨天生成的日志拉过来）。这种情况适用于日质量较小的情况。但是日质量很大了，比如一天1T的日志，那么在使用shell脚本就不能够满足要求了。有两个原因：

1. 拉取的速度较慢。因为从线上服务器拉取日志是需要跨机房的，而且机房的带宽有限，如果在拉取日志时不加限制，那么带宽就全被拉取日志占了，留给线上服务的带宽就会很小，所以这样就需要舍弃shell脚本的方式。

2. shell脚本拉取日志经常会出现失败的情况，不是很稳定。

现在我们使用的工具是flume，他的部署方式是多个client向一个server传日志。



部署的client和server都是用同一个安装包，只是配置文件不同。安装地址见：

http://download.csdn.net/detail/aaa1117a8w5s6d/7973839

我们在实时计算的时候需要使用一个MQ，我们选择的是淘宝的MetaQ，但是flume默认是不会向metaQ导数据的，所以就需要我们修改源码，

在flume-ng-core工程（flume的核心代码）下的org.apache.flume.sink包里添加一个类：

/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements.  See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership.  The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License.  You may obtain a copy of the License at
*
*     http://www.apache.org/licenses/LICENSE-2.0 *
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.apache.flume.sink;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.flume.Channel;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.taobao.metamorphosis.Message;
import com.taobao.metamorphosis.client.MessageSessionFactory;
import com.taobao.metamorphosis.client.MetaClientConfig;
import com.taobao.metamorphosis.client.MetaMessageSessionFactory;
import com.taobao.metamorphosis.client.producer.MessageProducer;
import com.taobao.metamorphosis.client.producer.SendResult;
import com.taobao.metamorphosis.exception.MetaClientException;
import com.taobao.metamorphosis.utils.ZkUtils.ZKConfig;

public class MetaQSink extends AbstractSink implements Configurable {

private static final Logger logger = LoggerFactory
.getLogger(MetaQSink.class);

private MessageSessionFactory sessionFactory;
private MessageProducer producer;

private String zkConnect;
private String zkRoot;
private String topic;
private int batchSize;
private int threadNum;
private ExecutorService executor;

@Override
public void configure(Context context) {

this.zkConnect = context.getString("sink.zkConnect");
this.zkRoot = context.getString("sink.zkRoot");
this.topic = context.getString("sink.topic");
this.batchSize = context.getInteger("sink.batchSize", 10000);
this.threadNum = context.getInteger("sink.threadNum", 50);
executor = Executors.newCachedThreadPool();

MetaClientConfig metaClientConfig = new MetaClientConfig();
ZKConfig zkConfig = new ZKConfig();
zkConfig.zkConnect = zkConnect;
zkConfig.zkRoot = zkRoot;
metaClientConfig.setZkConfig(zkConfig);
try {
sessionFactory = new MetaMessageSessionFactory(metaClientConfig);
} catch (MetaClientException e) {
e.printStackTrace();
logger.error("", e);
throw new RuntimeException("init error");
}
producer = sessionFactory.createProducer();
logger.info("zkConnect:" + zkConnect + ", zkRoot:" + zkRoot
+ ", topic:" + topic);
}

@Override
public Status process() throws EventDeliveryException {
long start = System.currentTimeMillis();
producer.publish(topic);
Status result = Status.READY;
final Channel channel = getChannel();
final AtomicInteger al = new AtomicInteger(0);
final CountDownLatch cdl = new CountDownLatch(threadNum);
for (int t = 0; t < threadNum; t++) {
executor.execute(new Runnable() {

@Override
public void run() {

Event event = null;
Transaction transaction = null;
int i = 0;
try {
transaction = channel.getTransaction();
transaction.begin();
boolean startTransaction = false;
for (i = 0; i < batchSize; i++) {
event = channel.take();
if (event != null) {
if (i == 0) {
producer.beginTransaction();
startTransaction = true;
}
final SendResult sendResult = producer
.sendMessage(new Message(topic, event
.getBody()));
// check result
if (!sendResult.isSuccess()) {
logger.error("Send message failed,error message:"
+ sendResult.getErrorMessage());
throw new RuntimeException(
"Send message failed,error message:"
+ sendResult
.getErrorMessage());
} else {
logger.debug("Send message successfully,sent to "
+ sendResult.getPartition());
}
} else {
// No event found, request back-off semantics
// from the sink
// runner
// result = Status.BACKOFF;
break;
}

}
if (startTransaction) {
producer.commit();
}
al.addAndGet(i);
transaction.commit();
} catch (Exception ex) {
logger.error("error while rollback:", ex);
try {
producer.rollback();
} catch (Exception e) {
e.printStackTrace();
}
transaction.rollback();
} finally {
cdl.countDown();
transaction.close();
}
}
});
}
try {
cdl.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
if (al.get() == 0) {
result = Status.BACKOFF;
}

logger.info("metaqSink_new,process:{},time:{},queue_size:{}",
new Object[] { al.get(), System.currentTimeMillis() - start,
channel.getSize() });
return result;

}
}

然后在flume里配置如下：

# example.conf: A single-node Flume configuration

# Name the components on this agent

info.sources = info_source

info.sinks = info_sink info_sink_to_metaq

info.channels = info_channel info_channel_to_metaq

# Describe/configure the source

info.sources.info_source.type = avro

info.sources.info_source.bind = 10.0.3.19

info.sources.info_source.port = 58001

info.sources.info_source.threads = 24

info.sinks.info_sink.type = file_roll

info.sinks.info_sink.sink.directory = /data1/logs/flume/info

info.sinks.info_sink.sink.name= info

ifno.sinks.info_sink.sink.batchSize= 20000

info.sinks.info_sink_to_metaq.type = org.apache.flume.sink.MetaQSink

info.sinks.info_sink_to_metaq.sink.zkConnect = 10.0.5.108:2181,10.0.5.109:2181,10.0.5.110:2181

info.sinks.info_sink_to_metaq.sink.zkRoot= /meta

info.sinks.info_sink_to_metaq.sink.topic= info

ifno.sinks.info_sink_to_metaq.sink.batchSize= 20000

# Describe the channel

info.channels.info_channel.type = memory

info.channels.info_channel.capacity = 10000000

info.channels.info_channel.transactionCapacity = 10000000

info.channels.info_channel_to_metaq.type = memory

info.channels.info_channel_to_metaq.capacity = 10000000

info.channels.info_channel_to_metaq.transactionCapacity = 10000000

# Bind the source and sink to the channel

info.sources.info_source.channels = info_channel info_channel_to_metaq

info.sinks.info_sink.channel = info_channel

info.sinks.info_sink_to_metaq.channel = info_channel_to_metaq

nginx.sources = nginx_source

nginx.sinks = nginx_sink

nginx.channels = nginx_channel

# Describe/configure the source

nginx.sources.nginx_source.type = avro

nginx.sources.nginx_source.bind = 10.0.3.19

nginx.sources.nginx_source.port = 58002

#

#

nginx.sinks.nginx_sink.type = file_roll

nginx.sinks.nginx_sink.sink.directory = /data1/logs/flume/nginx

nginx.sinks.nginx_sink.sink.name= nginx

nginx.sinks.nginx_sink.sink.batchSize= 2000

#

# Describe the channel

nginx.channels.nginx_channel.type = SPILLABLEMEMORY

# #1000W 1G

nginx.channels.nginx_channel.memoryCapacity = 20000000

nginx.channels.nginx_channel.overflowCapacity = 200000000

nginx.channels.nginx_channel.checkpointDir = /data1/logs/flume/nginx/check

nginx.channels.nginx_channel.dataDirs = /data1/logs/flume/nginx/data

# Bind the source and sink to the channel

nginx.sources.nginx_source.channels = nginx_channel

nginx.sinks.nginx_sink.channel = nginx_channel

usage.sources = usage_source

usage.sinks = usage_sink

usage.channels = usage_channel

# Describe/configure the source

usage.sources.usage_source.type = avro

usage.sources.usage_source.bind = 10.0.3.19

usage.sources.usage_source.port = 58003

#

#

usage.sinks.usage_sink.type = file_roll

usage.sinks.usage_sink.sink.directory = /data1/logs/flume/usage

usage.sinks.usage_sink.sink.name= usage

usage.sinks.usage_sink.sink.batchSize= 1000

# #

# # Describe the channel

usage.channels.usage_channel.type = SPILLABLEMEMORY

# # #1000W 1G

usage.channels.usage_channel.memoryCapacity = 20000000

usage.channels.usage_channel.overflowCapacity = 200000000

usage.channels.usage_channel.checkpointDir = /data1/logs/flume/usage/check

usage.channels.usage_channel.dataDirs = /data1/logs/flume/usage/data

# # Bind the source and sink to the channel

usage.sources.usage_source.channels = usage_channel

usage.sinks.usage_sink.channel = usage_channel

注：上面的配置文件是配置了3个flume的信息。

---------------------------------------------------------------------------------------------------------------

metaQ的安装包：

http://download.csdn.net/detail/aaa1117a8w5s6d/7974039

metaQ向spark传数据，见工程：

http://download.csdn.net/detail/aaa1117a8w5s6d/7974053