一、前言
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 = null
a1.sources.s1.channels = c1
a1.sinks.k1.channel = c1- 打包，打成成宽依赖包，即蕴含所有依赖。并上传到flume的lib目录下
三、环境配置
-
服务器环境配置
- 批改flume-ng启动命令文件: vim /opt/soft/flume/bin/flume-ng，批改为如下配置:
// 端口默认8000
JAVA_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
发表回复