flume 采集mysql

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.SimpleEvent;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;

public class MySQLSource extends PollableSource implements Configurable {

    private String url;
    private String user;
    private String password;
    private String query;

    @Override
    public void configure(Context context) {
        url = context.getString("url");
        user = context.getString("user");
        password = context.getString("password");
        query = context.getString("query");
    }

    @Override
    public Status process() throws EventDeliveryException {
        Connection conn = null;
        Statement stmt = null;
        ResultSet rs = null;
        try {
            Class.forName("com.mysql.jdbc.Driver");
            conn = DriverManager.getConnection(url, user, password);
            stmt = conn.createStatement();
            rs = stmt.executeQuery(query);

            while (rs.next()) {
                Event event = new SimpleEvent();
                event.setBody(rs.getString(1).getBytes());
                getChannelProcessor().processEvent(event);
            }
        } catch (Exception e) {
            e.printStackTrace();
            return Status.BACKOFF;
        } finally {
            try {
                if (rs != null) rs.close();
                if (stmt != null) stmt.close();
                if (conn != null) conn.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return Status.READY;
    }
}

agent.sources = mysqlSource
agent.channels = memoryChannel
agent.sinks = hdfsSink

agent.sources.mysqlSource.type = com.example.MySQLSource
agent.sources.mysqlSource.url = jdbc:mysql://localhost:3306/mydb
agent.sources.mysqlSource.user = root
agent.sources.mysqlSource.password = password
agent.sources.mysqlSource.query = SELECT * FROM mytable

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

agent.sinks.hdfsSink.type = hdfs
agent.sinks.hdfsSink.hdfs.path = hdfs://localhost:9000/user/flume/data
agent.sinks.hdfsSink.hdfs.filePrefix = events-
agent.sinks.hdfsSink.hdfs.fileType = DataStream
agent.sinks.hdfsSink.hdfs.writeFormat = Text
agent.sinks.hdfsSink.hdfs.rollInterval = 0
agent.sinks.hdfsSink.hdfs.rollSize = 1048576
agent.sinks.hdfsSink.hdfs.rollCount = 10000

agent.sources.mysqlSource.channels = memoryChannel
agent.sinks.hdfsSink.channel = memoryChannel