如何远程调试自定义开发的Flume应用 flumejava

一、前言
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 = s1 a1.sinks = k1 a1.channels = c1############################# #Source ############################# #自定义MySQL source类 a1.sources.s1.type = org.bigwinner.flume.sources.MysqlSource a1.sources.s1.mysql.driver = com.mysql.jdbc.Driver a1.sources.s1.mysql.url = jdbc:mysql://lsl001:3306/redis_temp a1.sources.s1.mysql.user = superboy a1.sources.s1.mysql.password = iamsuperboy a1.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 = c1 a1.sinks.k1.channel = c1