Flume的开发之自定义 source 自定义 sink 自定义拦截器

秒速五厘米 2024-02-19 18:52 102阅读 0赞

### 一：开发相关 ###

**加入 jar 包依赖：**

<dependency>
    <groupId>org.apache.flume</groupId>
    <artifactId>flume-ng-core</artifactId>
    <version>1.5.2</version>
    </dependency>
    <dependency>
    <groupId>org.apache.flume</groupId>
    <artifactId>flume-ng-configuration</artifactId>
    <version>1.5.2</version>
    </dependency

### 二：自定义 source ###

比如有一个需求，需要读取指定文件，要求读取4行才发送给channel

package package package com.gds.flume.source;
    import import import import java.io.FileNotFoundException;
    import import import import java.io.IOException;
    import import import import java.io.RandomAccessFile;
    import import import import java.util.concurrent.atomic.AtomicLong;
    import import import import org.apache.flume.Context;
    import import import import org.apache.flume.Event;
    import import import import org.apache.flume.EventDeliveryException;
    import import import import org.apache.flume.PollableSource;
    import import import import org.apache.flume.conf.Configurable;
    import import import import org.apache.flume.event.EventBuilder;
    import import import import org.apache.flume.source.AbstractSource;
    import import import import org.slf4j.Logger;
    import import import import org.slf4j.LoggerFactory;
    /**
    * 从某个日子文件中每次读取4行，发送给channel
    *
    * @author @author @author @author dczhao
    *
    */
    public public public public class class class class MySource extends extends extends extends AbstractSource implements implements implements implements Configurable,
    PollableSource {
    private private private private static static static static final final final final Logger logger = LoggerFactory
    .getLogger(MySource.class class class class);
    /**
    * 要读的日志文件地址
    */
    private private private private String logPath;
    private private private private RandomAccessFile file;
    private private private private AtomicLong autoLine;
    private private private private StringBuilder oldLine = new new new new StringBuilder();;
    @Override
    public public public public synchronized synchronized synchronized synchronized void void void void start() {
    logger.info("MySource starting~~~~~");
    autoLine = new new new new AtomicLong(0);
    try try try try {
    file = new new new new RandomAccessFile(logPath, "r");
    } catch catch catch catch (FileNotFoundException e) {
    logger.error("file oper error~~~~~", e);
    }
    super super super super.start();
    }
    @Override
    public public public public synchronized synchronized synchronized synchronized void void void void stop() {
    try try try try {
    file.close();
    } catch catch catch catch (IOException e) {
    logger.error("file close error~~~~~", e);
    }
    autoLine = new new new new AtomicLong(0);
    super super super super.stop();
    }
    public public public public void void void void configure(Context context) {
    this this this this.logPath = context.getString("logPath");
    }
    public public public public Status process() throws throws throws throws EventDeliveryException {
    try try try try {
    String line = file.readLine();
    ifififif(line==null null null null){
    return return return return Status.BACKOFF;
    }
    ifififif(autoLine.intValue()%4==0){
    logger.info("oldLine:" + oldLine.toString());
    byte byte byte byte[] body = oldLine.toString().getBytes();
    Event event = EventBuilder.withBody(body);
    getChannelProcessor().processEvent(event);
    oldLine = new new new new StringBuilder();
    autoLine=new new new new AtomicLong(0);
    }
    oldLine.append(line);
    } catch catch catch catch (IOException e) {
    logger.error("getFilePointer error~~~~~", e);
    return return return return Status.BACKOFF;
    }
    return return return return Status.READY;
    }
    }

其实要学会怎么自定义 source，可以参考 flume 提供的相关 source类，比如 AvroSource、NetcatSource、SpoolDirectorySource 等。

### 三：自定义 sink ###

从 channel 中读取数据，推送到 MySQL 数据库保存数据为例

