Flumeng和Mysql进行整合 Flumeng批量处理

 

 

package com.iteblog.flume;

	 
	import com.google.common.base.Preconditions;
	import com.google.common.base.Throwables;
	import com.google.common.collect.Lists;
	import org.apache.flume.*;
	import org.apache.flume.conf.Configurable;
	import org.apache.flume.sink.AbstractSink;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;
	 
	import java.sql.Connection;
	import java.sql.DriverManager;
	import java.sql.PreparedStatement;
	import java.sql.SQLException;
	import java.util.List;
	 
	public class MysqlSink extends AbstractSink implements Configurable {
	 
	    private Logger LOG = LoggerFactory.getLogger(MysqlSink.class);
	    private String hostname;
	    private String port;
	    private String databaseName;
	    private String tableName;
	    private String user;
	    private String password;
	    private PreparedStatement preparedStatement;
	    private Connection conn;
	    private int batchSize;
	 
	    public MysqlSink() {
	        LOG.info("MysqlSink start...");
	    }
	 
	    @Override
	    public void configure(Context context) {
	        hostname = context.getString("hostname");
	        Preconditions.checkNotNull(hostname, "hostname must be set!!");
	        port = context.getString("port");
	        Preconditions.checkNotNull(port, "port must be set!!");
	        databaseName = context.getString("databaseName");
	        Preconditions.checkNotNull(databaseName, "databaseName must be set!!");
	        tableName = context.getString("tableName");
	        Preconditions.checkNotNull(tableName, "tableName must be set!!");
	        user = context.getString("user");
	        Preconditions.checkNotNull(user, "user must be set!!");
	        password = context.getString("password");
	        Preconditions.checkNotNull(password, "password must be set!!");
	        batchSize = context.getInteger("batchSize", 100);
	        Preconditions.checkNotNull(batchSize > 0, "batchSize must be a positive number!!");
	    }
	 
	    @Override
	    public void start() {
	        super.start();
	        try {
	            //调用Class.forName()方法加载驱动程序
	            Class.forName("com.mysql.jdbc.Driver");
	        } catch (ClassNotFoundException e) {
	            e.printStackTrace();
	        }
	 
	        String url = "jdbc:mysql://" + hostname + ":" + port + "/" + databaseName;
	        //调用DriverManager对象的getConnection()方法，获得一个Connection对象
	 
	        try {
	            conn = DriverManager.getConnection(url, user, password);
	            conn.setAutoCommit(false);
	            //创建一个Statement对象
	            preparedStatement = conn.prepareStatement("insert into " + tableName +
	                                               " (content) values (?)");
	 
	        } catch (SQLException e) {
	            e.printStackTrace();
	            System.exit(1);
	        }
	 
	    }
	 
	    @Override
	    public void stop() {
	        super.stop();
	        if (preparedStatement != null) {
	            try {
	                preparedStatement.close();
	            } catch (SQLException e) {
	                e.printStackTrace();
	            }
	        }
	 
	        if (conn != null) {
	            try {
	                conn.close();
	            } catch (SQLException e) {
	                e.printStackTrace();
	            }
	        }
	    }
	 
	    @Override
	    public Status process() throws EventDeliveryException {
	        Status result = Status.READY;
	        Channel channel = getChannel();
	        Transaction transaction = channel.getTransaction();
	        Event event;
	        String content;
	 
	        List<String> actions = Lists.newArrayList();
	        transaction.begin();
	        try {
	            for (int i = 0; i < batchSize; i++) {
	                event = channel.take();
	                if (event != null) {
	                    content = new String(event.getBody());
	                    actions.add(content);
	                } else {
	                    result = Status.BACKOFF;
	                    break;
	                }
	            }
	 
	            if (actions.size() > 0) {
	                preparedStatement.clearBatch();
	                for (String temp : actions) {
	                    preparedStatement.setString(1, temp);
						preparedStatement.addBatch();
	                }
	                preparedStatement.executeBatch();
	 
	                conn.commit();
	            }
	            transaction.commit();
	        } catch (Throwable e) {
	            try {
	                transaction.rollback();
	            } catch (Exception e2) {
	                LOG.error("Exception in rollback. Rollback might not have been" +
	                        "successful.", e2);
	            }
	            LOG.error("Failed to commit transaction." +
	                    "Transaction rolled back.", e);
	            Throwables.propagate(e);
	        } finally {
	            transaction.close();
	        }
 
	        return result;
	    }
	}

 

 

pom文件中的依赖：

<dependencies>
	        <dependency>
	            <groupId>org.apache.flume</groupId>
	            <artifactId>flume-ng-core</artifactId>
	        </dependency>
	 
	        <dependency>
	            <groupId>org.apache.flume</groupId>
	            <artifactId>flume-ng-configuration</artifactId>
	        </dependency>
	 
	        <dependency>
	            <groupId>mysql</groupId>
	            <artifactId>mysql-connector-java</artifactId>
	            <version>5.1.25</version>
	        </dependency>
	 
	        <dependency>
	            <groupId>org.slf4j</groupId>
	            <artifactId>slf4j-api</artifactId>
	        </dependency>
	 
	        <dependency>
	            <groupId>org.slf4j</groupId>
	            <artifactId>slf4j-log4j12</artifactId>
	            <scope>test</scope>
	        </dependency>
	</dependencies>

 

 

运行程序时，先在Mysql中创建一个表

mysql> create table mysqltest(
	    -> id int(11) NOT NULL AUTO_INCREMENT,
	    -> content varchar(50000) NOT NULL,
	    -> PRIMARY KEY (`id`)
	    -> ) ENGINE=InnoDB AUTO_INCREMENT=4 DEFAULT CHARSET=utf8; 
	Query OK, 0 rows affected, 1 warning (0.05 sec)

 

 

然后在flume中创建以下配置

 

agent.sinks.mysqlSink.type = com.iteblog.flume.MysqlSink
agent.sinks.mysqlSink.hostname=localhost
agent.sinks.mysqlSink.port=3306
agent.sinks.mysqlSink.databaseName=ngmonitor
agent.sinks.mysqlSink.tableName=mysqltest
agent.sinks.mysqlSink.user=root
agent.sinks.mysqlSink.password=123456
agent.sinks.mysqlSink.channel = c1

 

 

用下面的命令就可以启动：

bin/flume-ng agent -c conf/ -f conf/mysql_test.conf  -n agent

 

 

再看下Mysql中的情况：

mysql> select count(*) from mysqltest;
	+----------+
	| count(*) |
	+----------+
	|    98300 |
	+----------+