如何遠端除錯自定義開發的Flume應用

一、前言

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

如何遠端除錯自定義開發的Flume應用

如何遠端除錯自定義開發的Flume應用

一、前言

二、環境準備

三、環境配置

四、驗證

五、總結

最新文章