前一篇文章《Flume简介与安装》已经介绍了Flume的相关知识，也提到了Flume的各个环节是解耦的，那么如果实际开发需要自定义开发组件呢？大概流程是怎样的？

下面以输出到PostgreSQL为例说明开发新组件的方法。

首先简单介绍PostgreSQL：

PostgreSQL是一种特性非常齐全的自由软件的对象-关系型数据库管理系统（ORDBMS），是以加州大学计算机系开发的POSTGRES，4.2版本为基础的对象关系型数据库管理系统。POSTGRES的许多领先概念只是在比较迟的时候才出现在商业网站数据库中。PostgreSQL支持大部分的SQL标准并且提供了很多其他现代特性，如复杂查询、外键、触发器、视图、事务完整性、多版本并发控制等。同样，PostgreSQL也可以用许多方法扩展，例如通过增加新的数据类型、函数、操作符、聚集函数、索引方法、过程语言等。另外，因为许可证的灵活，任何人都可以以任何目的免费使用、修改和分发PostgreSQL。

接下来就是开发过程了：

1. 核心代码

sink的实现需要start()、stop()、process()和configure()这四个方法，分别实现启动、结束、处理和取参数的功能。

具体代码如下：

package org.apache.flume.sink.pgsql;

import com.google.common.base.Preconditions;
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.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;

// 类需要继承AbstractSink这个类和Configurable这个接口
public class PGSQLSink extends AbstractSink implements Configurable {
    // 定义相关对象和变量
    private static final Logger logger = LoggerFactory.getLogger(PGSQLSink.class);
    private static final String DEFAULT_SPLIT_CHARS = ",";
    private Connection connection;
    private Statement statement;
    private String hostname;
    private String port;
    private String url;
    private String user;
    private String password;
    private String tablename;
    private String columnName;

// configure方法作用是从配置文件取出相关参数，包括数据库连接等信息
    @Override
    public void configure(Context context) {
        columnName = context.getString("column_name");
        Preconditions.checkNotNull(columnName, "column_name 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!!");
        hostname = context.getString("hostname");
        Preconditions.checkNotNull(hostname, "hostname must be set!!");
        port = context.getString("port");
        Preconditions.checkNotNull(port, "port must be set!!");
        url = "jdbc:postgresql://" + hostname + ":" + port + "/postgres";
        Preconditions.checkNotNull(url, "url must be set!!");
        tablename = context.getString("tablename");
        Preconditions.checkNotNull(tablename, "tablename must be set!!");
    }

    // 在整个sink结束时执行一遍
    @Override
    public synchronized void stop() {
        super.stop();
    }

    // 在整个sink开始时执行一遍
    @Override
    public synchronized void start() {
        super.start();
        try {
            connection = DriverManager.getConnection(url, user, password);
            logger.info("pgsql connected!!!"+connection);
            // 连接URL为 jdbc:mysql//服务器地址/数据库名 ，后面的2个参数分别是登陆用户名和密码
            statement = connection.createStatement();
            logger.info("statement created****"+statement);
        } catch (SQLException e) {
            // 处理异常
            e.printStackTrace();
        }
    }

    // 不断循环调用
    @Override
    public Status process() throws EventDeliveryException{

        Status result = Status.READY;
        Channel channel = getChannel();
        Transaction transaction = channel.getTransaction();
        Event event = null;

        try {
            transaction.begin();
            // 从channel取出event
            event = channel.take();

            if (event != null) {
                String body = new String(event.getBody(),"UTF-8");
                System.out.println("body: "+body);
                System.out.println("columnName: "+columnName);
                // 拼凑出SQL语句
                String sql = "insert into " + tablename + "(" + columnName + ") values(" + body + ")";
                logger.info(sql);
                // 执行SQL语句
                statement.executeUpdate(sql);
                logger.info("sql execute done.");
            } else {
                // event为空的处理
                result = Status.BACKOFF;
            }
            transaction.commit();
        } catch (Exception ex) {
            // 处理异常
            transaction.rollback();
            throw new EventDeliveryException("Failed to log event: " + event, ex);
        } finally {
            transaction.close();
        }
    // 返回状态
    return result;
    }
}

上面的代码编写好之后，需要将其打包为一个jar包（如flume-pgsql-sink-1.6.0.jar）并将其放到Flume安装目录下的lib目录里；另外，pgsql的驱动jar（如postgresql-42.2.11.jar）要加到该目录下，否则Flume将无法使用该功能。

2.配置文件

为了配合代码，需要在Flume安装目录下的conf文件夹中新建一个名为pgsql.conf的文件，将组件的需要用到的参数定义好：

# a1是agent的名称
# r1是source的名称
# k1是sink的名称
# c1是channel的名称
a1.sources = r1
a1.sinks = k1
a1.channels = c1


# 配置source
# 这里是从testData.log文件中采集数据
a1.sources.r1.type = exec
a1.sources.r1.command = tail -f /root/mycode/testData.log
a1.sources.r1.shell = /bin/sh -c


# 配置sink，包括类名、posgresql的ip地址端口和账号密码、表名、字段名
a1.sinks.k1.type = org.apache.flume.sink.pgsql.PGSQLSink
a1.sinks.k1.hostname = 192.168.38.132
a1.sinks.k1.port = 5432
a1.sinks.k1.password = postgres
a1.sinks.k1.user = postgres
a1.sinks.k1.tablename= testcar
a1.sinks.k1.column_name = id,price


# 配置channel的类型
a1.channels.c1.type = file


#  关联好source、sink和channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

3. 运行命令：

到bin文件夹下面执行，指定好agent，配置文件和日志类型：

[root@slave2 bin]# flume-ng agent --name a1 -c conf -f ../conf/pgsql.conf -Dflume.root.logger=INFO,console

新开一个窗口，到/root/mycode/testData.log文件中追加记录：

echo "1, 200" >> /root/mycode/testData.log
echo "2, 300" >> /root/mycode/testData.log

然后在PostgreSQL的testcar表就可以看到数据导入了，大功告成！

本文转自微信公众号：superdiao的果壳

喜欢的朋友可以去关注（关注后有福利哦~），记得先给本文点赞或者点喜欢呐！

http://weixin.qq.com/r/szlobI7EPkZCKcoPb2w3 (二维码自动识别)

ITPub博客

Flume自定义Sink到PostgreSQL

最新文章