flume 采集mysql

Flume 是一个分布式、可靠且可用的服务，用于高效地收集、聚合和传输大量日志数据。它具有容错性和可扩展性，能够将数据从一个位置移动到另一个位置，通常用于日志聚合、数据传输等场景。

基础概念

Flume 的核心概念包括：

Source：数据的来源，负责接收数据。
Channel：临时存储数据的组件，确保数据在传输过程中的可靠性。
Sink：数据的去向，负责将数据发送到目标位置。
Agent：包含 Source、Channel 和 Sink 的独立运行单元。

类型

Flume 的 Source 类型包括：

Avro Source：接收 Avro 数据。
Exec Source：执行外部命令并捕获输出。
JMS Source：从 JMS 消息队列接收数据。
Spooling Directory Source：从指定目录读取文件。

Sink 类型包括：

HDFS Sink：将数据写入 HDFS。
Kafka Sink：将数据发送到 Kafka。
Logger Sink：将数据输出到日志文件。
Avro Sink：将数据发送到 Avro 端点。

应用场景

Flume 常用于以下场景：

日志聚合：从多个服务器收集日志并集中存储。
数据传输：将数据从一个系统传输到另一个系统，如从数据库传输到数据仓库。
实时监控：收集实时数据并进行分析。

MySQL 数据采集

Flume 可以通过自定义 Source 来采集 MySQL 数据。以下是一个简单的示例：

自定义 MySQL Source

创建一个 Maven 项目，添加 Flume 和 JDBC 依赖。
编写自定义 Source：

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;
    }
}

配置 Flume Agent：

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