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

Posted IT大狮兄

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了如何远程调试自定义开发的Flume应用相关的知识,希望对你有一定的参考价值。

一、前言

Flume作为当下最流行的大数据采集组件之一。其本身拥有分布式/高可靠/高可用等优点,但相比较于Flink/Spark/Kafka等大数据组件,其对于本地调试的功能支持度并不高,如果我们没有掌握Flume的远程调试要领,就只能不停的进行打日志,部署,打日志,部署这样低效的工作,而这对于程序员来说无异于折磨。所以今天小编就和大家一起来探究Flume的远程调试方法。

二、环境准备

  1. flink官网下载上传服务器并解压。

  2. 开发自定义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);
    }
}
  1. 编辑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
  1. 打包,打成成宽依赖包,即包含所有依赖。并上传到flume的lib目录下

三、环境配置

  1. 服务器环境配置

    • 修改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,没有使用则可忽略。

  1. 本地IDE(本例以Idea为准)环境配置

    • 编辑配置界面,添加remote

  • 配置remote

四、验证

  1. 启动flume agent,结果如下图所示,即代表配置没有问题:

  1. 启动debug程序,查看是否正常debug:

由上,我们看到程序正确的进入了断点,并查询到了mysql的记录。

五、总结

以上就是今天和大家分享的Flume的远程调试方法,如果不知道的小伙伴赶紧实践起来吧,提升效率,珍爱自己!

案例代码参考:flume_demo

以上是关于如何远程调试自定义开发的Flume应用的主要内容,如果未能解决你的问题,请参考以下文章

利用Eclipse对Flume进行远程调试分析环境搭建

VSCode插件开发全攻略代码片段设置自定义欢迎页

片段中ListView的android自定义适配器

JMeter 扩展开发:自定义 JMeter 插件的调试

如何在 SQL Developer 上远程调试 PLSQL 代码?

Flume学习系列---- Custom Interceptors(自定义拦截器)