Sink 不斷地輪詢 Channel 中的事件且批量地移除它們,并将這些事件批量寫入到存儲或索引系統、或者被發送到另一個 Flume Agent。
Sink 是完全事務性的。
在從 Channel 批量删除資料之前,每個 Sink 用 Channel 啟動一個事務。
批量事件一旦成功寫出到存儲系統或下一個 Flume Agent,Sink 就利用 Channel 送出事務。
事務一旦被送出,該 Channel 從自己的内部緩沖區删除事件。
Sink 元件目的地包括 hdfs、logger、avro、thrift、ipc、file、null、HBase、solr、自定義。官方提供的 Sink 類型已經很多,但是有時候并不能滿足實際開發當中的需求,此時我們就需要根據實際需求自定義某些 Sink。
官方也提供了自定義 sink 的接口:https://flume.apache.org/FlumeDeveloperGuide.html#sink
根據官方說明自定義 Sink 需要繼承 AbstractSink 類并實作 Configurable 接口。
實作相應方法:
// 初始化 context(讀取配置檔案内容)
configure(Context context);
// 從 Channel 讀取擷取資料(event),這個方法将被循環調用
process();
使用場景:讀取 Channel 資料寫入 MySQL 或者其他檔案系統。
使用 flume 接收資料,并在 Sink 端給每條資料添加字首和字尾,輸出到控制台。前字尾可在 flume 任務配置檔案中配置。
一、建立自定義 Sink
1.添加 pom 依賴
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com</groupId>
<artifactId>flume</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.apache.flume</groupId>
<artifactId>flume-ng-core</artifactId>
<version>1.9.0</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.8</source>
<target>1.8</target>
<encoding>UTF-8</encoding>
</configuration>
</plugin>
</plugins>
</build>
</project>
2.編寫自定義的 Sink 類
package sink;
import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class MySink extends AbstractSink implements Configurable {
// 建立 Logger 對象
private static final Logger LOG = LoggerFactory.getLogger(AbstractSink.class);
private String prefix;
private String suffix;
/**
* 1.擷取 Channel
* 2.從 Channel 擷取事務和資料
* 3.發送資料
*/
@Override
public Status process() throws EventDeliveryException {
// 聲明傳回值狀态資訊
Status status;
// 擷取目前 Sink 綁定的 Channel
Channel ch = getChannel();
// 擷取事務
Transaction txn = ch.getTransaction();
// 聲明事件
Event event;
// 開啟事務
txn.begin();
// 讀取 Channel 中的事件,直到讀取到事件結束循環
while (true) {
event = ch.take();
if (event != null) {
break;
}
}
try {
// 處理事件(列印)
LOG.info(prefix + new String(event.getBody()) + suffix);
// 事務送出
txn.commit();
status = Status.READY;
} catch (Exception e) {
// 遇到異常,事務復原
txn.rollback();
status = Status.BACKOFF;
} finally {
// 關閉事務
txn.close();
}
return status;
}
@Override
public void configure(Context context) {
// 讀取配置檔案内容,有預設值
prefix = context.getString("prefix", "hello:");
// 讀取配置檔案内容,無預設值
suffix = context.getString("suffix");
}
@Override
public void start() {
// Initialize the connection to the external repository (e.g. HDFS) that this Sink will forward Events to ..
// 初始化與外部存儲庫(例如HDFS)的連接配接,此接收器會将事件轉發到。
}
@Override
public void stop () {
// Disconnect from the external respository and do any additional cleanup (e.g. releasing resources or nulling-out field values) ..
// 斷開與外部存儲庫的連接配接,然後進行其他任何清理操作(例如,釋放資源或清空字段值)。
}
}
二、打包測試
1.打包上傳
參考:https://www.cnblogs.com/jhxxb/p/11582804.html
2.編寫 flume 配置檔案
mysink.conf
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = 127.0.0.1
a1.sources.r1.port = 4444
# Describe the sink
a1.sinks.k1.type = sink.MySink
# a1.sinks.k1.prefix = jhxxb:
a1.sinks.k1.suffix = :end
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
啟動
cd /opt/apache-flume-1.9.0-bin
bin/flume-ng agent --conf conf/ --name a1 --conf-file /tmp/flume-job/sink/mysink.conf -Dflume.root.logger=INFO,console
向監聽端口發送資料
nc 127.0.0.1 4444
123
轉載于:https://www.cnblogs.com/jhxxb/p/11584363.html