flume source

Apache Flume是一个高可靠、高可用、分布式的大数据采集、聚合和传输系统，支持数据源的多样化，比如文件、JDBC、Syslog、Avro等。Flume的核心组件包括Source、Channel和Sink。其中，Source负责从数据源中读取数据并将其写入到Channel中，Channel负责缓存数据，而Sink则负责将数据从Channel中取出并将其发送到目标系统中。

在Flume中，Source是数据采集的入口，用于从数据源中读取数据并将其写入到Channel中。Flume提供了多种类型的Source组件，包括：

1. Avro Source：用于从Avro客户端接收数据。
2. Thrift Source：用于从Thrift客户端接收数据。
3. Spooling Directory Source：用于监控指定目录下的文件，并将其内容写入到Channel中。
4. Netcat Source：用于通过TCP/IP协议接收数据。
5. Syslog Source：用于从Syslog守护进程接收数据。
6. Exec Source：用于执行外部命令，并将其输出作为数据源。
7. HTTP Source：用于从HTTP客户端接收数据。

通过选择合适的Source组件，可以方便地实现对不同类型数据源的采集。同时，Flume也支持自定义Source组件，用户可以基于自己的需求进行扩展。

相关问题

flume source 自定义jdbc source

### 自定义 Flume JDBC Source 示例教程

#### 准备工作
为了创建自定义的 Flume JDBC Source，需先准备好开发环境。这包括但不限于安装 JDK 和 IDE（如 IntelliJ IDEA），以及配置 Maven 或 Gradle 构建工具来管理依赖项。

对于项目结构，在 `pom.xml` 文件中应加入必要的依赖库以支持与 MySQL 数据库交互的功能[^1]：

<dependencies>
    <!-- Apache Flume NG SDK -->
    <dependency>
        <groupId>org.apache.flume</groupId>
        <artifactId>flume-ng-sdk</artifactId>
        <version>${flume.version}</version>
    </dependency>

    <!-- MySQL Connector/J -->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>${mysql.connector.version}</version>
    </dependency>
</dependencies>

#### 创建自定义 Source 类
继承 `AbstractSource` 并实现 `Configurable`, `PollableSource` 接口可以构建一个轮询式的 JDBC Source。此类负责周期性查询数据库并将结果作为事件发送给下游组件处理。

public class JdbcPollingSource extends AbstractSource implements Configurable, PollableSource {
    
    private static final Logger logger = LoggerFactory.getLogger(JdbcPollingSource.class);
    private DataSource dataSource;
    private String querySql;

    @Override
    public void configure(Context context) {
        // 配置数据源连接参数和 SQL 查询语句
        Properties props = new Properties();
        props.setProperty("user", "root");
        props.setProperty("password", "your_password");

        try {
            dataSource = new DriverManagerDataSource(
                "jdbc:mysql://localhost:3306/testdb",
                props
            );
            querySql = "SELECT * FROM your_table WHERE processed_flag=0";
        } catch (Exception e) {
            throw new FlumeException("Failed to initialize database connection.", e);
        }
    }

    @Override
    public Status process() throws EventDeliveryException {
        List<Event> events = Lists.newArrayList();

        Connection conn = null;
        PreparedStatement stmt = null;
        ResultSet rs = null;

        try {
            conn = dataSource.getConnection();
            stmt = conn.prepareStatement(querySql);

            rs = stmt.executeQuery();
            
            while(rs.next()){
                Map<String, Object> rowMap = Maps.newHashMap();
                
                for(int i=1;i<=rs.getMetaData().getColumnCount();i++){
                    rowMap.put(rs.getMetaData().getColumnName(i), rs.getObject(i));
                }

                byte[] bodyBytes = JsonMapper.getInstance().writeValueAsBytes(rowMap);
                Event event = new SimpleEvent();
                event.setBody(bodyBytes);
                
                events.add(event);
            }

            getChannelProcessor().processEventBatch(events); 

            return Status.READY;
        } catch (SQLException | IOException ex){
            logger.error(ex.getMessage(),ex);
            return Status.BACKOFF;
        } finally{
            closeQuietly(conn,stmt,rs);
        }
    }

    private void closeQuietly(Connection c, Statement s, ResultSet r){
        if(r!=null){try{r.close();}catch(Exception ignored){}}
        if(s!=null){try{s.close();}catch(Exception ignored){}}
        if(c!=null){try{c.close();}catch(Exception ignored){}}
    }
}

上述代码展示了如何通过 Java 反射机制读取表中的记录并将其转换成 JSON 字符串形式存入 Flume 事件体中。注意这里假设每条记录都对应于单个 Flume 事件；实际应用时可根据业务逻辑调整这一映射关系。

#### 注册插件类路径
为了让 Flume 能够识别新创建的数据源类型，还需要修改 `$FLUME_HOME/conf/flume-env.sh` 文件，添加如下内容指向包含自定义 source jar 包的位置:

export FLUME_CLASSPATH=/path/to/your/custom-source.jar:$FLUME_CLASSPATH

最后一步是在 Flume Agent 的配置文件里声明使用该自定义 source，并提供相应的初始化参数:

agent.sources = jdbc_source
agent.channels = memory_channel
agent.sinks = logger_sink

agent.sources.jdbc_source.type = com.example.JdbcPollingSource
agent.sources.jdbc_source.query_sql = SELECT id,name,value FROM mytable LIMIT 100 OFFSET ${offset}

agent.channels.memory_channel.type = memory

agent.sinks.logger_sink.type = logger

agent.sources.jdbc_source.channels = memory_channel
agent.sinks.logger_sink.channel = memory_channel

以上即为完整的自定义 Flume JDBC Source 开发指南。希望这些信息能够帮助理解整个过程和技术细节。

flume avro source

### Flume 中 Avro Source 的配置与使用

#### 配置文件结构
为了使 Avro Source 正常工作，在 `flume-conf.properties` 文件中需定义如下属性：

# 定义agent名称以及source、channel和sink组件
a1.sources = r1
a1.channels = c1
a1.sinks = k1

# 设置source类型为Avro并指定其监听IP地址及端口号
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 41414

上述配置指定了名为 `r1` 的 Avro Source 将会监听所有网络接口上的 41414 端口等待客户端连接发送事件[^1]。

#### 进阶配置选项
除了基本的 IP 和端口外，还可以调整更多参数优化性能或安全性：

- **线程数控制**

    a1.sources.r1.threads = 8

上述设置允许同时处理多达八个并发请求，提高吞吐量[^3]。

- **启用SSL加密通信**

当需要保障数据传输安全时可以开启 SSL 支持:

    a1.sources.r1.ssl = true
    a1.sources.r1.keystore = /path/to/keystore.jks
    a1.sources.r1.keystore-password = secret
    a1.sources.r1.keystore-type = JKS

注意替换实际路径和密码值以匹配环境需求。

- **压缩算法选择**

对于带宽受限场景下推荐采用压缩技术减少流量消耗:

    a1.sources.r1.compression-type = zlib

只接受 "zlib" 或者留空表示不压缩；注意不是所有的接收方都支持解压功能所以要确保两端兼容性良好。

#### 使用示例
假设有一个简单的应用场景：通过命令行工具向运行中的 Flume Agent 发送 JSON 文档作为测试消息。此时只需安装 Apache Flume 并启动带有适当配置的服务实例即可开始实验。

echo '{"name":"John","age":30}' | nc localhost 41414

这条指令利用 Netcat (`nc`) 工具模拟了一个外部程序向本地主机上正在监听特定端口 (此处假定为 41414) 的 Avro Source 推送了一条记录[^2]。