如何遠(yuǎn)程調(diào)試自定義開發(fā)的Flume應(yīng)用

一、前言

Flume作為當(dāng)下最流行的大數(shù)據(jù)采集組件之一。其本身擁有分布式/高可靠/高可用等優(yōu)點(diǎn)，但相比較于Flink/Spark/Kafka等大數(shù)據(jù)組件，其對(duì)于本地調(diào)試的功能支持度并不高，如果我們沒有掌握Flume的遠(yuǎn)程調(diào)試要領(lǐng)，就只能不停的進(jìn)行打日志，部署，打日志，部署這樣低效的工作，而這對(duì)于程序員來說無異于折磨。所以今天小編就和大家一起來探究Flume的遠(yuǎn)程調(diào)試方法。

二、環(huán)境準(zhǔn)備

  1. flink官網(wǎng)下載上傳服務(wù)器并解壓。

  2. 開發(fā)自定義Source，這里以簡(jiǎn)單的讀取mysql表數(shù)據(jù)為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表的數(shù)據(jù)
 */
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生命周期結(jié)束，可以做一些保存等結(jié)束前的工作 */
    public void stop() {
        LOGGER.info("Mysql source stop......");
        if (conn != null) {
            try {
                conn.close();
            } catch (SQLException throwables) {
                LOGGER.error("連接關(guān)閉異常: {}", 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數(shù)據(jù)管道
a1.channels.c1.type = file
#最小需求空間
a1.channels.c1.minimumRequiredSpace = 3145728
#最大文件大小
a1.channels.c1.maxFileSize = 2146435071
#flume事件指針檢查點(diǎn)備份目錄
a1.channels.c1.checkpointDir = /opt/soft/flume/flume/data/checkpoint
#file-channel對(duì)event備份到本地的文件目錄
a1.channels.c1.dataDirs = /opt/soft/flume/data/file-channel-mysql/data
#文件管道中的數(shù)據(jù)容量，單位條數(shù)
a1.channels.c1.capacity = 200
#文件管道中的事務(wù)數(shù)據(jù)容量，單位條數(shù)
a1.channels.c1.transactionCapacity = 100
#檢查點(diǎn)備份flume時(shí)間指針的間隔時(shí)間
a1.channels.c1.checkpointInterval=60000

#############################
#          Sink
#############################
#本次測(cè)試重點(diǎn)在Source，所以sink用null即可，表示不輸出到任何地方
a1.sinks.k1.type = null

a1.sources.s1.channels = c1
a1.sinks.k1.channel = c1
  1. 打包，打成成寬依賴包，即包含所有依賴。并上傳到flume的lib目錄下

三、環(huán)境配置

  1. 服務(wù)器環(huán)境配置

    • 修改flume-ng啟動(dòng)命令文件: vim /opt/soft/flume/bin/flume-ng，修改為如下配置:
server配置1.png
// 端口默認(rèn)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，沒有使用則可忽略。
server配置2.png

  1. 本地IDE（本例以Idea為準(zhǔn)）環(huán)境配置

    • 編輯配置界面，添加remote
remote配置1.png
  • 配置remote
remote配置2.png

四、驗(yàn)證

  1. 啟動(dòng)flume agent，結(jié)果如下圖所示，即代表配置沒有問題：
debug_result1.png
  1. 啟動(dòng)debug程序，查看是否正常debug：
debug_result2.png

由上，我們看到程序正確的進(jìn)入了斷點(diǎn)，并查詢到了mysql的記錄。

五、總結(jié)

以上就是今天和大家分享的Flume的遠(yuǎn)程調(diào)試方法，如果不知道的小伙伴趕緊實(shí)踐起來吧，提升效率，珍愛自己！

案例代碼參考：flume_demo

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡(jiǎn)書系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容