ackage package package package com.gds.flume.sink;
    import import import import com.google.common.base.Preconditions;
    import import import import com.google.common.base.Throwables;
    import import import import com.google.common.collect.Lists;
    import import import import org.apache.flume.*;
    import import import import org.apache.flume.conf.Configurable;
    import import import import org.apache.flume.sink.AbstractSink;
    import import import import org.slf4j.Logger;
    import import import import org.slf4j.LoggerFactory;
    import import import import java.sql.Connection;
    import import import import java.sql.DriverManager;
    import import import import java.sql.PreparedStatement;
    import import import import java.sql.SQLException;
    import import import import java.util.List;
    public public public public class class class class MysqlSink extends extends extends extends AbstractSink implements implements implements implements Configurable {
    private private private private Logger LOG = LoggerFactory.getLogger(MysqlSink.class class class class);
    private private private private String hostname;
    private private private private String port;
    private private private private String databaseName;
    private private private private String tableName;
    private private private private String user;
    private private private private String password;
    private private private private PreparedStatement preparedStatement;
    private private private private Connection conn;
    private private private private int int int int batchSize;
    public public public public MysqlSink() {
    LOG.info("MysqlSink start...");
    }
    public public public public void void void 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 public public public void void void void start() {
    super super super super.start();
    try try try try {
    //调用Class.forName()方法加载驱动程序
    Class.forName("com.mysql.jdbc.Driver");
    } catch catch catch catch (ClassNotFoundException e) {
    e.printStackTrace();
    }
    String url = "jdbc:mysql://" + hostname + ":" + port + "/" + databaseName;
    //调用DriverManager对象的getConnection()方法，获得一个Connection对象
    try try try try {
    conn = DriverManager.getConnection(url, user, password);
    conn.setAutoCommit(false false false false);
    //创建一个Statement对象
    preparedStatement = conn.prepareStatement("insert into " + tableName +
    " (content) values (?)");
    } catch catch catch catch (SQLException e) {
    e.printStackTrace();
    System.exit(1);
    }
    }
    @Override
    public public public public void void void void stop() {
    super super super super.stop();
    ifififif (preparedStatement != null null null null) {
    try try try try {
    preparedStatement.close();
    } catch catch catch catch (SQLException e) {
    e.printStackTrace();
    }
    }
    ifififif (conn != null null null null) {
    try try try try {
    conn.close();
    } catch catch catch catch (SQLException e) {
    e.printStackTrace();
    }
    }
    }
    public public public public Status process() throws throws throws 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 try try try {
    for for for for (int int int int i = 0; i < batchSize; i++) {
    event = channel.take();
    ifififif (event != null null null null) {
    content = new new new new String(event.getBody());
    actions.add(content);
    } else else else else {
    result = Status.BACKOFF;
    break break break break;
    }
    }
    ifififif (actions.size() > 0) {
    preparedStatement.clearBatch();
    for for for for (String temp : actions) {
    preparedStatement.setString(1, temp);
    preparedStatement.addBatch();
    }
    preparedStatement.executeBatch();
    conn.commit();
    }
    transaction.commit();
    } catch catch catch catch (Throwable e) {
    try try try try {
    transaction.rollback();
    } catch catch catch 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 finally finally finally {
    transaction.close();
    }
    return return return return result;
    }
    }

其实要学会怎么自定义 sink，可以参考 flume 提供的相关 sink 类，  
比如 LoggerSink、AvroSink 等。

### 四：自定义拦截器 ###

### 五：其他说明 ###

1、使用 Spooling Directory Source 的时候，一定要避免同时读写 一 个 文 件 的 情 况 。 可 以 通 过 source1.ignorePattern =^(.)\*\\.tmp$这个配置，让 spoolingsource 不读取该格式的文件。  
.

### 六：参考文档 ###

[http://blog.csdn.net/xiao\_jun\_0820/article/category/2399621][http_blog.csdn.net_xiao_jun_0820_article_category_2399621] 这篇博客比较全

[http_blog.csdn.net_xiao_jun_0820_article_category_2399621]: http://blog.csdn.net/xiao_jun_0820/article/category/2399621