一、前言
Flume作为当下最风行的大数据采集组件之一。其自身领有分布式/高牢靠/高可用等长处，但相比拟于Flink/Spark/Kafka等大数据组件，其对于本地调试的性能反对度并不高，如果咱们没有把握Flume的近程调试要领，就只能不停的进行打日志，部署，打日志，部署这样低效的工作，而这对于程序员来说无异于折磨。所以明天小编就和大家一起来探索Flume的近程调试办法。
二、环境筹备
- flink官网下载上传服务器并解压。
- 开发自定义Source，这里以简略的读取mysql表数据为demo，局部代码如下：
package org.bigwinner.flume.sources;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.EventBuilder;import org.apache.flume.source.AbstractSource;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import java.sql.*;/** * @author: IT大狮兄 * @date: 2021/8/13 下午9:11 * @version: 1.0.0 * @description: 自定义Source--读取MySQL表的数据 */public class MysqlSource extends AbstractSource implements PollableSource, Configurable { private static final Logger LOGGER = LoggerFactory.getLogger(MysqlSource.class); private String mysqlUrl; private String mysqlUser; private String mysqlPassword; private String mysqlTable; private String mysqlDriver; private Connection conn = null; public Status process() throws EventDeliveryException { String sql = "select * from " + mysqlTable; try { PreparedStatement statement = conn.prepareStatement(sql); ResultSet resultSet = statement.executeQuery(); while (resultSet.next()) { String id = resultSet.getString(1); String uuid = resultSet.getString(2); String iccid = resultSet.getString(3); byte[] eventBytes = new StringBuilder().append(id).append("--") .append(uuid).append("--").append(iccid).toString().getBytes(); Event event = EventBuilder.withBody(eventBytes); getChannelProcessor().processEvent(event); } } catch (SQLException throwables) { throwables.printStackTrace(); } return Status.READY; } public long getBackOffSleepIncrement() { return 0; } public long getMaxBackOffSleepInterval() { return 0; } @Override /** Flume生命周期开始，能够做一些初始化的工作 */ public void start() { LOGGER.info("Mysql source start......"); try { Class.forName(mysqlDriver); conn = DriverManager.getConnection(mysqlUrl, mysqlUser, mysqlPassword); } catch (ClassNotFoundException e) { LOGGER.error("Driver class is not found!"); } catch (SQLException throwables) { LOGGER.error("get the connection error: {}", throwables); } } @Override /** Flume生命周期完结，能够做一些保留等完结前的工作 */ public void stop() { LOGGER.info("Mysql source stop......"); if (conn != null) { try { conn.close(); } catch (SQLException throwables) { LOGGER.error("连贯敞开异样: {}", throwables); } } super.stop(); } /** Flume配置文件读取的办法 */ public void configure(Context context) { mysqlUrl = context.getString("mysql.url", ""); mysqlUser = context.getString("mysql.user", ""); mysqlPassword = context.getString("mysql.password", ""); mysqlTable = context.getString("mysql.table", ""); LOGGER.info("mysql_driver: {} --> mysql_url: {} --> mysql_user: {} --> mysql_password: {} --> mysql_table: {}", mysqlDriver, mysqlUrl, mysqlUser, mysqlPassword, mysqlTable); }}- 编辑flume agent配置文件，并上传到flume conf目录下
a1.sources = s1a1.sinks = k1a1.channels = c1############################## Source##############################自定义MySQL source类a1.sources.s1.type = org.bigwinner.flume.sources.MysqlSourcea1.sources.s1.mysql.driver = com.mysql.jdbc.Drivera1.sources.s1.mysql.url = jdbc:mysql://lsl001:3306/redis_tempa1.sources.s1.mysql.user = superboya1.sources.s1.mysql.password = iamsuperboya1.sources.s1.mysql.table = redis_temp############################## Channel##############################配置file-channel数据管道a1.channels.c1.type = file#最小需要空间a1.channels.c1.minimumRequiredSpace = 3145728#最大文件大小a1.channels.c1.maxFileSize = 2146435071#flume事件指针检查点备份目录a1.channels.c1.checkpointDir = /opt/soft/flume/flume/data/checkpoint#file-channel对event备份到本地的文件目录a1.channels.c1.dataDirs = /opt/soft/flume/data/file-channel-mysql/data#文件管道中的数据容量，单位条数a1.channels.c1.capacity = 200#文件管道中的事务数据容量，单位条数a1.channels.c1.transactionCapacity = 100#检查点备份flume工夫指针的间隔时间a1.channels.c1.checkpointInterval=60000############################## Sink##############################本次测试重点在Source，所以sink用null即可，示意不输入到任何中央a1.sinks.k1.type = nulla1.sources.s1.channels = c1a1.sinks.k1.channel = c1- 打包，打成成宽依赖包，即蕴含所有依赖。并上传到flume的lib目录下
三、环境配置
服务器环境配置
- 批改flume-ng启动命令文件: vim /opt/soft/flume/bin/flume-ng，批改为如下配置:
// 端口默认8000JAVA_OPTS="-Xmx500m -Xdebug -Xrunjdwp:transport=dt_socket,address=8000,server=y,suspend=y"- 如果flume-env.sh文件在应用，须要正文掉flume-env.sh的JAVA_OPTS配置：vim /opt/soft/flume/conf/flume-env.sh，没有应用则可疏忽。
本地IDE（本例以Idea为准）环境配置
- 编辑配置界面，增加remote
- 配置remote
四、验证
- 启动flume agent，后果如下图所示，即代表配置没有问题：
- 启动debug程序，查看是否失常debug：
由上，咱们看到程序正确的进入了断点，并查问到了mysql的记录。
五、总结
以上就是明天和大家分享的Flume的近程调试办法，如果不晓得的小伙伴连忙实际起来吧，晋升效率，珍视本人！
案例代码参考：flume_demo