用java语言编写flume采集数据到mysql的MysqlSink

可以按照以下步骤使用Java编写Flume采集数据到MySQL的MysqlSink：

1. 首先，需要在pom.xml文件中添加以下依赖：

<dependency>
    <groupId>org.apache.flume</groupId>
    <artifactId>flume-ng-sdk</artifactId>
    <version>1.9.0</version>
</dependency>
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>8.0.23</version>
</dependency>

2. 创建一个类，命名为MysqlSink，实现org.apache.flume.sink.AbstractSink接口。

3. 在类中定义一个Connection对象和PreparedStatement对象，用于连接到MySQL数据库和执行SQL语句。

private Connection connection;
private PreparedStatement preparedStatement;

4. 在类的构造方法中，初始化连接和预处理语句对象。

public MysqlSink() {
    // 初始化连接
    String url = "jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8";
    String user = "root";
    String password = "123456";
    try {
        Class.forName("com.mysql.cj.jdbc.Driver");
        connection = DriverManager.getConnection(url, user, password);
    } catch (ClassNotFoundException | SQLException e) {
        e.printStackTrace();
    }

    // 初始化预处理语句对象
    String sql = "INSERT INTO `test`.`flume_data` (`timestamp`, `message`) VALUES (?, ?)";
    try {
        preparedStatement = connection.prepareStatement(sql);
    } catch (SQLException e) {
        e.printStackTrace();
    }
}

5. 实现process()方法，该方法接收一个事件(Event)对象，从中获取数据，然后执行SQL语句将数据插入到MySQL数据库中。

@Override
public Status process() throws EventDeliveryException {
    Status status = null;
    // 获取Channel
    Channel channel = getChannel();
    // 从Channel中获取事务
    Transaction transaction = channel.getTransaction();
    try {
        // 开始事务
        transaction.begin();
        // 从Channel中获取事件
        Event event = channel.take();
        // 获取事件中的数据
        byte[] body = event.getBody();
        String message = new String(body);
        // 将数据插入到MySQL数据库中
        preparedStatement.setTimestamp(1, new Timestamp(System.currentTimeMillis()));
        preparedStatement.setString(2, message);
        preparedStatement.executeUpdate();
        // 提交事务
        transaction.commit();
        status = Status.READY;
    } catch (SQLException e) {
        e.printStackTrace();
        // 回滚事务
        transaction.rollback();
        status = Status.BACKOFF;
    } catch (Exception e) {
        e.printStackTrace();
        // 回滚事务
        transaction.rollback();
        status = Status.BACKOFF;
    } finally {
        // 结束事务
        transaction.close();
    }
    return status;
}

6. 在Flume配置文件中添加MysqlSink的配置信息，例如：

a1.sources = r1
a1.sinks = k1
a1.channels = c1

a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444

a1.sinks.k1.type = com.example.MysqlSink
a1.sinks.k1.channel = c1

a1.channels.c1.type = memory
a1.channels.c1.capacity = 100
a1.channels.c1.transactionCapacity = 100

7. 启动Flume，然后使用netcat命令向Flume发送数据，例如：

$ nc localhost 44444
Hello, world!

8. 查看MySQL数据库中的数据，可以发现已经成功插入了一条数据。

以上就是使用Java编写Flume采集数据到MySQL的MysqlSink的步骤。