【亲测免费】 Flume-ng-sql-source 使用教程

1. 项目介绍

flume-ng-sql-source 是一个Apache Flume的扩展插件，允许Flume从SQL数据库中收集数据。项目集成了Hibernate，因此理论上支持所有Hibernate兼容的数据库引擎。它能够处理增量查询，且输出格式可配置为JSON。

2. 项目快速启动

依赖安装

首先确保已经安装了Apache Flume和Maven。

下载项目

克隆项目仓库：

git clone https://github.com/keedio/flume-ng-sql-source.git

编译与打包

进入项目目录并执行以下命令来编译和创建JAR包：

cd flume-ng-sql-source
mvn package

部署

将生成的JAR包复制到Flume的插件目录：

cp target/flume-ng-sql-source-*.jar $FLUME_HOME/plugins.d/

创建配置文件

创建一个Flume配置文件，例如 sql-source.conf：

agent.sources = sqlSrc
agent.channels = memChannel
agent.sinks = hdfsSink

agent.sources.sqlSrc.type = org.keedio.flume.source.SQLSource
agent.sources.sqlSrc.connectionURL = jdbc:mysql://localhost:3306/mydb?useSSL=false
agent.sources.sqlSrc.username = myuser
agent.sources.sqlSrc.password = password
agent.sources.sqlSrc.query = SELECT * FROM mytable WHERE timestamp > ${lastEventTime}
agent.sources.sqlSrc.resultHandler.className = org.keedio.flume.handler.JsonEventHandler
agent.sources.sqlSrc.useLastEventTimeAsTimestamp = true

agent.sources.sqlSrc.interceptors = tsi
agent.sources.sqlSrc.interceptors.tsi.type = timestamp
agent.sources.sqlSrc.interceptors.tsi.timestampFormat = yyyy-MM-dd'T'HH:mm:ss.SSSXXX

agent.channels.memChannel.type = memory
agent.channels.memChannel.capacity = 1000
agent.channels.memChannel.transactionCapacity = 100

agent.sinks.hdfsSink.type = hdfs
agent.sinks.hdfsSink.channel = memChannel
agent.sinks.hdfsSink.hdfs.path = hdfs://namenode:9000/user/flume/events
agent.sinks.hdfsSink.hdfs.filePrefix = events-
agent.sinks.hdfsSink.hdfs.fileSuffix = .txt
agent.sinks.hdfsSink.hdfs.writeFormat = Text
agent.sinks.hdfsSink.hdfs.batchSize = 100
agent.sinks.hdfsSink.hdfs.rollInterval = 3600
agent.sinks.hdfsSink.hdfs.rollCount = 0
agent.sinks.hdfsSink.hdfs.rollSize = 0

启动Flume Agent

运行Flume Agent：

flume-ng agent --conf ./conf --conf-file sql-source.conf --name agent -Dflume.root.logger=INFO,console

3. 应用案例和最佳实践

实时日志监控

使用flume-ng-sql-source从数据库的日志表中收集信息，实时监控系统状态，异常报警等。

数据迁移

在数据库升级或迁移过程中，利用该工具实现旧库到新库的数据实时同步。

流式分析

结合Hadoop生态系统中的其他工具（如Spark Streaming），实现实时数据分析和流式处理。

最佳实践

• 配置增量查询：通过设置特定的时间字段以避免重复读取数据。
• 错误处理：为源添加适当的拦截器，以便捕获并处理SQL查询可能遇到的错误。
• 性能优化：调整Flume配置参数，如通道容量和事务大小，以适应负载需求。

4. 典型生态项目

• Apache Flume: 数据收集框架，用于构建可靠、分布式的日志聚合系统。
• Apache Hadoop: 提供大规模数据集的存储和计算能力，是Flume数据流入的主要目的地之一。
• Apache Kafka: 引入Kafka作为中介，可以在Flume与Hadoop之间提供更高的吞吐量和更灵活的数据传输。
• HBase: 结合HDFS使用，提供NoSQL数据库服务，可用于存储Flume采集的实时数据。
• Apache Spark: 提供流处理和批处理能力，可以与Flume整合，进行实时分析和处理。