本次遇到的問題描述，日志采集同步時，當單條日志（日志檔案中一行日志）超過2M大小，資料無法采集同步到kafka，分析后，共踩到如下幾個坑，
1、flume采集時，通過shell+EXEC（tail -F xxx.log 的方式） source來獲取日志時，當單條日志過大超過1M時，source端無法從日志中獲取到Event，
2、日志超過1M后，flume的kafka sink 作為生產者發送給日志給kafka失敗，kafka無法收到訊息，
以下針對踩的這兩個坑做分析，flume 我使用的是1.9.0版本， kafka使用的是2.11-2.0.0版本

問題一、flume采集時，通過shell+EXEC（tail -F  xxx.log 的方式） source來獲取日志時，當單條日志過大超過1M時，source端無法從日志中獲取到Event，flume的配置如下：

 ...... agent.sources = seqGenSrc ...... # For each one of the sources, the type is definedagent.sources.seqGenSrc.type = exec#agent.sources.seqGenSrc.command = tail -F /opt/logs/test.log|grep businessCollection|awk -F '- {' '{print "{"$2}'agent.sources.seqGenSrc.command = tail -F /opt/logs/test.log|grep businessCollectionagent.sources.seqGenSrc.shell = /bin/bash -cagent.sources.seqGenSrc.batchSize = 1agent.sources.seqGenSrc.batchTimeout = 90000......

　　原因：采用shell+EXEC方式的時候，flume的原始碼中使用的是如下的方式來獲取日志

    private Process process = null;	//使用這種方式來執行命令，process = Runtime.getRuntime().exec(commandArgs);//讀取日志 reader = new BufferedReader(  new InputStreamReader(process.getInputStream(), charset));

在一行日志超過1M后，這個代碼就假死了，一直宕住，導致無法獲取到資料，

針對這個問題處理方式：
方式一:修改原始碼的實作方式，（1.9.0的原始碼 對應的是原始碼中的flume-ng-core 專案中的org.apache.flume.source.ExecSource.java 這個類）

//process的采用如下方式獲和執行命令，就改一行代碼，增加.redirectErrorStream(true)后，輸入流就都可以獲取到，哪怕超過1Mprocess = new ProcessBuilder(commandArgs).redirectErrorStream(true).start();

修改完成后，重新打包編譯，然后將生成的jar包替換原來老的jar包，

  方式二：放棄EXECSource，使用TAILDIR Source， 使用這個source時，對應的配置如下：

 ...... agent.sources = seqGenSrc ...... # For each one of the sources, the type is definedagent.sources.seqGenSrc.type = TAILDIRagent.sources.seqGenSrc.positionFile = ./taildir_position.jsonagent.sources.seqGenSrc.filegroups = seqGenSrcagent.sources.seqGenSrc.filegroups.seqGenSrc = https://www.cnblogs.com/opt/logs/test.logagent.sources.seqGenSrc.fileHeader = falseagent.sources.seqGenSrc.batchSize = 1......

　　建議采用TAILDIR Source 比較好，這個可以對多個日志進行監控和采集，而且日志采集時會記錄日志采集位置到positionFile 中，這樣日志采集不會重復，EXEC SOURCE在重啟采集時資料會重復采集，還需要其他的方式去避免重復采集

問題二、日志超過1M后，flume的kafka sink 作為生產者發送給日志給kafka失敗，kafka無法收到訊息
原因：kafka 在默認情況下，只能接收1M大小以內的訊息，在沒有做自定義設定時，所以單條訊息大于1M后是無法處理的，
處理方式如下：

1）、修改kafka 服務端server.properties檔案，做如下設定(修改大小限制)

# The send buffer (SO_SNDBUF) used by the socket serversocket.send.buffer.bytes=502400# The receive buffer (SO_RCVBUF) used by the socket serversocket.receive.buffer.bytes=502400# The maximum size of a request that the socket server will accept (protection against OOM)socket.request.max.bytes=104857600message.max.bytes=5242880replica.fetch.max.bytes=6291456

2）、修改producer.properties，做如下設定(修改大小限制)

# the maximum size of a request in bytesmax.request.size= 9242880

3）、java代碼中在初始化kafka 生產者時，也需要指定max.request.size= 9242880

 本文作者：張永清，轉載請注明出處：一次flume exec source采集日志到kafka因為單條日志資料非常大同步失敗的踩坑帶來的思考

        Properties properties = new Properties();		...		      properties.put("max.request.size", 5242880);			  ...			KafkaProducer<Object,Object>  kafkaProducer = new KafkaProducer<Object,Object>(properties);

　　4)、消費者在消費kafka資料時，也需要注意設定消費訊息的大小限制

 本文作者：張永清，轉載請注明出處：一次flume exec source采集日志到kafka因為單條日志資料非常大同步失敗的踩坑帶來的思考

            Properties properties = new Properties();			...            properties.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 6291456);						...				 KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);　　

對于flume不了的同學，可以看flume 1.9中文版用戶指南：https://www.h3399.cn/201906/700076.html　　

flume1.9 用戶指南 (中文版)

概述

Apache Flume 是一個分布式, 可靠且可用的系統, 用于有效地從許多不同的 source 收集, 聚合和移動大量日志資料到集中式資料存盤.

Apache Flume 的使用不僅限于日志資料聚合. 由于資料 source 是可定制的, 因此 Flume 可用于傳輸大量 event 資料, 包括但不限于網路流量資料, 社交媒體生成的資料, 電子郵件訊息以及幾乎任何可能的資料 source.

Apache Flume 是 Apache Software Foundation 的頂級專案.

系統要求

Java 運行時環境 - Java 1.8 或更高版本

記憶體 - 為 source,channel 或 sink 配置的記憶體

磁盤空間 - channel 或 sink 配置的磁盤空間

目錄權限 - agent 使用的目錄的讀 / 寫權限

架構

資料流模型

Flume event 被定義為具有位元組有效負載和可選字串屬性集的資料流單元. Flume agent 是一個 (JVM) 行程, 它承載 event 從外部 source 流向下一個目標 (躍點) 的組件.

?

Flume source 消耗由外部 source(如 Web 服務器)傳遞給它的 event . 外部 source 以目標 Flume source 識別的格式向 Flume 發送 event . 例如, Avro Flume source 可用于從 Avro 客戶端或從 Avrosink 發送 event 的流中的其他 Flume agent 接收 Avroevent . 可以使用 Thrift Flume Source 定義類似的流程, 以接收來自 Thrift Sink 或 Flume Thrift Rpc 客戶端或 Thrift 客戶端的 event , 這些客戶端使用 Flume thrift 協議生成的任何語言撰寫. 當 Flume source 接收 event 時, 它將其存盤到一個或多個 channels . 該 channel 是一個被動存盤器, 可以保持 event 直到它被 Flume sink 消耗. 檔案 channel 就是一個例子 - 它由本地檔案系統支持. sink 從 channel 中移除 event 并將其放入外部存盤庫 (如 HDFS(通過 Flume HDFS sink)) 或將其轉發到流中下一個 Flume agent (下一跳)的 Flume source. 給定 agent 中的 source 和 sink 與 channel 中暫存的 event 異步運行.

復雜的流程

Flume 允許用戶構建多跳流, 其中 event 在到達最終目的地之前經過多個 agent . 它還允許 fan-in 和 fan-out, 背景關系路由和故障跳躍的備份路由(故障轉移).

可靠性

event 在每個 agent 的 channel 中進行. 然后將 event 傳遞到流中的下一個 agent 或終端存盤庫(如 HDFS). 只有將 event 存盤在下一個 agent 的 channel 或終端存盤庫中后, 才會從 channel 中洗掉這些 event . 這就是 Flume 中的單跳訊息傳遞語意如何提供流的端到端可靠性.

Flume 使用事務方法來保證 event 的可靠傳遞. source 和 sink 分別在事務中封裝由 channel 提供的事務中放置或提供的 event 的存盤 / 檢索. 這可確保 event 集在流中從一個點到另一個點可靠地傳遞. 在多跳流的情況下, 來自前一跳的 sink 和來自下一跳的 source 都運行其事務以確保資料安全地存盤在下一跳的 channel 中.

可恢復性

event 在 channel 中進行, 該 channel 管理從故障中恢復. Flume 支持由本地檔案系統支持的持久檔案 channel. 還有一個記憶體 channel, 它只是將 event 存盤在記憶體中的佇列中, 這更快, 但是當 agent 行程死亡時仍然留在記憶體 channel 中的任何 event 都無法恢復.

設定

設定 agent

Flume agent 配置存盤在本地組態檔中. 這是一個遵循 Java 屬性檔案格式的文本檔案. 可以在同一組態檔中指定一個或多個 agent 的配置. 組態檔包括 agent 中每個 source,sink 和 channel 的屬性以及它們如何連接在一起以形成資料流.

配置單個組件

流中的每個組件 (source,sink 或 channel) 都具有特定于型別和實體化的名稱, 型別和屬性集. 例如, Avrosource 需要主機名 (或 IP 地址) 和埠號來接收資料. 記憶體 channel 可以具有最大佇列大小 ("容量"),HDFS sink 需要知道檔案系統 URI, 創建檔案的路徑, 檔案輪換頻率("hdfs.rollInterval") 等. 組件的所有此類屬性需要在托管 Flume agent 的屬性檔案中設定.

將各個部分連接在一起

agent 需要知道要加載哪些組件以及它們如何連接以構成流程. 這是通過列出 agent 中每個 source,sink 和 channel 的名稱, 然后為每個 sink 和 source 指定連接 channel 來完成的. 例如, agent 通過名為 file-channel 的檔案 channel 將 event 從名為 avroWeb 的 Avrosource 流向 HDFS sink hdfs-cluster1. 組態檔將包含這些組件的名稱和檔案 channel, 作為 avroWebsource 和 hdfs-cluster1 sink 的共享 channel.

啟動 agent

使用名為 flume-ng 的 shell 腳本啟動 agent 程式, 該腳本位于 Flume 發行版的 bin 目錄中. 您需要在命令列上指定 agent 名稱, config 目錄和組態檔:

$ bin/flume-ng agent -n $agent_name -c conf -f conf/flume-conf.properties.template

現在, agent 將開始運行在給定屬性檔案中配置的 source 和 sink.

一個簡單的例子

在這里, 我們給出一個示例組態檔, 描述單節點 Flume 部署. 此配置允許用戶生成 event , 然后將其記錄到控制臺.

# example.conf: A single-node Flume configuration# Name the components on this agenta1.sources = r1a1.sinks = k1a1.channels = c1# Describe/configure the sourcea1.sources.r1.type = netcata1.sources.r1.bind = localhosta1.sources.r1.port = 44444# Describe the sinka1.sinks.k1.type = logger# Use a channel which buffers events in memorya1.channels.c1.type = memorya1.channels.c1.capacity = 1000a1.channels.c1.transactionCapacity = 100# Bind the source and sink to the channela1.sources.r1.channels = c1a1.sinks.k1.channel = c1

此配置定義名為 a1 的單個 agent .a1 有一個監聽埠 44444 上的資料的 source, 一個緩沖記憶體中 event 資料的 channel, 以及一個將 event 資料記錄到控制臺的 sink. 組態檔命名各種組件, 然后描述其型別和配置引數. 給定的組態檔可能會定義幾個命名的 agent 當一個給定的 Flume 行程啟動時, 會傳遞一個標志, 告訴它要顯示哪個命名 agent.

鑒于此組態檔, 我們可以按如下方式啟動 Flume:

$ bin/flume-ng agent --conf conf --conf-file example.conf --name a1 -Dflume.root.logger=INFO,console

請注意, 在完整部署中, 我們通常會包含一個選項: --conf=<conf-dir> . 所述 <conf-dir> 目錄將包括一個 shell 腳本 f lume-env.sh 和潛在的一個 log4j 的屬性檔案. 在這個例子中, 我們傳遞一個 Java 選項來強制 Flume 登錄到控制臺, 我們沒有自定義環境腳本.

從一個單獨的終端, 我們可以 telnet 埠 44444 并向 Flume 發送一個 event :

$ telnet localhost 44444Trying 127.0.0.1...Connected to localhost.localdomain (127.0.0.1).Escape character is '^]'.Hello world! <ENTER>OK

原始的 Flume 終端將在日志訊息中輸出 event .

12/06/19 15:32:19 INFO source.NetcatSource: Source starting12/06/19 15:32:19 INFO source.NetcatSource: Created serverSocket:sun.nio.ch.ServerSocketChannelImpl[/127.0.0.1:44444]12/06/19 15:32:34 INFO sink.LoggerSink: Event: {headers:{} body: 48 65 6C 6C 6F 20 77 6F 72 6C 64 21 0D Hello world!.}

恭喜 - 您已成功配置并部署了 Flume agent ! 后續部分更詳細地介紹了 agent 配置.

在組態檔中使用環境變數

Flume 能夠替換配置中的環境變數. 例如:

a1.sources = r1a1.sources.r1.type = netcata1.sources.r1.bind = 0.0.0.0a1.sources.r1.port = ${NC_PORT}a1.sources.r1.channels = c1

注意: 它目前僅適用于 values , 不適用于 keys . (IE. only on the "right side" of the = mark of the config lines.)

通過設定 propertiesImplementation = org.apache.flume.node.EnvVarResolverProperties, 可以通過 agent 程式呼叫上的 Java 系統屬性啟用此功能.

例如:

$ NC_PORT=44444 bin/flume-ng agent -conf conf -conf-file example.conf -name a1 -Dflume.root.logger=INFO,console -DpropertiesImplementation=org.apache.flume.node.EnvVarResolverProperties

請注意, 上面只是一個示例, 可以通過其他方式配置環境變數, 包括在 conf/flume-env.sh.

記錄原始資料

在許多生產環境中記錄流經攝取 pipeline 的原始資料流不是所希望的行為, 因為這可能導致泄漏敏感資料或安全相關配置 (例如密鑰) 泄漏到 Flume 日志檔案. 默認情況下, Flume 不會記錄此類資訊. 另一方面, 如果資料管道被破壞, Flume 將嘗試提供除錯 DEBUG 的線索.

除錯 event 管道問題的一種方法是設定 連接到 Logger Sink 的附加記憶體 channel, 它將所有 event 資料輸出到 Flume 日志. 但是, 在某些情況下, 這種方法是不夠的.

為了能夠記錄 event 和配置相關的資料, 除了 log4j 屬性外, 還必須設定一些 Java 系統屬性.

要啟用與配置相關的日志記錄, 請設定 Java 系統屬性 - Dorg.apache.flume.log.printconfig=true . 這可以在命令列上傳遞, 也可以在 flume-env.sh 中的 JAVA_OPTS 變數中設定.

要啟用資料記錄, 請 按照上述相同方式設定 Java 系統屬性 -Dorg.apache.flume.log.rawdata=https://www.cnblogs.com/laoqing/p/true . 對于大多陣列件, 還必須將 log4j 日志記錄級別設定為 DEBUG 或 TRACE, 以使特定于 event 的日志記錄顯示在 Flume 日志中.

下面是啟用配置日志記錄和原始資料日志記錄的示例, 同時還將 Log4j 日志級別設定為 DEBUG 以用于控制臺輸出:

$ bin/flume-ng agent --conf conf --conf-file example.conf --name a1 -Dflume.root.logger=DEBUG,console -Dorg.apache.flume.log.printconfig=true -Dorgwdata=https://www.cnblogs.com/laoqing/p/true

基于 Zookeeper 的配置

Flume 通過 Zookeeper 支持 agent 配置. 這是一個實驗性功能. 組態檔需要在可配置前綴下的 Zookeeper 中上傳. 組態檔存盤在 Zookeeper 節點資料中. 以下是 agent 商 a1 和 a2 的 Zookeeper 節點樹的外觀

- /flume|- /a1 [Agent config file]|- /a2 [Agent config file]

上載組態檔后, 使用以下選項啟動 agent

$ bin/flume-ng agent -conf conf -z zkhost:2181,zkhost1:2181 -p /flume -name a1 -Dflume.root.logger=INFO,console

Flume 擁有完全基于插件的架構. 雖然 Flume 附帶了許多開箱即用的 source,channels,sink,serializers 等, 但許多實作都與 Flume 分開運行. 安裝第三方插件

雖然通過將自己的 jar 包添加到 flume-env.sh 檔案中的 FLUME_CLASSPATH 變數中, 始終可以包含自定義 Flume 組件, 但 Flume 現在支持一個名為 plugins.d 的特殊目錄, 該目錄會自動獲取以特定格式打包的插件. 這樣可以更輕松地管理插件打包問題, 以及更簡單的除錯和幾類問題的故障排除, 尤其是庫依賴性沖突.

目錄

該 plugins.d 目錄位于 $FLUME_HOME/plugins.d . 在啟動時, flume-ng 啟動腳本在 plugins.d 目錄中查找符合以下格式的插件, 并在啟動 java 時將它們包含在正確的路徑中.

插件的目錄布局

plugins.d 中的每個插件 (子目錄) 最多可以有三個子目錄:

lib - the plugin's jar(s)libext - the plugin's dependency jar(s)native - any required native libraries, such as .so files

plugins.d 目錄中的兩個插件示例:

plugins.d/plugins.d/custom-source-1/plugins.d/custom-source-1/lib/my-source.jarplugins.d/custom-source-1/libext/spring-core-2.5.6.jarplugins.d/custom-source-2/plugins.d/custom-source-2/lib/custom.jarplugins.d/custom-source-2/native/gettext.so

資料攝取

Flume 支持許多從外部來 source 攝取資料的機制.

RPC

Flume 發行版中包含的 Avro 客戶端可以使用 avro RPC 機制將給定檔案發送到 Flume Avrosource:

$ bin/flume-ng avro-client -H localhost -p 41414 -F /usr/logs/log.10

上面的命令會將 /usr/logs/log.10 的內容發送到監聽該埠的 Flume source.

執行命令

有一個 exec source 執行給定的命令并消耗輸出. 輸出的單 "行" 即. 文本后跟回車符 ('\ r') 或換行符 ('\ n') 或兩者一起.

網路流

Flume 支持以下機制從常用日志流型別中讀取資料, 例如:

1. Avro
2. Thrift
3. Syslog
4. Netcat

設定多 agent 流程

?

為了跨多個 agent 或跳資料流, 先前 agent 的 sink 和當前跳的 source 需要是 avro 型別, sink 指向 source 的主機名 (或 IP 地址) 和埠.

合并

日志收集中非常常見的情況是大量日志生成客戶端將資料發送到連接到存盤子系統的少數消費者 agent . 例如, 從數百個 Web 服務器收集的日志發送給寫入 HDFS 集群的十幾個 agent .

?

這可以通過使用 avrosink 配置多個第一層 agent 在 Flume 中實作, 所有這些 agent 都指向單個 agent 的 avrosource(同樣, 您可以在這種情況下使用 thriftsource/sink / 客戶端). 第二層 agent 上的此 source 將接收的 event 合并到單個信道中, 該信道由信宿器消耗到其最終目的地.

多路復用流程

Flume 支持將 event 流多路復用到一個或多個目的地. 這是通過定義可以復制或選擇性地將 event 路由到一個或多個信道的流復用器來實作的.

?

上面的例子顯示了來自 agent "foo" 的 source 代碼將流程擴展到三個不同的 channel. 扇出可以復制或多路復用. 在復制流的情況下, 每個 event 被發送到所有三個 channel. 對于多路復用情況, 當 event 的屬性與預配置的值匹配時, event 將被傳遞到可用 channel 的子集. 例如, 如果一個名為 "txnType" 的 event 屬性設定為 "customer", 那么它應該轉到 channel1 和 channel3, 如果它是 "vendor", 那么它應該轉到 channel2, 否則轉到 channel3. 可以在 agent 的組態檔中設定映射.

配置

如前面部分所述, Flume agent 程式配置是從類似于具有分層屬性設定的 Java 屬性檔案格式的檔案中讀取的.

定義流程

要在單個 agent 中定義流, 您需要通過 channel 鏈接 source 和 sink. 您需要列出給定 agent 的 source,sink 和 channel, 然后將 source 和 sink 指向 channel.source 實體可以指定多個 channel, 但 sink 實體只能指定一個 channel. 格式如下:

# list the sources, sinks and channels for the agent<Agent>.sources = <Source><Agent>.sinks = <Sink><Agent>.channels = <Channel1> <Channel2># set channel for source<Agent>.sources.<Source>.channels = <Channel1> <Channel2> ...# set channel for sink<Agent>.sinks.<Sink>.channel = <Channel1>

例如, 名為 agent_foo 的 agent 正在從外部 avro 客戶端讀取資料并通過記憶體 channel 將其發送到 HDFS.

組態檔 weblog.config 可能如下所示:

# list the sources, sinks and channels for the agentagent_foo.sources = avro-appserver-src-1agent_foo.sinks = hdfs-sink-1agent_foo.channels = mem-channel-1# set channel for sourceagent_foo.sources.avro-appserver-src-1.channels = mem-channel-1# set channel for sinkagent_foo.sinks.hdfs-sink-1.channel = mem-channel-1

這將使 event 從 avro-AppSrv-source 流向 hdfs-Cluster1-sink, 通過記憶體 channelmem-channel-1.

當使用 weblog.config 作為其組態檔啟動 agent 程式時, 它將實體化該流程.

配置單個組件

定義流后, 您需要設定每個 source,sink 和 channel 的屬性. 這是以相同的分層命名空間方式完成的, 您可以在其中設定組件型別以及特定于每個組件的屬性的其他值:

# properties for sources<Agent>.sources.<Source>.<someProperty> = <someValue># properties for channels<Agent>.channel.<Channel>.<someProperty> = <someValue># properties for sinks<Agent>.sources.<Sink>.<someProperty> = <someValue>

需要為 Flume 的每個組件設定屬性 "type", 以了解它需要什么型別的物件. 每個 source,sink 和 channel 型別都有自己的一組屬性, 使其能夠按預期運行. 所有這些都需要根據需要進行設定. 在前面的示例中, 我們有一個從 avro-AppSrv-source 到 hdfs-Cluster1-sink 的流程通過記憶體 channelmem-channel-1. 這是一個顯示每個組件配置的示例:

agent_foo.sources = avro-AppSrv-sourceagent_foo.sinks = hdfs-Cluster1-sinkagent_foo.channels = mem-channel-1# set channel for sources, sinks# properties of avro-AppSrv-sourceagent_foo.sources.avro-AppSrv-source.type = avroagent_foo.sources.avro-AppSrv-source.bind = localhostagent_foo.sources.avro-AppSrv-source.port = 10000# properties of mem-channel-1agent_foo.channels.mem-channel-1.type = memoryagent_foo.channels.mem-channel-1.capacity = 1000agent_foo.channels.mem-channel-1.transactionCapacity = 100# properties of hdfs-Cluster1-sinkagent_foo.sinks.hdfs-Cluster1-sink.type = hdfsagent_foo.sinks.hdfs-Cluster1-sink.hdfs.path = hdfs://namenode/flume/webdata#...

在 agent 中添加多個流

單個 Flume agent 可以包含多個獨立流. 您可以在配置中列出多個 source,sink 和 channel. 可以鏈接這些組件以形成多個流:

# list the sources, sinks and channels for the agent<Agent>.sources = <Source1> <Source2><Agent>.sinks = <Sink1> <Sink2><Agent>.channels = <Channel1> <Channel2>

然后, 您可以將 source 和 sink 鏈接到 channel(用于 sink)的相應 channel(用于 source), 以設定兩個不同的流. 例如, 如果您需要在 agent 中設定兩個流, 一個從外部 avro 客戶端到外部 HDFS, 另一個從尾部輸出到 avrosink, 那么這是一個配置來執行此操作:

# list the sources, sinks and channels in the agentagent_foo.sources = avro-AppSrv-source1 exec-tail-source2agent_foo.sinks = hdfs-Cluster1-sink1 avro-forward-sink2agent_foo.channels = mem-channel-1 file-channel-2# flow #1 configurationagent_foo.sources.avro-AppSrv-source1.channels = mem-channel-1agent_foo.sinks.hdfs-Cluster1-sink1.channel = mem-channel-1# flow #2 configurationagent_foo.sources.exec-tail-source2.channels = file-channel-2agent_foo.sinks.avro-forward-sink2.channel = file-channel-2

配置多 agent 流程

要設定多層流, 您需要有 sink 指向下一跳的 avro/thrift source. 這將導致第一個 Flume agent 將 event 轉發到下一個 Flume agent . 例如, 如果您使用 avro 客戶端定期向本地 Flume agent 發送檔案(每個 event 1 個檔案), 則此本地 agent 可以將其轉發到另一個已安裝存盤的 agent

Weblog agent 配置:

# list sources, sinks and channels in the agentagent_foo.sources = avro-AppSrv-sourceagent_foo.sinks = avro-forward-sinkagent_foo.channels = file-channel# define the flowagent_foo.sources.avro-AppSrv-source.channels = file-channelagent_foo.sinks.avro-forward-sink.channel = file-channel# avro sink propertiesagent_foo.sinks.avro-forward-sink.type = avroagent_foo.sinks.avro-forward-sink.hostname = 10.1.1.100agent_foo.sinks.avro-forward-sink.port = 10000# configure other pieces#...

HDFS agent 配置:

# list sources, sinks and channels in the agentagent_foo.sources = avro-collection-sourceagent_foo.sinks = hdfs-sinkagent_foo.channels = mem-channel# define the flowagent_foo.sources.avro-collection-source.channels = mem-channelagent_foo.sinks.hdfs-sink.channel = mem-channel# avro source propertiesagent_foo.sources.avro-collection-source.type = avroagent_foo.sources.avro-collection-source.bind = 10.1.1.100agent_foo.sources.avro-collection-source.port = 10000# configure other pieces#...

在這里, 我們將 weblog agent 的 avro-forward-sink 鏈接到 hdfs agent 的 avro-collection-source. 這將導致來自外部應用程式服務器 source 的 event 最終存盤在 HDFS 中.

扇出流量

如前一節所述, Flume 支持扇出從一個 source 到多個 channel 的流量. 扇出有兩種模式 : 復制和多路復用. 在復制流程中, event 將發送到所有已配置的 channel. 在多路復用的情況下, event 僅被發送到合格 channels 的子集. 為了散開流量, 需要指定 source 的 channel 串列以及扇出它的策略. 這是通過添加可以復制或多路復用的 channel"選擇器" 來完成的. 如果它是多路復用器, 則進一步指定選擇規則. 如果您沒有指定選擇器, 那么默認情況下它會復制:

# List the sources, sinks and channels for the agent<Agent>.sources = <Source1><Agent>.sinks = <Sink1> <Sink2><Agent>.channels = <Channel1> <Channel2># set list of channels for source (separated by space)<Agent>.sources.<Source1>.channels = <Channel1> <Channel2># set channel for sinks<Agent>.sinks.<Sink1>.channel = <Channel1><Agent>.sinks.<Sink2>.channel = <Channel2><Agent>.sources.<Source1>.selector.type = replicating

多路復用選擇具有另一組屬性以分流流. 這需要指定 event 屬性到 channel 集的映射. 選擇器檢查 event 頭中的每個已配置屬性. 如果它與指定的值匹配, 則該 event 將發送到映射到該值的所有 channel. 如果沒有匹配項, 則將 event 發送到配置為默認值的 channel 集:

# Mapping for multiplexing selector<Agent>.sources.<Source1>.selector.type = multiplexing<Agent>.sources.<Source1>.selector.header = <someHeader><Agent>.sources.<Source1>.selector.mapping.<Value1> = <Channel1><Agent>.sources.<Source1>.selector.mapping.<Value2> = <Channel1> <Channel2><Agent>.sources.<Source1>.selector.mapping.<Value3> = <Channel2>#...<Agent>.sources.<Source1>.selector.default = <Channel2>

映射允許為每個值重疊 channel.

以下示例具有多路復用到兩個路徑的單個流. 名為 agent_foo 的 agent 具有單個 avrosource 和兩個鏈接到兩個 sink 的 channel:

# list the sources, sinks and channels in the agentagent_foo.sources = avro-AppSrv-source1agent_foo.sinks = hdfs-Cluster1-sink1 avro-forward-sink2agent_foo.channels = mem-channel-1 file-channel-2# set channels for sourceagent_foo.sources.avro-AppSrv-source1.channels = mem-channel-1 file-channel-2# set channel for sinksagent_foo.sinks.hdfs-Cluster1-sink1.channel = mem-channel-1agent_foo.sinks.avro-forward-sink2.channel = file-channel-2# channel selector configurationagent_foo.sources.avro-AppSrv-source1.selector.type = multiplexingagent_foo.sources.avro-AppSrv-source1.selector.header = Stateagent_foo.sources.avro-AppSrv-source1.selector.mapping.CA = mem-channel-1agent_foo.sources.avro-AppSrv-source1.selector.mapping.AZ = file-channel-2agent_foo.sources.avro-AppSrv-source1.selector.mapping.NY = mem-channel-1 file-channel-2agent_foo.sources.avro-AppSrv-source1.selector.default = mem-channel-1

選擇器檢查名為 "State" 的標頭. 如果該值為 "CA", 則將其發送到 mem-channel-1, 如果其為 "AZ", 則將其發送到檔案 channel-2, 或者如果其為 "NY" 則為兩者. 如果 "狀態" 標題未設定或與三者中的任何一個都不匹配, 則它將轉到 mem-channel-1, 其被指定為 "default".

選擇器還支持可選 channel. 要為標頭指定可選 channel, 可通過以下方式使用 config 引數 "optional":

# channel selector configurationagent_foo.sources.avro-AppSrv-source1.selector.type = multiplexingagent_foo.sources.avro-AppSrv-source1.selector.header = Stateagent_foo.sources.avro-AppSrv-source1.selector.mapping.CA = mem-channel-1agent_foo.sources.avro-AppSrv-source1.selector.mapping.AZ = file-channel-2agent_foo.sources.avro-AppSrv-source1.selector.mapping.NY = mem-channel-1 file-channel-2agent_foo.sources.avro-AppSrv-source1.selector.optional.CA = mem-channel-1 file-channel-2agent_foo.sources.avro-AppSrv-source1.selector.mapping.AZ = file-channel-2agent_foo.sources.avro-AppSrv-source1.selector.default = mem-channel-1

選擇器將首先嘗試寫入所需的 channel, 如果其中一個 channel 無法使用 event , 則會使事務失敗. 在所有渠道上重新嘗試交易. 一旦所有必需的 channel 消耗了 event , 則選擇器將嘗試寫入可選 channel. 任何可選 channel 使用該 event 的失敗都會被忽略而不會重試.

如果可選信道與特定報頭的所需信道之間存在重疊, 則認為該信道是必需的, 并且信道中的故障將導致重試所有必需信道集. 例如, 在上面的示例中, 對于標題 "CA",mem-channel-1 被認為是必需的 channel, 即使它被標記為必需和可選, 并且寫入此 channel 的失敗將導致該 event 在為選擇器配置的所有 channel 上重試.

請注意, 如果標頭沒有任何所需的 channel, 則該 event 將被寫入默認 channel, 并將嘗試寫入該標頭的可選 channel. 如果未指定所需的 channel, 則指定可選 channel 仍會將 event 寫入默認 channel. 如果沒有將 channel 指定為默認 channel 且沒有必需 channel, 則選擇器將嘗試將 event 寫入可選 channel. 在這種情況下, 任何失敗都會被忽略.

支持

多個 Flume 組件支持 SSL / TLS 協議, 以便安全地與其他系統通信.

SSL 兼容組件具有若干配置引數來設定 SSL, 例如啟用 SSL 標志, 密鑰庫 / 信任庫引數 (位置, 密碼, 型別) 和其他 SSL 引數(例如禁用的協議)

始終在 agent 組態檔的組件級別指定為組件啟用 SSL. 因此, 某些組件可能配置為使用 SSL, 而其他組件則不配置(即使具有相同的組件型別)

密鑰庫 / 信任庫設定可以在組件級別或全域指定.

在組件級別設定的情況下, 通過組件特定引數在 agent 組態檔中配置密鑰庫 / 信任庫. 此方法的優點是組件可以使用不同的密鑰庫(如果需要). 缺點是必須為 agent 組態檔中的每個組件復制密鑰庫引數. 組件級別設定是可選的, 但如果已定義, 則其優先級高于全域引數.

使用全域設定, 只需定義一次密鑰庫 / 信任庫引數, 并對所有組件使用相同的設定, 這意味著更少和更集中的配置.

可以通過系統屬性或通過環境變數來配置全域設定.

可以在命令列上傳遞 SSL 系統屬性, 也可以在 conf / flume-env.sh 中設定 JAVA_OPTS 環境變數(盡管使用命令列是不可取的, 因為包含密碼的命令將保存到命令歷史記錄中.)

export JAVA_OPTS="$JAVA_OPTS -Djavax.net.ssl.keyStore=/path/to/keystore.jks"export JAVA_OPTS="$JAVA_OPTS -Djavax.net.ssl.keyStorePassword=password"

Flume 使用 JSSE(Java 安全套接字擴展)中定義的系統屬性, 因此這是設定 SSL 的標準方法. 另一方面, 在系統屬性中指定密碼意味著可以在行程串列中看到密碼. 對于不可接受的情況, 也可以在環境變數中定義引數. 在這種情況下, Flume 在內部從相應的環境變數初始化 JSSE 系統屬性.

SSL 環境變數可以在啟動 Flume 之前在 shell 環境中設定, 也可以在 conf / flume-env.sh 中設定(盡管使用命令列是不可取的, 因為包含密碼的命令將保存到命令歷史記錄中.)

export FLUME_SSL_KEYSTORE_PATH=/path/to/keystore.jksexport FLUME_SSL_KEYSTORE_PASSWORD=password

** 請注意:**

必須在組件級別啟用 SSL. 僅指定全域 SSL 引數不會產生任何影響.

如果在多個級別指定全域 SSL 引數, 則優先級如下(從高到低):

agent 配置中的組件引數

系統屬性

環境變數

如果為組件啟用了 SSL, 但未以上述任何方式指定 SSL 引數, 則

在密鑰庫的情況下: 配置錯誤

在 truststores 的情況下: 將使用默認信任庫(Oracle JDK 中的 jssecacerts / cacerts)

在所有情況下, 可信任密碼都是可選的. 如果未指定, 則在 JDK 打開信任庫時, 不會對信任庫執行完整性檢查.

source 和接收批量大小和 channel 事務容量

source 和 sink 可以具有批量大小引數, 該引數確定它們在一個批次中處理的最大 event 數. 這發生在具有稱為事務容量的上限的 channel 事務中. 批量大小必須小于渠道的交易容量. 有一個明確的檢查, 以防止不兼容的設定. 只要讀取配置, 就會進行此檢查.

1. Flume Source
2. Avro Source

監聽 Avro 埠并從外部 Avro 客戶端流接收 event . 當與另一個(上一跳)Flume agent 上的內置 Avro Sink 配對時, 它可以創建分層集合拓撲. 必需屬性以粗體顯示

agent 名為 a1 的示例:

a1.sources = r1a1.channels = c1a1.sources.r1.type = avroa1.sources.r1.channels = c1a1.sources.r1.bind = 0.0.0.0a1.sources.r1.port = 4141

ipFilterRules 的示例

ipFilterRules 定義由逗號分隔的 N 個 netty ipFilters 模式規則必須采用此格式.

<'allow' or deny>:<'ip' or 'name' for computer name>:<pattern> or allow/deny:ip/name:patternexample: ipFilterRules=allow:ip:127.*,allow:name:localhost,deny:ip:*

請注意, 匹配的第一個規則將適用, 如下例所示, 來自 localhost 上的客戶端

這將允許 localhost 上的客戶端拒絕來自任何其他 ip 的客戶端 "allow:name:localhost,deny:ip: 這將拒絕 localhost 上的客戶端允許來自任何其他 ip 的客戶端"deny:name:localhost,allow:ip:

Thrift Source

偵聽 Thrift 埠并從外部 Thrift 客戶端流接收 event . 當與另一個(上一跳)Flume agent 上的內置 ThriftSink 配對時, 它可以創建分層集合拓撲. 可以通過啟用 kerberos 身份驗證將 Thriftsource 配置為以安全模式啟動. agent-principal 和 agent-keytab 是 Thriftsource 用于向 kerberos KDC 進行身份驗證的屬性. 必需屬性以粗體顯示

agent 名為 a1 的示例:

a1.sources = r1a1.channels = c1a1.sources.r1.type = thrifta1.sources.r1.channels = c1a1.sources.r1.bind = 0.0.0.0a1.sources.r1.port = 4141Exec Source

Exec source 在啟動時運行給定的 Unix 命令, 并期望該行程在標準輸出上連續生成資料 (stderr 被簡單地丟棄, 除非屬性 logStdErr 設定為 true). 如果行程因任何原因退出, 則 source 也會退出并且不會生成其他資料. 這意味著 cat [named pipe] 或 tail -F [file] 等配置將產生所需的結果, 而日期 可能不會 - 前兩個命令產生資料流, 而后者產生單個 event 并退出. 必需屬性以粗體顯示

警告

Exec Source 和其他異步 source 的問題在于, 如果無法將 event 放入 Channel 中, 則 source 無法保證客戶端知道它. 在這種情況下, 資料將丟失. 例如, 最常請求的功能之一是 tail -F [file] 類似用例, 其中應用程式寫入磁盤上的日志檔案, Flume 將檔案作為尾部發送, 將每一行作為 event 發送. 雖然這是可能的, 但是有一個明顯的問題; 如果 channel 填滿并且 Flume 無法發送 event , 會發生什么?由于某種原因, Flume 無法向撰寫日志檔案的應用程式指示它需要保留日志或 event 尚未發送. 如果這沒有意義, 您只需要知道: 當使用 Exec Source 等單向異步介面時, 您的應用程式永遠無法保證已收到資料!

作為此警告的延伸 - 并且完全清楚 - 使用此 source 時, event 傳遞絕對沒有保證.

為了獲得更強的可靠性保證, 請考慮 Spooling Directory Source,Taildir Source 或通過 SDK 直接與 Flume 集成.

agent 名為 a1 的示例:

a1.sources = r1a1.channels = c1a1.sources.r1.type = execa1.sources.r1.command = tail -F /var/log/securea1.sources.r1.channels = c1

'shell'配置用于通過命令 shell(例如 Bash 或 Powershell)呼叫'命令'.'command'作為引數傳遞給'shell'執行. 這允許'命令'使用 shell 中的功能, 例如通配符, 后退標記, 管道, 回圈, 條件等. 如果沒有'shell'配置, 將直接呼叫'command'.'shell'的常用值: '/bin/sh -c', '/bin/ksh -c', 'cmd /c', 'powershell -Command', etc.

a1.sources.tailsource-1.type = execa1.sources.tailsource-1.shell = /bin/bash -ca1.sources.tailsource-1.command = for i in /path/*.txt; do cat $i; doneJMS Source

JMS Source 從 JMS 目標 (例如佇列或主題) 讀取訊息. 作為 JMS 應用程式, 它應該與任何 JMS 提供程式一起使用, 但僅使用 ActiveMQ 進行測驗. JMSsource 提供可配置的批量大小, 訊息選擇器, 用戶 / 傳遞和訊息到水槽 event 轉換器. 請注意, 供應商提供的 JMS jar 應該包含在 Flume 類路徑中, 使用 plugins.d 目錄(首選), 命令列上的 - classpath 或 flume-env.sh 中的 FLUME_CLASSPATH 變數. 必需屬性以粗體顯示

訊息轉換器

JMSsource 允許可插拔轉換器, 盡管默認轉換器可能適用于大多數用途. 默認轉換器能夠將位元組, 文本和物件訊息轉換為 FlumeEvents. 在所有情況下, 訊息中的屬性都將作為標題添加到 FlumeEvent 中.

BytesMessage:

訊息的位元組被復制到 FlumeEvent 的主體. 每封郵件無法轉換超過 2GB 的資料.

TextMessage 的:

訊息文本轉換為位元組陣列并復制到 FlumeEvent 的主體. 默認轉換器默認使用 UTF-8, 但這是可配置的.

ObjectMessage:

Object 被寫入包含在 ObjectOutputStream 中的 ByteArrayOutputStream, 并將生成的陣列復制到 FlumeEvent 的主體.

agent 名為 a1 的示例:

a1.sources = r1a1.channels = c1a1.sources.r1.type = jmsa1.sources.r1.channels = c1a1.sources.r1.initialContextFactory = org.apache.activemq.jndi.ActiveMQInitialContextFactorya1.sources.r1.connectionFactory = GenericConnectionFactorya1.sources.r1.providerURL = tcp://mqserver:61616a1.sources.r1.destinationName = BUSINESS_DATAa1.sources.r1.destinationType = QUEUE

SSL 和 JMS Source

JMS 客戶端實作通常支持通過 JSSE(Java 安全套接字擴展)定義的某些 Java 系統屬性來配置 SSL / TLS. 為 Flume 的 JVM 指定這些系統屬性, JMSsource(或更準確地說是 JMSsource 使用的 JMS 客戶端實作)可以通過 SSL 連接到 JMS 服務器(當然, 僅當 JMS 服務器也已設定為使用 SSL 時). 它應該可以與任何 JMS 提供程式一起使用, 并且已經過 ActiveMQ,IBM MQ 和 Oracle WebLogic 的測驗.

以下部分僅介紹 Flume 方面所需的 SSL 配置步驟. 您可以在 Flume Wiki 上找到有關不同 JMS 提供程式的服務器端設定以及完整作業配置示例的更詳細說明.

** SSL 傳輸 / 服務器身份驗證:**

如果 JMS 服務器使用自簽名證書或其證書由不受信任的 CA(例如公司自己的 CA)簽名, 則需要設定信任庫 (包含正確的證書) 并傳遞給 Flume. 它可以通過全域 SSL 引數完成. 有關全域 SSL 設定的更多詳細資訊, 請參閱 SSL / TLS 支持部分.

使用 SSL 時, 某些 JMS 提供程式需要 SSL 特定的 JNDI 初始背景關系工廠和 / 或提供程式 URL 設定(例如, ActiveMQ 使用 ssl:// URL 前綴而不是 tcp://). 在這種情況下, 必須在 agent 組態檔中調整 source 屬性(initialContextFactory 和 / 或 providerURL)

客戶端證書身份驗證(雙向 SSL):

JMS Source 可以通過客戶端證書身份驗證而不是通常的用戶 / 密碼登錄來對 JMS 服務器進行身份驗證(使用 SSL 并且 JMS 服務器配置為接受此類身份驗證時).

包含用于身份驗證的 Flume 密鑰的密鑰庫需要再次通過全域 SSL 引數進行配置. 有關全域 SSL 設定的更多詳細資訊, 請參閱 SSL / TLS 支持部分.

密鑰庫應該只包含一個密鑰(如果存在多個密鑰, 則將使用第一個密鑰). 密鑰密碼必須與密鑰庫密碼相同.

在客戶端證書身份驗證的情況下, 不需要在 Flume agent 組態檔中為 JMSsource 指定 userName / passwordFile 屬性.

請注意:

與其他組件不同, JMS Source 沒有組件級別的配置引數. 也沒有啟用 SSL 標志. SSL 設定由 JNDI / Provider URL 設定 (最終是 JMS 服務器設定) 以及 truststore / keystore 的存在 / 不存在控制.

Spooling Directory Source

此 source 允許您通過將要攝取的檔案放入磁盤上的 "spooling" 目錄來攝取資料. 此 source 將查看新檔案的指定目錄, 并將在新檔案出現時決議 event .event 決議邏輯是可插入的. 在給定檔案完全讀入 channel 后, 默認情況下通過重命名檔案來指示完成, 或者可以洗掉它或使用 trackerDir 來跟蹤已處理的檔案.

與 Exec source 不同, 即使 Flume 重新啟動或被殺死, 此 source 也是可靠的并且不會遺漏資料. 作為這種可靠性的交換, 只有不可變的, 唯一命名的檔案必須被放入 spooling directory. 中. Flume 試圖檢測這些問題, 如果違反則會宣告失敗:

如果在放入 spooling directory 后寫入檔案, Flume 會將錯誤列印到其日志檔案并停止處理.

如果稍后重復使用檔案名, Flume 將在其日志檔案中輸出錯誤并停止處理.

為避免上述問題, 在將檔案名移動到 spooling directory 中時, 添加唯一識別符號 (例如時間戳) 可能很有用.

盡管該 source 的可靠性保證, 但仍存在如果發生某些下游故障則可能重復 event 的情況. 這與其他 Flume 組件提供的保證一致.

名為 agent-1 的 agent 示例:

a1.channels = ch-1a1.sources = src-1a1.sources.src-1.type = spooldira1.sources.src-1.channels = ch-1a1.sources.src-1.spoolDir = /var/log/apache/flumeSpoola1.sources.src-1.fileHeader = trueEvent Deserializers

以下 event 反序列化器隨 Flume 一起提供.

LINE

此解串器為每行文本輸入生成一個 event

AVRO

此解串器能夠讀取 Avro 容器檔案, 并在檔案中為每個 Avro 記錄生成一個 event . 每個 event 都使用標頭進行注釋, 該標頭指示所使用的模式. event 的主體是二進制 Avro 記錄資料, 不包括模式或容器檔案元素的其余部分.

請注意, 如果假脫機目錄 source 必須重試將其中一個 event 放入 channel(例如, 因為 channel 已滿), 則它將重置并從最新的 Avro 容器檔案同步點重試. 為了減少此類故障情況下的潛在 event 重復, 請在 Avro 輸入檔案中更頻繁地寫入同步標記.

BlobDeserializer>

此反序列化器為每個 event 讀取二進制大物件(BLOB), 通常每個檔案一個 BLOB. 例如 PDF 或 JPG 檔案. 請注意, 此方法不適用于非常大的物件, 因為整個 BLOB 都快取在 RAM 中.

Taildir Source

注意: 此 source 作為預覽功能提供. 它不適用于 Windows.

觀察指定的檔案, 并在檢測到添加到每個檔案的新行后幾乎實時地拖尾它們. 如果正在寫入新行, 則此 source 將重試讀取它們以等待寫入完成.

此 source 是可靠的, 即使 tail 檔案旋轉也不會丟失資料. 它定期以 JSON 格式寫入給定位置檔案上每個檔案的最后讀取位置. 如果 Flume 由于某種原因停止或停止, 它可以從寫在現有位置檔案上的位置重新開始 tail.

在其他用例中, 此 source 也可以使用給定的位置檔案從每個檔案的任意位置開始拖尾. 當指定路徑上沒有位置檔案時, 默認情況下它將從每個檔案的第一行開始拖尾.

檔案將按修改時間順序使用. 將首先使用具有最早修改時間的檔案.

此 source 不會重命名或洗掉或對正在掛載的檔案執行任何修改. 目前此 source 不支持 tail 二進制檔案. 它逐行讀取文本檔案.

agent 名為 a1 的示例:

a1.sources = r1a1.channels = c1a1.sources.r1.type = TAILDIRa1.sources.r1.channels = c1a1.sources.r1.positionFile = /var/log/flume/taildir_position.JSONa1.sources.r1.filegroups = f1 f2a1.sources.r1.filegroups.f1 = /var/log/test1/example.loga1.sources.r1.headers.f1.headerKey1 = value1a1.sources.r1.filegroups.f2 = /var/log/test2/.*log.*a1.sources.r1.headers.f2.headerKey1 = value2a1.sources.r1.headers.f2.headerKey2 = value2-2a1.sources.r1.fileHeader = truea1.sources.ri.maxBatchCount = 1000Kafka Source

Kafka Source 是一個 Apache Kafka 消費者, 它從 Kafka 主題中讀取訊息. 如果您運行了多個 Kafka source, 則可以使用相同的使用者組配置它們, 以便每個 source 都讀取一組唯一的主題磁區. 這目前支持 Kafka 服務器版本 0.10.1.0 或更高版本. 測驗完成了 2.0.1, 這是發布時最高的可用版本.

** 注意 **

Kafka Source 會覆寫兩個 Kafka 使用者引數: source 和每次 batch 提交的時候會將 auto.commit.enable 設定為 "false", 并提交每個批處理. Kafka Source 至少保證一次訊息檢索策略.

source 啟動時可以存在重復項.

Kafka Source 還提供了 key.deserializer(org.apache.kafka.common.serialization.StringSerializer)和 value.deserializer(org.apache.kafka.common.serialization.ByteArraySerializer)的默認值.

不建議修改這些引數.

不推薦使用的屬性

通過逗號分隔的主題串列進行主題訂閱的示例:

tier1.sources.source1.type = org.apache.flume.source.kafka.KafkaSourcetier1.sources.source1.channels = channel1tier1.sources.source1.batchSize = 5000tier1.sources.source1.batchDurationMillis = 2000tier1.sources.source1.kafka.Bootstrap.servers = localhost:9092tier1.sources.source1.kafka.topics = test1, test2tier1.sources.source1.kafka.consumer.group.id = custom.g.id

正則運算式主題訂閱的示例:

ier1.source.source1.type = org.apache.flume.source.kafka.KafkaSourcetier1.source.source1.channel = channel1

tier1.source.source1.kafka.Bootstrap.servers = 本地主機: 9092

tier1.source.source1.kafka .topics.regex = ^ topic [0-9] $

＃默認使用 kafka.consumer.group.id = flume

** Security 和 Kafka Source:**

Flume 和 Kafka 之間的通信渠道支持安全認證和資料加密. 對于安全身份驗證, 可以使用 Kafka 0.9.0 版中的 SASL / GSSAPI(Kerberos V5)或 SSL(即使該引數名為 SSL, 實際協議是 TLS 實作)

截至目前, 資料加密僅由 SSL / TLS 提供.

將 kafka.consumer.security.protocol 設定為以下任何值意味著:

SASL_PLAINTEXT - 無資料加密的 Kerberos 或純文本身份驗證

SASL_SSL - 使用資料加密的 Kerberos 或純文本身份驗證

SSL - 基于 TLS 的加密, 帶有可選的身份驗證

** 警告 **

啟用 SSL 時性能會下降, 其大小取決于 CPU 型別和 JVM 實作. 參考: Kafka 安全概述 和跟蹤此問題的 jira: https://issues/jira/browse/KAFKA-2561

** TLS 和 Kafka Source:**

請閱讀配置 Kafka 客戶端 SSL 中描述的步驟, 以了解用于微調的其他配置設定, 例如以下任何一項: 安全提供程式, 密碼套件, 啟用的協議, 信任庫或密鑰庫型別.

配置服務器端身份驗證和資料加密的示例:

a1.sources.source1.type = org.apache.flume.source.kafka.KafkaSourcea1.sources.source1.kafka.Bootstrap.servers = kafka-1:9093,kafka-2:9093,kafka-3:9093a1.sources.source1.kafka.topics = mytopica1.sources.source1.kafka.consumer.group.id = flume-consumera1.sources.source1.kafka.consumer.security.protocol = SSL# optional, the global truststore can be used alternativelya1.sources.source1.kafka.consumer.ssl.truststore.location=/path/to/truststore.jksa1.sources.source1.kafka.consumer.ssl.truststore.password=<password to access the truststore>

在此處指定信任庫是可選的, 可以使用全域信任庫. 有關全域 SSL 設定的更多詳細資訊, 請參閱 SSL / TLS 支持部分

注意: 默認情況下, 未定義屬性 ssl.endpoint.identification.algorithm, 因此不會執行主機名驗證. 要啟用主機名驗證, 請設定以下屬性

a1.sources.source1.kafka.consumer.ssl.endpoint.identification.algorithm=HTTPS

啟用后, 客戶端將根據以下兩個欄位之一驗證服務器的完全限定域名(FQDN):

通用名稱(CN) https://tools.ietf.org/html/rfc6125#section-2.3 https://tools.ietf.org/html/rfc6125#section-2.3

主題備選名稱(SAN)

如果還需要客戶端身份驗證, 則還需要將以下內容添加到 Flume agent 配置中, 或者可以使用全域 SSL 設定(請參閱 SSL / TLS 支持部分). 每個 Flume agent 都必須擁有其客戶證書, Kafka 經紀人必須單獨或通過其簽名鏈來信任. 常見示例是由單個根 CA 簽署每個客戶端證書, 而后者又由 Kafka 經紀人信任.

# optional, the global keystore can be used alternativelya1.sources.source1.kafka.consumer.ssl.keystore.location=/path/to/client.keystore.jksa1.sources.source1.kafka.consumer.ssl.keystore.password=<password to access the keystore>

** Kerberos 和 Kafka source **

要將 Kafka source 與使用 Kerberos 保護的 Kafka 群集一起使用, 請為消費者設定上面提到的 consumer.security.protocol 屬性.與 Kafka agent 一起使用的 Kerberos 密鑰表和主體在 JAAS 檔案的 "KafkaClient" 部分中指定."客戶端" 部分描述了 Zookeeper 連接(如果需要). 有關 JAAS 檔案內容的資訊, 請參閱 Kafka doc. 可以通過 flume-env.sh 中的 JAVA_OPTS 指定此 JAAS 檔案的位置以及可選的系統范圍的 kerberos 配置:

JAVA_OPTS="$JAVA_OPTS -Djava.security.krb5.conf=/path/to/krb5.conf"JAVA_OPTS="$JAVA_OPTS -Djava.security.auth.login.config=/path/to/flume_jaas.conf"

使用 SASL_PLAINTEXT 的示例安全配置:

a1.sources.source1.type = org.apache.flume.source.kafka.KafkaSourcea1.sources.source1.kafka.Bootstrap.servers = kafka-1:9093,kafka-2:9093,kafka-3:9093a1.sources.source1.kafka.topics = mytopica1.sources.source1.kafka.consumer.group.id = flume-consumera1.sources.source1.kafka.consumer.security.protocol = SASL_PLAINTEXTa1.sources.source1.kafka.consumer.sasl.mechanism = GSSAPIa1.sources.source1.kafka.consumer.sasl.kerberos.service.name = kafka

使用 SASL_SSL 的安全配置示例:

a1.sources.source1.type = org.apache.flume.source.kafka.KafkaSourcea1.sources.source1.kafka.Bootstrap.servers = kafka-1:9093,kafka-2:9093,kafka-3:9093a1.sources.source1.kafka.topics = mytopica1.sources.source1.kafka.consumer.group.id = flume-consumera1.sources.source1.kafka.consumer.security.protocol = SASL_SSLa1.sources.source1.kafka.consumer.sasl.mechanism = GSSAPIa1.sources.source1.kafka.consumer.sasl.kerberos.service.name = kafka# optional, the global truststore can be used alternativelya1.sources.source1.kafka.consumer.ssl.truststore.location=/path/to/truststore.jksa1.sources.source1.kafka.consumer.ssl.truststore.password=<password to access the truststore>

示例 JAAS 檔案. 有關其內容的參考, 請參閱 SASL 配置的 Kafka 檔案中所需認證機制 (GSSAPI / PLAIN) 的客戶端配置部分. 由于 Kafka Source 也可能連接到 Zookeeper 以進行偏移遷移, 因此 "Client" 部分也添加到此示例中. 除非您需要偏移遷移, 否則不需要這樣做, 或者您需要此部分用于其他安全組件. 另外, 請確保 Flume 行程的作業系統用戶對 jaas 和 keytab 檔案具有讀權限.

Client {com.sun.security.auth.module.Krb5LoginModule requireduseKeyTab=truestoreKey=truekeyTab="/path/to/keytabs/flume.keytab"principal="flume/flumehost1.example.com@YOURKERBEROSREALM";};KafkaClient {com.sun.security.auth.module.Krb5LoginModule requireduseKeyTab=truestoreKey=truekeyTab="/path/to/keytabs/flume.keytab"principal="flume/flumehost1.example.com@YOURKERBEROSREALM";};NetCat TCP source

類似于 netcat 的 source, 它偵聽給定埠并將每行文本轉換為 event . 像 nc -k -l [host] [port]這樣的行為. 換句話說, 它打開一個指定的埠并監聽資料. 期望是提供的資料是換行符分隔的文本. 每行文本都轉換為 Flume event , 并通過連接的 channel 發送. 必需屬性以 粗體顯示

Example for agent named a1:a1.sources = r1a1.channels = c1a1.sources.r1.type = netcata1.sources.r1.bind = 0.0.0.0a1.sources.r1.port = 6666a1.sources.r1.channels = c1NetCat UDP source

根據原始的 Netcat(TCP)source, 該 source 監聽給定埠并將每行文本轉換為 event 并通過連接的 channel 發送. 像 nc -u -k -l [host] [port]這樣的行為. 必需屬性以 粗體顯示

agent 名為 a1 的示例:

a1.sources = r1a1.channels = c1a1.sources.r1.type = netcatudpa1.sources.r1.bind = 0.0.0.0a1.sources.r1.port = 6666a1.sources.r1.channels = c1Sequence Generator Source

一個簡單的 sequence 生成器, 它使用從 0 開始的計數器連續生成 event , 遞增 1 并在 totalEvents 處停止. 無法向 channel 發送 event 時重試. 主要用于測驗. 在重試期間, 它使重試訊息的主體保持與以前相同, 以便在目的地重復資料洗掉之后, 唯一 event 的數量應等于指定的 totalEvents. 必需屬性以粗體顯示

Example for agent named a1:a1.sources = r1a1.channels = c1a1.sources.r1.type = seqa1.sources.r1.channels = c1Syslog Source

讀取 syslog 資料并生成 Flume event .UDP source 將整個訊息視為單個 event .TCPsource 為每個由換行符 ('n') 分隔的字串創建一個新 event

必需屬性以粗體顯示

Syslog TCPsource

原始的, 經過驗證的 syslog TCPsource 代碼

For example, a syslog TCP source for agent named a1:a1.sources = r1a1.channels = c1a1.sources.r1.type = syslogtcpa1.sources.r1.port = 5140a1.sources.r1.host = localhosta1.sources.r1.channels = c1

多埠 Syslog TCP source

這是 Syslog TCPsource 的更新, 更快, 多埠版本. 請注意, 埠配置設定已替換埠. 多埠功能意味著它可以以有效的方式一次監聽多個埠. 此 source 使用 Apache Mina 庫來執行此操作. 提供對 RFC-3164 和許多常見 RFC-5424 格式訊息的支持. 還提供了配置基于每個埠的字符集的功能.

For example, a multiport syslog TCP source for agent named a1:a1.sources = r1a1.channels = c1a1.sources.r1.type = multiport_syslogtcpa1.sources.r1.channels = c1a1.sources.r1.host = 0.0.0.0a1.sources.r1.ports = 10001 10002 10003a1.sources.r1.portHeader = portSyslog UDPsource

For example, a syslog UDP source for agent named a1:a1.sources = r1a1.channels = c1a1.sources.r1.type = syslogudpa1.sources.r1.port = 5140a1.sources.r1.host = localhosta1.sources.r1.channels = c1

HTTP Source < 通過 HTTP POST 和 GET 接受 Flumeevent 的 source.GET 應僅用于實驗. HTTP 請求由可插入的 "處理程式" 轉換為 flume event , 該處理程式必須實作 HTTPSourceHandler 介面. 該處理程式獲取 HttpServletRequest 并回傳 flume event 串列.從一個 Http 請求處理的所有 event 都在一個事務中提交給 channel, 從而允許在諸如檔案 channel 之類的 channel 上提高效率. 如果處理程式拋出例外, 則此 source 將回傳 HTTP 狀態 400. 如果 channel 已滿, 或者 source 無法將 event 附加到 channel, 則 source 將回傳 HTTP 503 - 暫時不可用狀態

在一個發布請求中發送的所有 event 都被視為一個批處理, 并在一個事務中插入到 channel 中

此 source 基于 Jetty 9.4, 并提供了設定其他 Jetty 特定引數的功能, 這些引數將直接傳遞給 Jetty 組件

不推薦使用的屬性

NB 使用上面列出的物件上的 setter-methods 設定 Jetty 特定設定. 有關完整的詳細資訊, 請參閱這些類的 Javadoc(, , 和 )

使用特定于 Jetty 的設定時, 上面命名的屬性將優先(例如, excludeProtocols 將優先于 SslContextFactory.ExcludeProtocols). 所有房產都是小寫的.

An example http source for agent named a1:a1.sources = r1a1.channels = c1a1.sources.r1.type = httpa1.sources.r1.port = 5140a1.sources.r1.channels = c1a1.sources.r1.handler = org.example.REST.RestHandlera1.sources.r1.handler.nickname = random propsa1.sources.r1.HttpConfiguration.sendServerVersion = falsea1.sources.r1.ServerConnector.idleTimeout = 300JSONHandler

開箱即用的處理程式可以處理以 JSON 格式表示的 event , 并支持 UTF-8,UTF-16 和 UTF-32 字符集. 處理程式接受一個 event 陣列(即使只有一個 event ,event 必須在陣列中發送), 并根據請求中指定的編碼將它們轉換為 Flumeevent . 如果未指定編碼, 則假定為 UTF-8.JSON 處理程式支持 UTF-8,UTF-16 和 UTF-32.event 表示如下.

[{"headers" : {"timestamp" : "434324343","host" : "random_host.example.com"},"body" : "random_body"},{"headers" : {"namenode" : "namenode.example.com","datanode" : "random_datanode.example.com"},"body" : "really_random_body"}]

要設定 charset, 請求必須具有指定為 application/JSON 的 內容型別 ; charset = UTF-8(根據需要用 UTF-16 或 UTF-32 替換 UTF-8)

以此處理程式所期望的格式創建 event 的一種方法是使用 Flume SDK 中提供的 JSONEvent 并使用 Google Gson 使用 Gson#fromJson(Object,Type)方法創建 JSON 字串. 要作為 event 串列的此方法的第二個引數傳遞的型別標記可以通過以下方式創建:

Type type = new TypeToken<List<JSONEvent>>() {}.getType();BlobHandler

默認情況下, HTTPSource 將 JSON 輸入拆分為 Flumeevent . 作為替代方案, BlobHandler 是 HTTPSource 的處理程式, 它回傳包含請求引數的 event 以及使用此請求上載的二進制大物件(BLOB). 例如 PDF 或 JPG 檔案. 請注意, 此方法不適用于非常大的物件, 因為它會將整個 BLOB 快取在 RAM 中.

Stress source

StressSource 是一個內部負載生成 source 實作, 對壓力測驗非常有用. 它允許用戶使用空標頭配置 event 有效負載的大小. 用戶可以配置要發送的 event 總數以及要傳遞的最大成功 event 數.

必需屬性以粗體顯示

Example for agent named a1:a1.sources = stresssource-1a1.channels = memoryChannel-1a1.sources.stresssource-1.type = org.apache.flume.source.StressSourcea1.sources.stresssource-1.size = 10240a1.sources.stresssource-1.maxTotalEvents = 1000000a1.sources.stresssource-1.channels = memoryChannel-1Legacy Sources

legacy source 允許 Flume 1.x agent 從 Flume 0.9.4 agent 接收 event . 它接受 Flume 0.9.4 格式的 event , 將它們轉換為 Flume 1.0 格式, 并將它們存盤在連接的 channel 中. 0.9.4event 屬性 (如 timestamp,pri,host,nanos 等) 將轉換為 1.xevent 頭屬性. 舊版 source 支持 Avro 和 Thrift RPC 連接. 要在兩個 Flume 版本之間使用此橋接, 您需要使用 avroLegacy 或 thriftLegacysource 啟動 Flume 1.x agent .0.9.4 agent 應該讓 agent Sink 指向 1.x agent 的主機 / 埠.

注意

Flume 1.x 的可靠性語意與 Flume 0.9.x 的可靠性語意不同. 舊版 source 不支持 Flume 0.9.x agent 的 E2E 或 DFO 模式. 唯一支持的 0.9.x 模式是盡力而為, 盡管 1.x 流的可靠性設定將適用于傳統 source 保存到 Flume 1.xchannel 后的 event

必需屬性以粗體顯示

Avro Legacy Source

Example for agent named a1:a1.sources = r1a1.channels = c1a1.sources.r1.type = org.apache.flume.source.avroLegacy.AvroLegacySourcea1.sources.r1.host = 0.0.0.0a1.sources.r1.bind = 6666a1.sources.r1.channels = c1Thrift Legacy Source

Example for agent named a1:a1.sources = r1a1.channels = c1a1.sources.r1.type = org.apache.flume.source.thriftLegacy.ThriftLegacySourcea1.sources.r1.host = 0.0.0.0a1.sources.r1.bind = 6666a1.sources.r1.channels = c1Custom Source

自定義 source 是您自己的 Source 介面實作. 啟動 Flume agent 時, 自定義 source 的類及其依賴項必須包含在 agent 程式的類路徑中. 自定義 source 的型別是其 FQCN.

Example for agent named a1:a1.sources = r1a1.channels = c1a1.sources.r1.type = org.example.MySourcea1.sources.r1.channels = c1Scribe Source

Scribe 是另一種攝取系統. 要采用現有的 Scribe 攝取系統, Flume 應該使用基于 Thrift 的 ScribeSource 和兼容的傳輸協議. 要部署 Scribe, 請遵循 Facebook 的指南. 必需屬性以粗體顯示

Example for agent named a1:a1.sources = r1a1.channels = c1a1.sources.r1.type = org.apache.flume.source.scribe.ScribeSourcea1.sources.r1.port = 1463a1.sources.r1.workerThreads = 5a1.sources.r1.channels = c1Flume SinksHDFS Sink

此 sink 將 event 寫入 Hadoop 分布式檔案系統 (HDFS). 它目前支持創建文本和序列檔案. 它支持兩種檔案 type 的壓縮. 可以根據經過的時間或資料大小或 event 數量定期滾動檔案(關閉當前檔案并創建新檔案). 它還根據 event source 自的時間戳或機器等屬性對資料進行磁區 / 磁區. HDFS 目錄路徑可能包含格式轉義序列, 將由 HDFS sink 替換, 以生成用于存盤 event 的目錄 / 檔案名. 使用此 sink 需要安裝 hadoop, 以便 Flume 可以使用 Hadoop jar 與 HDFS 集群進行通信. 請注意, 需要支持 sync() 呼叫的 Hadoop 版本.

以下是支持的轉義序列:

注意: 轉義字串％[localhost],％[IP]和％[FQDN]都依賴于 Java 獲取 hostname 的能力, 這在某些網路環境中可能會失敗.

正在使用的檔案將在名稱末尾包含 ".tmp". 檔案關閉后, 將洗掉此擴展程式. 這允許排除目錄中的部分完整檔案. 必需屬性以粗體顯示

注意

對于所有與時間相關的轉義序列, event 標題中必須存在帶有 "timestamp" 鍵的標頭(除非 hdfs.useLocalTimeStamp 設定為 true).自動添加此方法的一種方法是使用 TimestampInterceptor.

不推薦使用的屬性:

Example for agent named a1:a1.channels = c1a1.sinks = k1a1.sinks.k1.type = hdfsa1.sinks.k1.channel = c1a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H%M/%Sa1.sinks.k1.hdfs.filePrefix = events-a1.sinks.k1.hdfs.round = truea1.sinks.k1.hdfs.roundValue = https://www.cnblogs.com/laoqing/p/10a1.sinks.k1.hdfs.roundUnit = minute

以上配置將時間戳向下舍入到最后 10 分鐘. 例如, 時間戳為 2012 年 6 月 12 日上午 11:54:34 的 event 將導致 hdfs 路徑變為

1. /flume/events/2012-06-12/1150/00.
2. Hive Sink

此 sink 將包含分隔文本或 JSON 資料的 event 直接流式傳輸到 Hive 表或磁區. event 使用 Hive 事務撰寫. 一旦將一組 event 提交給 Hive, 它們就會立即顯示給 Hive 查詢. 水槽將流入的磁區既可以預先創建, 也可以選擇 Flume 創建它們, 如果它們缺失的話. 傳入 event 資料中的欄位將映射到 Hive 表中的相應列

為 Hive sink 提供了以下序列化程式:

JSON: 處理 UTF8 編碼的 JSON(嚴格語法)event , 不需要配置. JSON 中的物件名稱直接映射到 Hive 表中具有相同名稱的列. 內部使用 org.apache.hive.hcatalog.data.JsonSerDe, 但獨立于 Hive 表的 Serde. 此序列化程式需要安裝 HCatalog.

DELIMITED: 處理簡單的分隔文本 event . 內部使用 LazySimpleSerde, 但獨立于 Hive 表的 Serde.

以下是支持的轉義序列:

注意

對于所有與時間相關的轉義序列, event 標題中必須存在具有鍵 "timestamp" 的標頭(除非 useLocalTimeStamp 設定為 true). 自動添加此方法的一種方法是使用 TimestampInterceptor.

示例 Hive 表:

create table weblogs ( id int , msg string )partitioned by (continent string, country string, time string)clustered by (id) into 5 bucketsstored as orc;Example for agent named a1:a1.channels = c1a1.channels.c1.type = memorya1.sinks = k1a1.sinks.k1.type = hivea1.sinks.k1.channel = c1a1.sinks.k1.hive.metastore = thrift://127.0.0.1:9083a1.sinks.k1.hive.database = logsdba1.sinks.k1.hive.table = weblogsa1.sinks.k1.hive.partition = asia,%{country},%y-%m-%d-%H-%Ma1.sinks.k1.useLocalTimeStamp = falsea1.sinks.k1.round = truea1.sinks.k1.roundValue = https://www.cnblogs.com/laoqing/p/10a1.sinks.k1.roundUnit = minutea1.sinks.k1.serializer = DELIMITEDa1.sinks.k1.serializer.delimiter ="\t"a1.sinks.k1.serializer.serdeSeparator = '\t'a1.sinks.k1.serializer.fieldnames =id,,msg

以上配置將時間戳向下舍入到最后 10 分鐘. 例如, 將時間戳標頭設定為 2012 年 6 月 12 日上午 11:54:34 且 "country" 標頭設定為 "india" 的 event 將評估為磁區(continent ='asia',country ='india',time ='2012-06-12-11-50'. 序列化程式配置為接受包含三個欄位的制表符分隔輸入并跳過第二個欄位.

Logger Sink

在 INFO 級別記錄 event . 通常用于測驗 / 除錯目的. 必需屬性以粗體顯示. 此 sink 是唯一的例外, 它不需要在 "記錄原始資料" 部分中說明的額外配置.

Example for agent named a1:a1.channels = c1a1.sinks = k1a1.sinks.k1.type = loggera1.sinks.k1.channel = c1Avro Sink

這個 Sink 形成了 Flume 的分層收集支持的一半. 發送到此 sink 的 Flume event 將轉換為 Avro event 并發送到配置的 hostname/port pair .event 將從配置的 channel 中批量獲取配置的批處理大小. 必需屬性以粗體顯示

Example for agent named a1:a1.channels = c1a1.sinks = k1a1.sinks.k1.type = avroa1.sinks.k1.channel = c1a1.sinks.k1.hostname = 10.10.10.10a1.sinks.k1.port = 4545Thrift Sink

這個水槽形成了 Flume 的分層收集支持的一半. 發送到此 sink 的 Flume event 將轉換為 Thrift event 并發送到配置的 hostname/port pair.event 將從配置的 channel 中批量獲取配置的批處理大小.

可以通過啟用 kerberos 身份驗證將 Thrift sink 配置為以安全模式啟動. 要與以安全模式啟動的 Thriftsource 通信, Thriftsink 也應該以安全模式運行. client-principal 和 client-keytab 是 Thriftsink 用于向 kerberos KDC 進行身份驗證的屬性. server-principal 表示 Thriftsource 的主體, 此 sink 配置為以安全模式連接. 必需屬性以粗體顯示

Example for agent named a1:a1.channels = c1a1.sinks = k1a1.sinks.k1.type = thrifta1.sinks.k1.channel = c1a1.sinks.k1.hostname = 10.10.10.10a1.sinks.k1.port = 4545IRC Sink

IRC sink 從附加 channel 接收訊息, 并將這些訊息中繼到已配置的 IRC 目標. 必需屬性以粗體顯示

Example for agent named a1:a1.channels = c1a1.sinks = k1a1.sinks.k1.type = irca1.sinks.k1.channel = c1a1.sinks.k1.hostname = irc.yourdomain.coma1.sinks.k1.nick = flumea1.sinks.k1.chan = #flumeFile Roll Sink

在本地檔案系統上存盤 event . 必需屬性以粗體顯示

Example for agent named a1:a1.channels = c1a1.sinks = k1a1.sinks.k1.type = file_rolla1.sinks.k1.channel = c1a1.sinks.k1.sink.directory = /var/log/flumeNull Sink

丟棄從 channel 收到的所有 event . 必需屬性以粗體顯示

Example for agent named a1:a1.channels = c1a1.sinks = k1a1.sinks.k1.type = nulla1.sinks.k1.channel = c1HBase Sink

此 sink 將資料寫入 HBase.Hbase 配置是從類路徑中遇到的第一個 hbase-site.xml 中獲取的. 實作由配置指定的 HbaseEventSerializer 的類用于將 event 轉換為 HBase put 和 / 或 increments . 然后將這些 put 和 increments 寫入 HBase. 該 sink 提供與 HBase 相同的一致性保證, HBase 是當前行的原子性. 如果 Hbase 無法寫入某些 event , 則 sink 將重播該事務中的所有 event

HBase Sink 支持將資料寫入安全 HBase. 要寫入安全 HBase, agent 程式正在運行的用戶必須具有對 sink 配置為寫入的表的寫入權限. 可以在配置中指定用于對 KDC 進行身份驗證的主體和密鑰表. Flume agent 程式類路徑中的 hbase-site.xml 必須將身份驗證設定為 kerberos(有關如何執行此操作的詳細資訊, 請參閱 HBase 檔案)

為方便起見, Flume 提供了兩個序列化器.

SimpleHbaseEventSerializer(org.apache.flume.sink.hbase.SimpleHbaseEventSerializer)按原樣將 event 主體寫入 HBase, 并可選擇增加 Hbase 中的列. 這主要是一個示例實作.

RegexHbaseEventSerializer(org.apache.flume.sink.hbase.RegexHbaseEventSerializer)根據給定的正則運算式打破 event 體, 并將每個部分寫入不同的列.

type 是 FQCN:org.apache.flume.sink.hbase.HBaseSink.

必需屬性以粗體顯示

Example for agent named a1:a1.channels = c1a1.sinks = k1a1.sinks.k1.type = hbasea1.sinks.k1.table = foo_tablea1.sinks.k1.columnFamily = bar_cfa1.sinks.k1.serializer = org.apache.flume.sink.hbase.RegexHbaseEventSerializera1.sinks.k1.channel = c1HBase2 Sink

HBase2Sink 相當于 HBase 版本 2 的 HBaseSink. 提供的功能和配置引數與 HBaseSink 的情況相同(除了 sinktype 中的 hbase2 標簽和包 / 類名稱)

type 是 FQCN:org.apache.flume.sink.hbase2.HBase2Sink

必需屬性以粗體顯示

Example for agent named a1:a1.channels = c1a1.sinks = k1a1.sinks.k1.type = hbase2a1.sinks.k1.table = foo_tablea1.sinks.k1.columnFamily = bar_cfa1.sinks.k1.serializer = org.apache.flume.sink.hbase2.RegexHBase2EventSerializera1.sinks.k1.channel = c1AsyncHBaseSink

此 sink 使用異步模型將資料寫入 HBase. 實作由配置指定的 AsyncHbaseEventSerializer 的類用于將 event 轉換為 HBase put 和 / 或 increments. 然后將這些放置和增量寫入 HBase. 此 sink 使用 Asynchbase API https://github.com/OpenTSDB/asynchbase 寫入 HBase. 該 sink 提供與 HBase 相同的一致性保證, HBase 是當前行的原子性. 如果 Hbase 無法寫入某些 event , 則 sink 將重播該事務中的所有 event .AsyncHBaseSink 只能與 HBase 1.x 一起使用. AsyncHBaseSink 使用的異步客戶端庫不適用于 HBase 2.type 為 FQCN:org.apache.flume.sink.hbase.AsyncHBaseSink. 必需屬性以粗體顯示

請注意, 此 sink 在配置中獲取 Zookeeper Quorum 和父 znode 資訊. 可以在 flume 組態檔中指定 Zookeeper Quorum 和父節點配置. 或者, 這些配置值取自類路徑中的第一個 hbase-site.xml 檔案.

如果配置中未提供這些, 則 sink 將從類路徑中的第一個 hbase-site.xml 檔案中讀取此資訊.

Example for agent named a1:a1.channels = c1a1.sinks = k1a1.sinks.k1.type = asynchbasea1.sinks.k1.table = foo_tablea1.sinks.k1.columnFamily = bar_cfa1.sinks.k1.serializer = org.apache.flume.sink.hbase.SimpleAsyncHbaseEventSerializera1.sinks.k1.channel = c1MorphlineSolrSink

此 sink 從 Flume event 中提取資料, 對其進行轉換, 并將其近乎實時地加載到 Apache Solr 服務器中, 后者又向最終用戶或搜索應用程式提供查詢.

此 sink 非常適合將原始資料流式傳輸到 HDFS(通過 HdfsSink)并同時提取, 轉換并將相同資料加載到 Solr(通過 MorphlineSolrSink)的用例. 特別是, 此 sink 可以處理來自不同資料 source 的任意異構原始資料, 并將其轉換為對搜索應用程式有用的資料模型

ETL 功能可使用 morphline 組態檔進行自定義, 該檔案定義了一系列轉換命令, 用于將 event 記錄從一個命令傳遞到另一個命令

Morphlines 可以看作是 Unix 管道的演變, 其中資料模型被推廣為使用通用記錄流, 包括任意二進制有效載荷. morphline 命令有點像 Flume 攔截器. Morphlines 可以嵌入到 Flume 等 Hadoop 組件中.

用于決議和轉換一組標準資料格式 (如日志檔案, Avro,CSV, 文本, HTML,xml,PDF,Word,Excel 等) 的命令是開箱即用的, 還有其他自定義命令和決議器用于其他資料格式可以添加為 morphline 插件. 可以索引任何 type 的資料格式, 并且可以生成任何 type 的 Solr 模式的任何 Solr 檔案, 并且可以注冊和執行任何自定義 ETL 邏輯.

Morphlines 操縱連續的記錄流. 資料模型可以描述如下: 記錄是一組命名欄位, 其中每個欄位具有一個或多個值的有序串列. 值可以是任何 Java 物件. 也就是說, 記錄本質上是一個哈希表, 其中每個哈希表條目包含一個 String 鍵和一個 Java 物件串列作為值.(該實作使用 Guava 的 ArrayListMultimap, 它是一個 ListMultimap). 請注意, 欄位可以具有多個值, 并且任何兩個記錄都不需要使用公共欄位名稱.

這個 sink 將 Flumeevent 的主體填充到_attachment_body 中記錄欄位中, 并將 Flumeevent 的標頭復制到同名的記錄欄位中. 然后命令可以對此資料執行操作.

支持路由到 SolrCloud 集群以提高可伸縮性. 索引負載可以分布在大量 MorphlineSolrSink 上, 以提高可伸縮性. 可以跨多個 MorphlineSolrSink 復制索引負載以實作高可用性, 例如使用 Flume 功能, 例如負載平衡 sink 處理器. MorphlineInterceptor 還可以幫助實作到多個 Solr 集合的動態路由(例如, 用于多租戶)

必須將環境所需的 morphline 和 Solr jar 放在 Apache Flume 安裝的 lib 目錄中.

type 是 FQCN:org.apache.flume.sink.Solr.morphline.MorphlineSolrSink

必需屬性以粗體顯示

Example for agent named a1:a1.channels = c1a1.sinks = k1a1.sinks.k1.type = org.apache.flume.sink.Solr.morphline.MorphlineSolrSinka1.sinks.k1.channel = c1a1.sinks.k1.morphlineFile = /etc/flume-ng/conf/morphline.conf# a1.sinks.k1.morphlineId = morphline1# a1.sinks.k1.batchSize = 1000# a1.sinks.k1.batchDurationMillis = 1000ElasticSearchSink

此 sink 將資料寫入彈性搜索集群. 默認情況下, 將寫入 event 以便 Kibana http://kibana.org/ 圖形界面可以顯示它們 - 就像 logstash 一樣 https://logstash.net/ 撰寫它們一樣

必須將環境所需的 Elasticsearch 和 lucene-core jar 放在 Apache Flume 安裝的 lib 目錄中. Elasticsearch 要求客戶端 JAR 的主要版本與服務器的主要版本匹配, 并且兩者都運行相同的 JVM 次要版本. 如果這不正確, 將出現 SerializationExceptions. 要選擇所需的版本, 請首先確定 Elasticsearch 的版本以及目標群集正在運行的 JVM 版本. 然后選擇與主要版本匹配的 Elasticsearch 客戶端庫.0.19.x 客戶端可以與 0.19.x 群集通信; 0.20.x 可以與 0.20.x 對話, 0.90.x 可以與 0.90.x 對話. 確定 Elasticsearch 版本后, 讀取 pom.xml 檔案以確定要使用的正確 lucene-core JAR 版本.

event 將每天寫入新索引. 名稱將是 < indexName> -yyyy-MM-dd, 其中 < indexName > 是 indexName 引數. sink 將在午夜 UTC 開始寫入新索引.

默認情況下, ElasticSearchLogStashEventSerializer 會為彈性搜索序列化 event . 可以使用 serializer 引數覆寫此行為. 此引數接受 org.apache.flume.sink.Elasticsearch.ElasticSearchEventSerializer 或 org.apache.flume.sink.Elasticsearch.ElasticSearchIndexRequestBuilderFactory 的實作. 不推薦使用 ElasticSearchEventSerializer 來支持更強大的 ElasticSearchIndexRequestBuilderFactory

type 是 FQCN:org.apache.flume.sink.Elasticsearch.ElasticSearchSink

必需屬性以粗體顯示

注意

使用標頭替換可以方便地使用 event 標頭的值來動態決定在存盤 event 時要使用的 indexName 和 indexType. 使用此功能時應謹慎, 因為 event 提交者現在可以控制 indexName 和 indexType. 此外, 如果使用 Elasticsearch REST 客戶端, 則 event 提交者可以控制所使用的 URL 路徑

Example for agent named a1:a1.channels = c1a1.sinks = k1a1.sinks.k1.type = Elasticsearcha1.sinks.k1.hostNames = 127.0.0.1:9200,127.0.0.2:9300a1.sinks.k1.indexName = foo_indexa1.sinks.k1.indexType = bar_typea1.sinks.k1.clusterName = foobar_clustera1.sinks.k1.batchSize = 500a1.sinks.k1.ttl = 5da1.sinks.k1.serializer = org.apache.flume.sink.Elasticsearch.ElasticSearchDynamicSerializera1.sinks.k1.channel = c1Kite Dataset Sink

將 event 寫入 Kite 資料集的 http://kitesdk.org/docs/current/guide/ 實驗 sink. 此 sink 將反序列化每個傳入 event 的主體, 并將結果記錄存盤在風箏資料集中. 它通過按 URI 加載資料集來確定目標資料集.

唯一支持的序列化的 Avro 和記錄模式必須在 event 標題使用或者被通過, flume.avro.schema.literal 與 JSON 模式表示或 flume.avro.schema.url 有一個網址, 該模式可能會發現(hdfs:/ ... 支持 URI). 這與使用 deserializer.schemaType = LITERAL 的 Log4jAppender 水槽客戶端和 false 脫機目錄 source 的 Avro 反序列化器 兼容.

注 1: 不支持 flume.avro.schema.hash 標頭. 注意 2: 在某些情況下, 在超過滾動間隔后可能會略微發生檔案滾動. 但是, 此延遲不會超過 5 秒. 在大多數情況下, 延遲是可以忽略的.

Kafka Sink

這是一個 Flume Sink 實作, 可以將資料發布到 Kafka http://kafka.apache.org/ 主題. 其中一個目標是將 Flume 與 Kafka 集成, 以便基于拉的處理系統可以處理來自各種 Flume source 的資料.

這目前支持 Kafka 服務器版本 0.10.1.0 或更高版本. 測驗完成了 2.0.1, 這是發布時最高的可用版本

必需屬性以粗體字體標記

注意

Kafka Sink 使用 Flume Event 標頭中的 topic 和 key 屬性將 event 發送到 Kafka. 如果 topic 中存在 headers , 則會將 event 發送到該特定 topic, 從而覆寫為 Sink 配置的主題. 如果 key 在 topic 存在, 關鍵還是使用 Kafkatopic 磁區磁區資料. 具有相同密鑰的 event 將發送到同一磁區. 如果密鑰為空, 則將 event 發送到隨機磁區.

Kafka sink 還為 key.serializer(org.apache.kafka.common.serialization.StringSerializer)和 value.serializer(org.apache.kafka.common.serialization.ByteArraySerializer)

提供默認值. 不建議修改這些引數

不推薦使用的屬性:

下面給出 Kafka sink 的示例配置. 以 kafka 生產者前綴 kafka.producer 開頭的屬性. 創建 Kafka 生成器時傳遞的屬性不限于此示例中給出的屬性. 此外, 可以在此處包含您的自定義屬性, 并通過作為方法引數傳入的 Flume Context 物件在前處理器中訪問它們.

a1.sinks.k1.channel = c1a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSinka1.sinks.k1.kafka.topic = mytopica1.sinks.k1.kafka.Bootstrap.servers = localhost:9092a1.sinks.k1.kafka.flumeBatchSize = 20a1.sinks.k1.kafka.producer.acks = 1a1.sinks.k1.kafka.producer.linger.ms = 1a1.sinks.k1.kafka.producer.compression.type = snappy

安全和 Kafka Sink

Flume 和 Kafka 之間的通信 channel 支持安全認證和資料加密. 對于安全身份驗證, 可以使用 Kafka 0.9.0 版中的 SASL / GSSAPI(Kerberos V5)或 SSL(即使該引數名為 SSL, 實際協議是 TLS 實作)

截至目前, 資料加密僅由 SSL / TLS 提供.

將 kafka.producer.security.protocol 設定為以下任何值意味著:

SASL_PLAINTEXT - 無資料加密的 Kerberos 或純文本身份驗證

SASL_SSL - 使用資料加密的 Kerberos 或純文本身份驗證

SSL - 基于 TLS 的加密, 帶有可選的身份驗證

警告

啟用 SSL 時性能會下降, 其大小取決于 CPUtype 和 JVM 實作. 參考: Kafka 安全概述 和跟蹤此問題的 jira: https://issues/jira/browse/KAFKA-2561

TLS 和 Kafka Sink:

請閱讀配置 Kafka 客戶端 SSL 中描述的步驟, 以了解用于微調的其他配置設定, 例如以下任何一項: 安全提供程式, 密碼套件, 啟用的協議, 信任庫或密鑰庫 type.

配置服務器端身份驗證和資料加密的示例.

在此處指定信任庫是可選的, 可以使用全域信任庫. 有關全域 SSL 設定的更多詳細資訊, 請參閱 SSL / TLS 支持部分.

注意: 默認情況下, 未定義屬性 ssl.endpoint.identification.algorithm, 因此不會執行 hostname 驗證. 要啟用 hostname 驗證, 請設定以下屬性

a1.sinks.sink1.kafka.producer.ssl.endpoint.identification.algorithm = HTTPS

啟用后, 客戶端將根據以下兩個欄位之一驗證服務器的完全限定域名(FQDN):

通用名稱(CN) https://tools.ietf.org/html/rfc6125#section-2.3 https://tools.ietf.org/html/rfc6125#section-2.3

主題備選名稱(SAN)

如果還需要客戶端身份驗證, 則還需要將以下內容添加到 Flume agent 配置中, 或者可以使用全域 SSL 設定(請參閱 SSL / TLS 支持部分). 每個 Flume agent 都必須擁有其客戶證書, Kafka 經紀人必須單獨或通過其簽名鏈來信任. 常見示例是由單個根 CA 簽署每個客戶端證書, 而后者又由 Kafka 經紀人信任

# optional, the global keystore can be used alternativelya1.sinks.sink1.kafka.producer.ssl.keystore.location = /path/to/client.keystore.jksa1.sinks.sink1.kafka.producer.ssl.keystore.password = <password to access the keystore>

如果密鑰庫和密鑰使用不同的密碼保護, 則 ssl.key.password 屬性將為生產者密鑰庫提供所需的額外密鑰

a1.sinks.sink1.kafka.producer.ssl.key.password = <password to access the key>

Kerberos 和 Kafka Sink

要將 Kafkasink 與使用 Kerberos 保護的 Kafka 集群一起使用, 請為生產者設定上面提到的 producer.security.protocol 屬性. 與 Kafka agent 一起使用的 Kerberos 密鑰表和主體在 JAAS 檔案的 "KafkaClient" 部分中指定."客戶端" 部分描述了 Zookeeper 連接(如果需要). 有關 JAAS 檔案內容的資訊, 請參閱 Kafka doc. 可以通過 flume-env.sh 中的 JAVA_OPTS 指定此 JAAS 檔案的位置以及可選的系統范圍的 kerberos 配置:

JAVA_OPTS="$JAVA_OPTS -Djava.security.krb5.conf=/path/to/krb5.conf"JAVA_OPTS="$JAVA_OPTS -Djava.security.auth.login.config=/path/to/flume_jaas.conf"

使用 SASL_PLAINTEXT 的示例安全配置:

a1.sinks.sink1.type = org.apache.flume.sink.kafka.KafkaSinka1.sinks.sink1.kafka.Bootstrap.servers = kafka-1:9093,kafka-2:9093,kafka-3:9093a1.sinks.sink1.kafka.topic = mytopica1.sinks.sink1.kafka.producer.security.protocol = SASL_PLAINTEXTa1.sinks.sink1.kafka.producer.sasl.mechanism = GSSAPIa1.sinks.sink1.kafka.producer.sasl.kerberos.service.name = kafka

使用 SASL_SSL 的安全配置示例:

a1.sinks.sink1.type = org.apache.flume.sink.kafka.KafkaSinka1.sinks.sink1.kafka.Bootstrap.servers = kafka-1:9093,kafka-2:9093,kafka-3:9093a1.sinks.sink1.kafka.topic = mytopica1.sinks.sink1.kafka.producer.security.protocol = SASL_SSLa1.sinks.sink1.kafka.producer.sasl.mechanism = GSSAPIa1.sinks.sink1.kafka.producer.sasl.kerberos.service.name = kafka# optional, the global truststore can be used alternativelya1.sinks.sink1.kafka.producer.ssl.truststore.location = /path/to/truststore.jksa1.sinks.sink1.kafka.producer.ssl.truststore.password = <password to access the truststore>

示例 JAAS 檔案. 有關其內容的參考, 請參閱 SASL 配置的 Kafka 檔案中所需認證機制 (GSSAPI / PLAIN) 的客戶端配置部分. 與 Kafka Source 或 Kafka Channel 不同, 不需要 "Client" 部分, 除非其他連接組件需要它. 另外, 請確保 Flume 行程的作業系統用戶對 jaas 和 keytab 檔案具有讀權限.

KafkaClient {com.sun.security.auth.module.Krb5LoginModule requireduseKeyTab=truestoreKey=truekeyTab="/path/to/keytabs/flume.keytab"principal="flume/flumehost1.example.com@YOURKERBEROSREALM";};HTTP Sink

此 sink 的行為是它將從 channel 獲取 event , 并使用 HTTP POST 請求將這些 event 發送到遠程服務. event 內容作為 POST 正文發送

此 sink 的錯誤處理行為取決于目標服務器回傳的 HTTP 回應. sink 退避 / 就緒狀態是可配置的, 事務提交 / 回滾結果以及 event 是否有助于成功的 event 排放計數也是可配置的

狀態代碼不可讀的服務器回傳的任何格式錯誤的 HTTP 回應都將導致退避信號, 并且不會從該 channel 中消耗該 event

必需屬性以粗體顯示

請注意, 最具體的 HTTP 狀態代碼匹配用于 backoff,rollback 和 incrementMetrics 配置選項. 如果存在 2XX 和 200 狀態代碼的配置值, 則 200 個 HTTP 代碼將使用 200 值, 而 201-299 范圍內的所有其他 HTTP 代碼將使用 2XX 值.

消耗任何慷訓空 event 而不向 HTTP 端點發出任何請求.

Example for agent named a1:a1.channels = c1a1.sinks = k1a1.sinks.k1.type = httpa1.sinks.k1.channel = c1a1.sinks.k1.endpoint = http://localhost:8080/someuria1.sinks.k1.connectTimeout = 2000a1.sinks.k1.requestTimeout = 2000a1.sinks.k1.acceptHeader = application/JSONa1.sinks.k1.contentTypeHeader = application/JSONa1.sinks.k1.defaultBackoff = truea1.sinks.k1.defaultRollback = truea1.sinks.k1.defaultIncrementMetrics = falsea1.sinks.k1.backoff.4XX = falsea1.sinks.k1.rollback.4XX = falsea1.sinks.k1.incrementMetrics.4XX = truea1.sinks.k1.backoff.200 = falsea1.sinks.k1.rollback.200 = falsea1.sinks.k1.incrementMetrics.200 = trueCustom Sink

自定義 sink 是您自己的 Sink 介面實作. 啟動 Flume agent 程式時, 必須在 agent 程式的類路徑中包含自定義 sink 的類及其依賴項.自定義 sink 的 type 是其 FQCN. 必需屬性以粗體顯示

Example for agent named a1:a1.channels = c1a1.sinks = k1a1.sinks.k1.type = org.example.MySinka1.sinks.k1.channel = c1Flume Channels

channel 是 event 在 agent 上暫存的存盤庫. Source 添加 event ,Sink 洗掉它.

Memory Channel

event 存盤在具有可配置最大大小的記憶體中佇列中. 它非常適合需要更高吞吐量的流量, 并且在 agent 發生故障時準備丟失分階段資料. 必需屬性以粗體顯示

Example for agent named a1:a1.channels = c1a1.channels.c1.type = memorya1.channels.c1.capacity = 10000a1.channels.c1.transactionCapacity = 10000a1.channels.c1.byteCapacityBufferPercentage = 20a1.channels.c1.byteCapacity = 800000JDBC Channel

event 存盤在由資料庫支持的持久存盤中. JDBC Channel 當前支持嵌入式 Derby. 這是一個持久的 channel, 非常適合可恢復性很重要的流程. 必需屬性以粗體顯示

Example for agent named a1:a1.channels = c1a1.channels.c1.type = jdbcKafka Channel

event 存盤在 Kafka 群集中(必須單獨安裝).Kafka 提供高可用性和復制, 因此如果 agent 或 kafka agent 崩潰, event 可立即用于其他 sinks

Kafka 頻道可用于多種場景:

使用 Flume source 和 sink - 它為 event 提供了可靠且高度可用的 channel

使用 Flume source 和攔截器但沒有 sink - 它允許將 Flumeevent 寫入 Kafka 主題, 供其他應用程式使用

使用 Flume sink, 但沒有 source - 它是一種低延遲, 容錯的方式將 event 從 Kafka 發送到 Flumesink, 如 HDFS,HBase 或 Solr

這目前支持 Kafka 服務器版本 0.10.1.0 或更高版本. 測驗完成了 2.0.1, 這是發布時最高的可用版本.

配置引陣列織如下:

通常與 channel 相關的配置值應用于 channel 配置級別, 例如: a1.channel.k1.type =

與 Kafka 相關的配置值或 Channel 運行的前綴以 "kafka." 為前綴(這對 CommonClient Configs 是有效的), 例如: a1.channel.k1.kafka.topic 和 a1.channel.k1.kafka.Bootstrap.servers. 這與 hdfssink 的運行方式沒有什么不同

特定于生產者 / 消費者的屬性以 kafka.producer 或 kafka.consumer 為前綴

在可能的情況下, 使用 Kafka 引數名稱, 例如: Bootstrap.servers 和 acks

此版本的 flume 與以前的版本向后兼容, 但是下表中顯示了已棄用的屬性, 并且在組態檔中存在時會在啟動時記錄警告訊息.

必需屬性以粗體顯示

不推薦使用的屬性:

注意

Example for agent named a1:a1.channels.channel1.type = org.apache.flume.channel.kafka.KafkaChannela1.channels.channel1.kafka.Bootstrap.servers = kafka-1:9092,kafka-2:9092,kafka-3:9092a1.channels.channel1.kafka.topic = channel1a1.channels.channel1.kafka.consumer.group.id = flume-consumer

安全和 Kafka 頻道:

Flume 和 Kafka 之間的通信 channel 支持安全認證和資料加密. 對于安全身份驗證, 可以使用 Kafka 0.9.0 版中的 SASL / GSSAPI(Kerberos V5)或 SSL(即使該引數名為 SSL, 實際協議是 TLS 實作)

截至目前, 資料加密僅由 SSL / TLS 提供

將 kafka.producer | consumer.security.protocol 設定為以下任何值意味著:

SASL_PLAINTEXT - 無資料加密的 Kerberos 或純文本身份驗證

SASL_SSL - 使用資料加密的 Kerberos 或純文本身份驗證

SSL - 基于 TLS 的加密, 帶有可選的身份驗證.

警告

啟用 SSL 時性能會下降, 其大小取決于 CPUtype 和 JVM 實作. 參考: Kafka 安全概述 和跟蹤此問題的 jira: https://issues/jira/browse/KAFKA-2561

** TLS 和 Kafka 頻道:**

請閱讀配置 Kafka 客戶端 SSL 中描述的步驟 以了解用于微調的其他配置設定, 例如以下任何一項: 安全提供程式, 密碼套件, 啟用的協議, 信任庫或密鑰庫 type.

配置服務器端身份驗證和資料加密的示例.

a1.channels.channel1.type = org.apache.flume.channel.kafka.KafkaChannela1.channels.channel1.kafka.Bootstrap.servers = kafka-1:9093,kafka-2:9093,kafka-3:9093a1.channels.channel1.kafka.topic = channel1a1.channels.channel1.kafka.consumer.group.id = flume-consumera1.channels.channel1.kafka.producer.security.protocol = SSL# optional, the global truststore can be used alternativelya1.channels.channel1.kafka.producer.ssl.truststore.location = /path/to/truststore.jksa1.channels.channel1.kafka.producer.ssl.truststore.password = <password to access the truststore>a1.channels.channel1.kafka.consumer.security.protocol = SSL# optional, the global truststore can be used alternativelya1.channels.channel1.kafka.consumer.ssl.truststore.location = /path/to/truststore.jksa1.channels.channel1.kafka.consumer.ssl.truststore.password = <password to access the truststore>

在此處指定信任庫是可選的, 可以使用全域信任庫. 有關全域 SSL 設定的更多詳細資訊, 請參閱 SSL / TLS 支持部分.

注意: 默認情況下, 未定義屬性 ssl.endpoint.identification.algorithm, 因此不會執行 hostname 驗證. 要啟用 hostname 驗證, 請設定以下屬性

1. a1.channels.channel1.kafka.producer.ssl.endpoint.identification.algorithm = HTTPS
2. a1.channels.channel1.kafka.consumer.ssl.endpoint.identification.algorithm = HTTPS

啟用后, 客戶端將根據以下兩個欄位之一驗證服務器的完全限定域名(FQDN):

通用名稱(CN) https://tools.ietf.org/html/rfc6125#section-2.3 https://tools.ietf.org/html/rfc6125#section-2.3

主題備選名稱(SAN)

如果還需要客戶端身份驗證, 則還需要將以下內容添加到 Flume agent 配置中, 或者可以使用全域 SSL 設定(請參閱 SSL / TLS 支持部分). 每個 Flume agent 都必須擁有其客戶證書, Kafka 經紀人必須單獨或通過其簽名鏈來信任. 常見示例是由單個根 CA 簽署每個客戶端證書, 而后者又由 Kafka 經紀人信任.

# optional, the global keystore can be used alternativelya1.channels.channel1.kafka.producer.ssl.keystore.location = /path/to/client.keystore.jksa1.channels.channel1.kafka.producer.ssl.keystore.password = <password to access the keystore># optional, the global keystore can be used alternativelya1.channels.channel1.kafka.consumer.ssl.keystore.location = /path/to/client.keystore.jksa1.channels.channel1.kafka.consumer.ssl.keystore.password = <password to access the keystore>

如果密鑰庫和密鑰使用不同的密碼保護, 則 ssl.key.password 屬性將為使用者和生產者密鑰庫提供所需的額外密鑰:

a1.channels.channel1.kafka.producer.ssl.key.password = <password to access the key>a1.channels.channel1.kafka.consumer.ssl.key.password = <password to access the key>

** Kerberos 和 Kafka 頻道:**

要將 Kafkachannel 與使用 Kerberos 保護的 Kafka 群集一起使用, 請為生產者和 / 或使用者設定上面提到的 producer / consumer.security.protocol 屬性. 與 Kafka agent 一起使用的 Kerberos 密鑰表和主體在 JAAS 檔案的 "KafkaClient" 部分中指定."客戶端" 部分描述了 Zookeeper 連接(如果需要). 有關 JAAS 檔案內容的資訊, 請參閱 Kafka doc. 可以通過 flume-env.sh 中的 JAVA_OPTS 指定此 JAAS 檔案的位置以及可選的系統范圍的 kerberos 配置:

JAVA_OPTS="$JAVA_OPTS -Djava.security.krb5.conf=/path/to/krb5.conf"JAVA_OPTS="$JAVA_OPTS -Djava.security.auth.login.config=/path/to/flume_jaas.conf"

使用 SASL_PLAINTEXT 的示例安全配置:

a1.channels.channel1.type = org.apache.flume.channel.kafka.KafkaChannela1.channels.channel1.kafka.Bootstrap.servers = kafka-1:9093,kafka-2:9093,kafka-3:9093a1.channels.channel1.kafka.topic = channel1a1.channels.channel1.kafka.consumer.group.id = flume-consumera1.channels.channel1.kafka.producer.security.protocol = SASL_PLAINTEXTa1.channels.channel1.kafka.producer.sasl.mechanism = GSSAPIa1.channels.channel1.kafka.producer.sasl.kerberos.service.name = kafkaa1.channels.channel1.kafka.consumer.security.protocol = SASL_PLAINTEXTa1.channels.channel1.kafka.consumer.sasl.mechanism = GSSAPIa1.channels.channel1.kafka.consumer.sasl.kerberos.service.name = kafka

使用 SASL_SSL 的安全配置示例:

a1.channels.channel1.type = org.apache.flume.channel.kafka.KafkaChannela1.channels.channel1.kafka.Bootstrap.servers = kafka-1:9093,kafka-2:9093,kafka-3:9093a1.channels.channel1.kafka.topic = channel1a1.channels.channel1.kafka.consumer.group.id = flume-consumera1.channels.channel1.kafka.producer.security.protocol = SASL_SSLa1.channels.channel1.kafka.producer.sasl.mechanism = GSSAPIa1.channels.channel1.kafka.producer.sasl.kerberos.service.name = kafka# optional, the global truststore can be used alternativelya1.channels.channel1.kafka.producer.ssl.truststore.location = /path/to/truststore.jksa1.channels.channel1.kafka.producer.ssl.truststore.password = <password to access the truststore>a1.channels.channel1.kafka.consumer.security.protocol = SASL_SSLa1.channels.channel1.kafka.consumer.sasl.mechanism = GSSAPIa1.channels.channel1.kafka.consumer.sasl.kerberos.service.name = kafka# optional, the global truststore can be used alternativelya1.channels.channel1.kafka.consumer.ssl.truststore.location = /path/to/truststore.jksa1.channels.channel1.kafka.consumer.ssl.truststore.password = <password to access the truststore>

示例 JAAS 檔案. 有關其內容的參考, 請參閱 SASL 配置的 Kafka 檔案中所需認證機制 (GSSAPI / PLAIN) 的客戶端配置部分. 由于 Kafka Source 也可能連接到 Zookeeper 以進行偏移遷移, 因此 "Client" 部分也添加到此示例中. 除非您需要偏移遷移, 否則不需要這樣做, 或者您需要此部分用于其他安全組件. 另外, 請確保 Flume 行程的作業系統用戶對 jaas 和 keytab 檔案具有讀權限.

Client {com.sun.security.auth.module.Krb5LoginModule requireduseKeyTab=truestoreKey=truekeyTab="/path/to/keytabs/flume.keytab"principal="flume/flumehost1.example.com@YOURKERBEROSREALM";};KafkaClient {com.sun.security.auth.module.Krb5LoginModule requireduseKeyTab=truestoreKey=truekeyTab="/path/to/keytabs/flume.keytab"principal="flume/flumehost1.example.com@YOURKERBEROSREALM";};File Channel

必需屬性以粗體顯示

注意

默認情況下, 檔案 channel 使用上面指定的用戶主目錄內的檢查點和資料目錄的路徑. 因此, 如果 agent 中有多個活動的檔案 channel 實體, 則只有一個實體可以鎖定目錄并導致其他 channel 初始化失敗.

因此, 有必要為所有已配置的 channel 提供顯式路徑, 最好是在不同的磁盤上

此外, 由于檔案 channel 將在每次提交后同步到磁盤, 因此將其與將 event 一起批處理的 sink/source 耦合可能是必要的, 以便在多個磁盤不可用于檢查點和資料目錄時提供良好的性能.

Example for agent named a1:a1.channels = c1a1.channels.c1.type = filea1.channels.c1.checkpointDir = /mnt/flume/checkpointa1.channels.c1.dataDirs = /mnt/flume/data

加密

以下是一些示例配置:

使用密鑰存盤密碼分隔生成密碼密鑰:

keytool -genseckey -alias key-0 -keypass keyPassword -keyalg AES \-keysize 128 -validity 9000 -keystore test.keystore \-storetype jceks -storepass keyStorePassword

使用與密鑰庫密碼相同的密碼生成密鑰:

keytool -genseckey -alias key-1 -keyalg AES -keysize 128 -validity 9000 \-keystore src/test/resources/test.keystore -storetype jceks \-storepass keyStorePassworda1.channels.c1.encryption.activeKey = key-0a1.channels.c1.encryption.cipherProvider = AESCTRNOPADDINGa1.channels.c1.encryption.keyProvider = key-provider-0a1.channels.c1.encryption.keyProvider = JCEKSFILEa1.channels.c1.encryption.keyProvider.keyStoreFile = /path/to/my.keystorea1.channels.c1.encryption.keyProvider.keyStorePasswordFile = /path/to/my.keystore.passworda1.channels.c1.encryption.keyProvider.keys = key-0

false 設您已使用密鑰 0 輸出密鑰, 并且應使用密鑰 1 加密新檔案:

a1.channels.c1.encryption.activeKey = key-1a1.channels.c1.encryption.cipherProvider = AESCTRNOPADDINGa1.channels.c1.encryption.keyProvider = JCEKSFILEa1.channels.c1.encryption.keyProvider.keyStoreFile = /path/to/my.keystorea1.channels.c1.encryption.keyProvider.keyStorePasswordFile = /path/to/my.keystore.passworda1.channels.c1.encryption.keyProvider.keys = key-0 key-1

與上面相同的場景, 但 key-0 有自己的密碼:

a1.channels.c1.encryption.activeKey = key-1a1.channels.c1.encryption.cipherProvider = AESCTRNOPADDINGa1.channels.c1.encryption.keyProvider = JCEKSFILEa1.channels.c1.encryption.keyProvider.keyStoreFile = /path/to/my.keystorea1.channels.c1.encryption.keyProvider.keyStorePasswordFile = /path/to/my.keystore.passworda1.channels.c1.encryption.keyProvider.keys = key-0 key-1a1.channels.c1.encryption.keyProvider.keys.key-0.passwordFile = /path/to/key-0.passwordCustom Channel

自定義 channel 是您自己的 Channel 介面實作. 啟動 Flume agent 程式時, 必須在 agent 程式的類路徑中包含自定義 channel 的類及其依賴項. 自定義 channel 的 type 是其 FQCN. 必需屬性以粗體顯示

Example for agent named a1:a1.channels = c1a1.channels.c1.type = org.example.MyChannelFlume Channel Selectors

如果未指定 type, 則默認為 "replicating"

Replicating Channel Selector (default)

必需屬性以粗體顯示

Example for agent named a1 and it's source called r1:a1.sources = r1a1.channels = c1 c2 c3a1.sources.r1.selector.type = replicatinga1.sources.r1.channels = c1 c2 c3a1.sources.r1.selector.optional = c3

在上面的配置中, c3 是可選的 channel.

無法寫入 c3 只是被忽略了. 由于 c1 和 c2 未標記為可選, 因此無法寫入這些 channel 將導致事務失敗.

Multiplexing Channel Selector (多路復用)

必需屬性以粗體顯示

Example for agent named a1 and it's source called r1:a1.sources = r1a1.channels = c1 c2 c3 c4a1.sources.r1.selector.type = multiplexinga1.sources.r1.selector.header = statea1.sources.r1.selector.mapping.CZ = c1a1.sources.r1.selector.mapping.US = c2 c3a1.sources.r1.selector.default = c4Custom Channel Selector

自定義 channel 選擇器是您自己的 Channelelector 介面實作. 啟動 Flume agent 程式時, 自定義 channel 選擇器的類及其依賴項必須包含在 agent 程式的類路徑中. 自定義 channel 選擇器的 type 是其 FQCN.

Example for agent named a1 and its source called r1:a1.sources = r1a1.channels = c1a1.sources.r1.selector.type = org.example.MyChannelSelectorFlume Sink Processors

接收組允許用戶將多個 sink 分組到一個物體中. sink 處理器可用于在組內的所有 sink 上提供負載平衡功能, 或在時間故障的情況下實作從一個 sink 到另一個 sink 的故障轉移.

必需屬性以粗體顯示

Example for agent named a1:a1.sinkgroups = g1a1.sinkgroups.g1.sinks = k1 k2a1.sinkgroups.g1.processor.type = load_balanceDefault Sink Processor

默認 sink 只接受一個 sink. 用戶不必為單個 sink 創建處理器(sink 組). 相反, 用戶可以遵循本用戶指南中上面解釋的 source - channel - sink 模式

Failover Sink Processor

故障轉移 sink 維護一個優先級的 sink 串列, 保證只要有一個可用的 event 將被處理(傳遞)

故障轉移機制的作業原理是將故障 sink 降級到池中, 在池中為它們分配一個冷卻期, 在重試之前隨順序故障而增加. sink 成功發送 event 后, 它將恢復到實時池. sink 具有與之相關的優先級, 數量越大, 優先級越高. 如果在發送 event 時 sink 發生故障, 則接下來將嘗試下一個具有最高優先級的 sink 以發送 event . 例如, 在優先級為 80 的 sink 之前激活優先級為 100 的 sink. 如果未指定優先級, 則根據配置中指定 sink 的順序確定 Sinks 優先級.

要進行配置, 請將 sink 組處理器設定為故障轉移并為所有單個 sink 設定優先級. 所有指定的優先級必須是唯一的 此外, 可以使用 maxpenalty 屬性設定故障轉移時間的上限(以毫秒為單位)

必需屬性以粗體顯示

Example for agent named a1:a1.sinkgroups = g1a1.sinkgroups.g1.sinks = k1 k2a1.sinkgroups.g1.processor.type = failovera1.sinkgroups.g1.processor.priority.k1 = 5a1.sinkgroups.g1.processor.priority.k2 = 10a1.sinkgroups.g1.processor.maxpenalty = 10000Load balancing Sink Processor(負載均衡接收處理器)

負載平衡 sink 處理器提供了在多個 sink 上進行負載均衡流量的功能. 它維護一個索引的活動 sink 串列, 必須在其上分配負載. 實作支持使用 round_robin 或 random 機制分配負載. 選擇機制的選擇默認為 round_robin 型別, 但可以通過配置覆寫. 通過從 AbstractSinkSelector 繼承的自定義類支持自定義選擇機制.

呼叫時, 此選擇器使用其配置的選擇機制選擇下一個 sink 并呼叫它. 對于 round_robin 和 random 如果所選的 sink 無法傳遞 event , 則處理器通過其配置的選擇機制選擇下一個可用的 sink. 此實作不會將失敗的 sink 列入黑名單, 而是繼續樂觀地嘗試每個可用的 sink. 如果所有 sink 呼叫都導致失敗, 則選擇器將故障傳播到 sink 運行器.

如果啟用了 backoff , 則 sink 處理器會將失敗的 sink 列入黑名單, 將其洗掉以供給定超時的選擇. 當超時結束時, 如果 sink 仍然沒有回應, 則超時會以指數方式增加, 以避免在無回應的 sink 上長時間等待時卡住. 在禁用此功能的情況下, 在回圈中, 所有失敗的 sink 負載將被傳遞到下一個 sink, 因此不均衡

必需屬性以粗體顯示

Example for agent named a1:a1.sinkgroups = g1a1.sinkgroups.g1.sinks = k1 k2a1.sinkgroups.g1.processor.type = load_balancea1.sinkgroups.g1.processor.backoff = truea1.sinkgroups.g1.processor.selector = randomCustom Sink Processor

目前不支持自定義 sink 處理器

Event Serializers

該 file_roll sink 和 HDFS sink 都支持 EventSerializer 介面. 下面提供了隨 Flume 一起提供的 Event Serializer 的詳細資訊.

Body Text Serializer

別名: text . 此攔截器將 event 的主體寫入輸出流, 而不進行任何轉換或修改. event 標題將被忽略. 配置選項如下:

Example for agent named a1:a1.sinks = k1a1.sinks.k1.type = file_rolla1.sinks.k1.channel = c1a1.sinks.k1.sink.directory = /var/log/flumea1.sinks.k1.sink.serializer = texta1.sinks.k1.sink.serializer.appendNewline = false"Flume Event" Avro Event Serializer

Alias: avro_event.

此攔截器將 Flumeevent 序列化為 Avro 容器檔案. 使用的模式與 Avro RPC 機制中用于 Flumeevent 的模式相同.

此序列化程式繼承自 AbstractAvroEventSerializer 類

配置選項如下:

Example for agent named a1:a1.sinks.k1.type = hdfsa1.sinks.k1.channel = c1a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H%M/%Sa1.sinks.k1.serializer = avro_eventa1.sinks.k1.serializer.compressionCodec = snappy

Avro event 序列化器

別名: 此序列化程式沒有別名, 必須使用完全限定的類名類名指定.

這將 Flume event 序列化為 Avro 容器檔案, 如 "Flume Event" Avro Event Serializer, 但記錄模式是可配置的. 記錄模式可以指定為 Flume 配置屬性, 也可以在 event 頭中傳遞.

要將記錄模式作為 Flume 配置的一部分傳遞, 請使用下面列出的屬性 schemaURL.

要在 event 標頭中傳遞記錄模式, 請指定 包含模式的 JSON 格式表示的 event 標頭 flume.avro.schema.literal 或包含可以找到模式的 URL 的 flume.avro.schema.url( hdfs:/... URIs are supported I)

此序列化程式繼承自 AbstractAvroEventSerializer 類.

配置選項如下:

Example for agent named a1:a1.sinks.k1.type = hdfsa1.sinks.k1.channel = c1a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H%M/%Sa1.sinks.k1.serializer = org.apache.flume.sink.hdfs.AvroEventSerializer$Buildera1.sinks.k1.serializer.compressionCodec = snappya1.sinks.k1.serializer.schemaURL = hdfs://namenode/path/to/schema.avscFlume Interceptors

Flume 具有在運行中修改 / 洗掉 event 的能力. 這是在攔截器的幫助下完成的. 攔截器是實作 org.apache.flume.interceptor.Interceptor 的類介面. 攔截器可以根據攔截器開發者選擇的任何標準修改甚至洗掉 event .Flume 支持攔截器的鏈接. 通過在配置中指定攔截器構建器類名串列, 可以實作此目的. 攔截器在 source 配置中指定為以空格分隔的串列. 指定攔截器的順序是它們被呼叫的順序. 一個攔截器回傳的 event 串列將傳遞給鏈中的下一個攔截器. 攔截器可以修改或洗掉 event . 如果攔截器需要洗掉 event , 它就不會在它回傳的串列中回傳該 event . 如果要洗掉所有 event , 則只回傳一個空串列. 攔截器是命名組件, 下面是它們如何通過配置創建的示例:

a1.sources = r1a1.sinks = k1a1.channels = c1a1.sources.r1.interceptors = i1 i2a1.sources.r1.interceptors.i1.type = org.apache.flume.interceptor.HostInterceptor$Buildera1.sources.r1.interceptors.i1.preserveExisting = falsea1.sources.r1.interceptors.i1.hostHeader = hostnamea1.sources.r1.interceptors.i2.type = org.apache.flume.interceptor.TimestampInterceptor$Buildera1.sinks.k1.filePrefix = FlumeData.%{CollectorHost}.%Y-%m-%da1.sinks.k1.channel = c1

請注意, 攔截器構建器將傳遞給 type config 引數. 攔截器本身是可配置的, 可以傳遞配置值, 就像傳遞給任何其他可配置組件一樣. 在上面的示例中, event 首先傳遞給 HostInterceptor, 然后 HostInterceptor 回傳的 event 傳遞給 TimestampInterceptor. 您可以指定完全限定的類名 (FQCN) 或別名時間戳. 如果您有多個收集器寫入相同的 HDFS 路徑, 那么您也可以使用 HostInterceptor

時間戳攔截器 (Timestamp Interceptor)

此攔截器將 event 標頭插入到 event 標頭中, 以毫秒為單位處理 event . 此攔截器插入帶有鍵時間戳 (或由 header 屬性指定) 的標頭, 其值為相關時間戳. 如果已在配置中存在, 則此攔截器可以保留現有時間戳

Example for agent named a1:a1.sources = r1a1.channels = c1a1.sources.r1.channels = c1a1.sources.r1.type = seqa1.sources.r1.interceptors = i1a1.sources.r1.interceptors.i1.type = timestamp

主機攔截器(Host Interceptor)

此攔截器插入運行此 agent 程式的主機的 hostname 或 IP 地址. 它根據配置插入帶有密鑰主機或已配置密鑰的標頭, 其值為主機的 hostname 或 IP 地址.

Example for agent named a1:a1.sources = r1a1.channels = c1a1.sources.r1.interceptors = i1a1.sources.r1.interceptors.i1.type = host

靜態攔截器 (Static Interceptor)

靜態攔截器允許用戶將具有靜態值的靜態頭附加到所有 event

當前實作不允許一次指定多個標頭. 相反, 用戶可能會鏈接多個靜態攔截器, 每個靜態攔截器定義一個

Example for agent named a1:a1.sources = r1a1.channels = c1a1.sources.r1.channels = c1a1.sources.r1.type = seqa1.sources.r1.interceptors = i1a1.sources.r1.interceptors.i1.type = statica1.sources.r1.interceptors.i1.key = datacentera1.sources.r1.interceptors.i1.value = https://www.cnblogs.com/laoqing/p/NEW_YORKRemove Header Interceptor (洗掉標題攔截器)

此攔截器通過洗掉一個或多個 headers 來操縱 Flume event headers. 它可以洗掉靜態定義的 headers, 基于正則運算式的 headers 或串列中的 headers. 如果這些都沒有定義, 或者沒有符合條件的 headers, 則不會修改 Flume 事件.

請注意, 如果只需要洗掉一個 headers, 則按名稱指指定 header 可提供相對于其他兩種方法的性能優勢.

UUID 攔截器 (UUID Interceptor)

此攔截器在所有截獲的 event 上設定通用唯一識別符號. 示例 UUID 是 b5755073-77a9-43c1-8fad-b7a586fc1b97, 表示 128 位值.

如果 event 的應用程式級別唯一鍵不可用, 請考慮使用 UUIDInterceptor 自動為 event 分配 UUID. 一旦 UUID 進入 Flume 網路就將其分配給 event 非常重要; 也就是說, 在流量的第一個 Flume Source 中. 這使得在 Flume 網路中面對復制和重新發送時 event 的后續重復資料洗掉可以實作高可用性和高性能. 如果應用程式級別密鑰可用, 則優于自動生成的 UUID, 因為它使用所述公知的應用程式級別密鑰在資料存盤中啟用后續更新和 event 洗掉.

Morphline Interceptor (形態攔截器)

此攔截器通過 morphline 組態檔過濾 event , 該檔案定義了一系列轉換命令, 用于將記錄從一個命令傳遞到另一個命令. 例如, morphline 可以忽略某些 event 或通過基于正則運算式的模式匹配來更改或插入某些 event 頭, 或者它可以通過 Apache Tika 在截獲的 event 上自動檢測和設定 MIMEtype. 例如, 這種資料包嗅探可用于 Flume 拓撲中基于內容的動態路由. MorphlineInterceptor 還可以幫助實作到多個 Apache Solr 集合的動態路由(例如, 用于多租戶)

目前, 存在一個限制, 即攔截器的形態線不能為每個輸入 event 生成多個輸出記錄. 此攔截器不適用于重型 ETL 處理 - 如果需要, 請考慮將 ETL 處理從 Flume Source 移至 Flume Sink, 例如移至 MorphlineSolrSink.

必需屬性以粗體顯示

Sample flume.conf file:a1.sources.avroSrc.interceptors = morphlineinterceptora1.sources.avroSrc.interceptors.morphlineinterceptor.type = org.apache.flume.sink.Solr.morphline.MorphlineInterceptor$Buildera1.sources.avroSrc.interceptors.morphlineinterceptor.morphlineFile = /etc/flume-ng/conf/morphline.confa1.sources.avroSrc.interceptors.morphlineinterceptor.morphlineId = morphline1

搜索和替換攔截器 (Search and Replace Interceptor)

此攔截器基于 Java 正則運算式提供簡單的基于字串的搜索和替換功能. 還可以進行回溯 / 群組捕捉. 此攔截器使用與 Java Matcher.replaceAll()方法中相同的規則

Example configuration:a1.sources.avroSrc.interceptors = search-replacea1.sources.avroSrc.interceptors.search-replace.type = search_replace# Remove leading alphanumeric characters in an event body.a1.sources.avroSrc.interceptors.search-replace.searchPattern = ^[A-Za-z0-9_]+a1.sources.avroSrc.interceptors.search-replace.replaceString =Another example:a1.sources.avroSrc.interceptors = search-replacea1.sources.avroSrc.interceptors.search-replace.type = search_replace# Use grouping operators to reorder and munge words on a line.a1.sources.avroSrc.interceptors.search-replace.searchPattern = The quick brown ([a-z]+) jumped over the lazy ([a-z]+)a1.sources.avroSrc.interceptors.search-replace.replaceString = The hungry $2 ate the careless $1

正則運算式過濾攔截器 (Regex Filtering Interceptor)

此攔截器通過將 event 主體解釋為文本并將文本與配置的正則運算式進行匹配來有選擇地過濾 event . 提供的正則運算式可用于包括 event 或排除 event

正則運算式提取器攔截器(Regex Extractor Interceptor)

此攔截器使用指定的正則運算式提取正則運算式匹配組, 并將匹配組作為標題附加到 event 上. 它還支持可插入序列化程式, 用于在將匹配組添加為 event 標頭之前對其進行格式化.

序列化器用于將匹配映射到標題名稱和格式化標題值; 默認情況下, 您只需指定標題名稱, 將使用默認的 org.apache.flume.interceptor.RegexExtractorInterceptorPassThroughSerializer. 此序列化程式只是將匹配映射到指定的標題名稱, 并在正則運算式提取時傳遞值. 您可以使用完全限定的類名 (FQCN) 將自定義序列化程式實作插入到提取器中, 以便以您喜歡的方式格式化匹配.

例 1:

如果 Flumeevent 主體包含 1:2:3.4foobar5 , 則使用以下配置

a1.sources.r1.interceptors.i1.regex = (\\d):(\\d):(\\d)a1.sources.r1.interceptors.i1.serializers = s1 s2 s3a1.sources.r1.interceptors.i1.serializers.s1.name = onea1.sources.r1.interceptors.i1.serializers.s2.name = twoa1.sources.r1.interceptors.i1.serializers.s3.name = three

提取的 event 將包含相同的主體, 但后面的標題將添加一個 => 1, 兩個 => 2, 三個 => 3

例 2:

如果 Flumeevent 正文包含 2012-10-18 18:47:57,614 某些 日志 行, 則使用以下配置

a1.sources.r1.interceptors.i1.regex = ^(?:\\n)?(\\d\\d\\d\\d-\\d\\d-\\d\\d\\s\\d\\d:\\d\\d)a1.sources.r1.interceptors.i1.serializers = s1a1.sources.r1.interceptors.i1.serializers.s1.type = org.apache.flume.interceptor.RegexExtractorInterceptorMillisSerializera1.sources.r1.interceptors.i1.serializers.s1.name = timestampa1.sources.r1.interceptors.i1.serializers.s1.pattern = yyyy-MM-dd HH:mm

提取的 event 將包含相同的主體, 但后面的標題將添加 timestamp => 1350611220000

Flume 屬性

屬性: flume.called.from.service

Flume 每隔 30 秒定期輪詢一次指定組態檔的更改. 如果首次輪詢現有檔案, 或者自上次輪詢以來現有檔案的修改日期發生更改, Flume agent 將從組態檔加載新配置. 重命名或移動檔案不會更改其修改時間. 當 Flume agent 輪詢一個不存在的檔案時, 會發生以下兩種情況之一:

當 agent 首次輪詢不存在的組態檔時, agent 將根據 flume.called.from.service 屬性執行操作. 如果設定了屬性, 則 agent 將繼續輪詢(始終在同一時間 - 每 30 秒). 如果未設定該屬性, 則 agent 會立即終止

當 agent 輪詢一個不存在的組態檔并且這不是第一次輪詢檔案時, agent 不會對此輪詢周期進行配置更改. agent 繼續輪詢而不是終止.

配置過濾器

Flume 提供了一種工具, 用于以配置過濾器的形式將敏感或生成的資料注入配置. 配置密鑰可以設定為配置屬性的值, 它將由配置過濾器替換為它所代表的值

配置過濾器的常見用法

格式類似于 Java 運算式語言, 但它目前不是一個完全有效的 EL 運算式決議器, 只是一種看起來像它的格式

<agent_name>.configfilters =<filter_name><agent_name>.configfilters.<filter_name>.type =<filter_type><agent_name>.sources.<source_name>.parameter = ${<filter_name>['<key_for_sensitive_or_generated_data>']}<agent_name>.sinks.<sink_name>.parameter = ${<filter_name>['<key_for_sensitive_or_generated_data>']}<agent_name>.<component_type>.<component_name>.parameter = ${<filter_name>['<key_for_sensitive_or_generated_data>']} #or<agent_name>.<component_type>.<component_name>.parameter = ${<filter_name>["<key_for_sensitive_or_generated_data>"]} #or<agent_name>.<component_type>.<component_name>.parameter = ${<filter_name>[<key_for_sensitive_or_generated_data>]} #or<agent_name>.<component_type>.<component_name>.parameter = some_constant_data${<filter_name>[<key_for_sensitive_or_generated_data>]}

環境變數配置過濾器

示例, 要在配置中隱藏密碼, 請設定其值, 如以下示例所示

a1.sources = r1a1.channels = c1a1.configfilters = f1a1.configfilters.f1.type = enva1.sources.r1.channels = c1a1.sources.r1.type = httpa1.sources.r1.keystorePassword = ${f1['my_keystore_password']}

這里 a1.source.r1.keystorePassword 配置屬性將獲取 my_keystore_password 環境變數的值. 設定環境變數的一種方法是運行如下所示的 flume agent:

$ my_keystore_password = Secret123 bin / flume -ng agent --conf conf --conf-file example.conf ...

外部處理配置過濾器(External Process Config Filter)

示例, 要在配置中隱藏密碼, 請設定其值, 如以下示例所示:

a1.sources = r1a1.channels = c1a1.configfilters = f1a1.configfilters.f1.type = externala1.configfilters.f1.command = /usr/bin/passwordResolver.sha1.configfilters.f1.charset = UTF-8a1.sources.r1.channels = c1a1.sources.r1.type = httpa1.sources.r1.keystorePassword = ${f1['my_keystore_password']}

在此示例中, flume 將運行以下命令以獲取值

$ /usr/bin/passwordResolver.sh my_keystore_password

該 passwordResolver.sh 將回傳 Secret123 與退出代碼 0.

示例, 要生成滾動檔案 sink 的目錄的一部分, 請設定其值, 如以下示例所示:

a1.sources = r1a1.channels = c1a1.configfilters = f1a1.configfilters.f1.type = externala1.configfilters.f1.command = /usr/bin/generateUniqId.sha1.configfilters.f1.charset = UTF-8a1.sinks = k1a1.sinks.k1.type = file_rolla1.sinks.k1.channel = c1a1.sinks.k1.sink.directory = /var/log/flume/agent_${f1['agent_name']}

在此示例中, flume 將運行以下命令以獲取值

$ /usr/bin/generateUniqId.sh agent_name

該 generateUniqId.sh 將回傳 1234 與退出代碼 0.

Hadoop 憑據存盤配置過濾器

此功能的類路徑上需要一個 hadoop-common 庫(2.6 + 版本). 如果安裝了 hadoop, 則 agent 會自動將其添加到類路徑中

示例, 要在配置中隱藏密碼, 請設定其值, 如以下示例所示:

a1.sources = r1a1.channels = c1a1.configfilters = f1a1.configfilters.f1.type = hadoopa1.configfilters.f1.credential.provider.path = jceks://file/<path_to_jceks file>a1.sources.r1.channels = c1a1.sources.r1.type = httpa1.sources.r1.keystorePassword = ${f1['my_keystore_password']}

Log4j 追加 (Log4J Appender)

將 Log4jevent 附加到 flume agent 的 avrosource. 使用此 appender 的客戶端必須在類路徑中包含 flume-ng-sdk(例如, flume-ng-sdk-1.9.0.jar). 必需屬性以粗體顯示

示例 log4j.properties 檔案:

#...log4j.appender.flume = org.apache.flume.clients.log4jappender.Log4jAppenderlog4j.appender.flume.Hostname = example.comlog4j.appender.flume.Port = 41414log4j.appender.flume.UnsafeMode = true# configure a class's logger to output to the flume appenderlog4j.logger.org.example.MyClass = DEBUG,flume#...

默認情況下, 通過呼叫 toString()或使用 Log4j 布局 (如果指定) 將每個 event 轉換為字串.

如果 event 是 org.apache.avro.generic.GenericRecord,org.apache.avro.specific.SpecificRecord 的實體 , 或者屬性 AvroReflectionEnabled 設定為 true, 則將使用 Avro 序列化序列化 event .

使用 Avro 架構序列化每個 event 效率很低, 因此最好提供一個架構 URL, 下游 sink(通常是 HDFSsink)可以從該架構 URL 檢索架構.如果未指定 AvroSchemaUrl, 則架構將作為 Flume 標頭包含在內.

示例 log4j.properties 檔案配置為使用 Avro 序列化:

#...log4j.appender.flume = org.apache.flume.clients.log4jappender.Log4jAppenderlog4j.appender.flume.Hostname = example.comlog4j.appender.flume.Port = 41414log4j.appender.flume.AvroReflectionEnabled = truelog4j.appender.flume.AvroSchemaUrl = hdfs://namenode/path/to/schema.avsc# configure a class's logger to output to the flume appenderlog4j.logger.org.example.MyClass = DEBUG,flume#...

負載均衡 Log4j 追加 (Load Balancing Log4J Appender)

將 Log4jevent 追加到 flume agent 的 avrosource 串列中. 使用此 appender 的客戶端必須在類路徑中包含 flume-ng-sdk(例如, flume-ng-sdk-1.9.0.jar). 該 appender 支持回圈和隨機方案, 用于執行負載平衡. 它還支持可配置的退避超時, 以便從主機集臨時洗掉向下 agent . 必需屬性以粗體顯示

使用默認值配置的示例 log4j.properties 檔案:

#...log4j.appender.out2 = org.apache.flume.clients.log4jappender.LoadBalancingLog4jAppenderlog4j.appender.out2.Hosts = localhost:25430 localhost:25431# configure a class's logger to output to the flume appenderlog4j.logger.org.example.MyClass = DEBUG,flume#...

使用 RANDOM 負載平衡配置的示例 log4j.properties 檔案:

#...log4j.appender.out2 = org.apache.flume.clients.log4jappender.LoadBalancingLog4jAppenderlog4j.appender.out2.Hosts = localhost:25430 localhost:25431log4j.appender.out2.Selector = RANDOM# configure a class's logger to output to the flume appenderlog4j.logger.org.example.MyClass = DEBUG,flume#...

使用退避配置的示例 log4j.properties 檔案:

#...log4j.appender.out2 = org.apache.flume.clients.log4jappender.LoadBalancingLog4jAppenderlog4j.appender.out2.Hosts = localhost:25430 localhost:25431 localhost:25432log4j.appender.out2.Selector = ROUND_ROBINlog4j.appender.out2.MaxBackoff = 30000# configure a class's logger to output to the flume appenderlog4j.logger.org.example.MyClass = DEBUG,flume#...

安全

HDFSsink,HBasesink,Thriftsource,Thriftsink 和 Kite Datasetsink 都支持 Kerberos 身份驗證. 請參閱相應的部分以配置與 Kerberos 相關的選項.

Flume agent 將作為單個主體對 kerberos KDC 進行身份驗證, 這將由需要進行 kerberos 身份驗證的不同組件使用. 為 Thriftsource,Thriftsink,HDFSsink,HBasesink 和 DataSetsink 配置的主體和 keytab 應該相同, 否則組件將無法啟動.

監控

Flume 中的監控仍在進行中. 變化可能經常發生. 幾個 Flume 組件向 JMX 平臺 MBean 服務器報告度量標準. 可以使用 Jconsole 查詢這些指標.

可用的組件指標

下表顯示了可用于組件的度量標準. 每個組件僅維護一組度量, 由 "x" 表示, 未維護的值顯示默認值, 即 0. 這些表告訴您可以在何處獲得有意義的資料. 度量標準的名稱應該足夠描述, 有關更多資訊, 您必須深入了解組件的 source 代碼.

Sources1

Sources 2

Sinks 1

Sinks 2

Channels

JMX 報告

可以通過使用 flume-env.sh 在 JAVA_OPTS 環境變數中指定 JMX 引數來啟用 JMX 報告, 如

export JAVA_OPTS ="- Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port = 5445 -Dcom.sun.management.jmxremote.authenticate = false -Dcom.sun.management.jmxremote.ssl = false"

注意: 上面的示例禁用安全性. 要啟用安全性, 請參閱

Ganglia 報告

Flume 還可以將這些指標報告給 Ganglia 3 或 Ganglia 3.1 元節點. 要向 Ganglia 報告指標, 必須使用此支持啟動水槽 agent . 必須通過傳遞以下引數來啟動 Flume agent , 作為以 flume.monitoring 為前綴的系統屬性., 可以在 flume-env.sh 中指定:

我們可以通過 Ganglia 支持啟動 Flume, 如下所示:

$ bin/flume-ng agent --conf-file example.conf --name a1 -Dflume.monitoring.type=ganglia -Dflume.monitoring.hosts=com.example:1234,com.example2:5455

JSON 報告

Flume 還可以以 JSON 格式報告指標. 為了以 JSON 格式啟用報告, Flume 在可配置埠上托管 Web 服務器. Flume 以以下 JSON 格式報告指標:

{"typeName1.componentName1" : {"metric1" : "metricValue1", "metric2" : "metricValue2"},"typeName2.componentName2" : {"metric3" : "metricValue3", "metric4" : "metricValue4"}}

這是一個例子:

{"CHANNEL.fileChannel":{"EventPutSuccessCount":"468085","Type":"CHANNEL","StopTime":"0","EventPutAttemptCount":"468086","ChannelSize":"233428","StartTime":"1344882233070","EventTakeSuccessCount":"458200","ChannelCapacity":"600000","EventTakeAttemptCount":"458288"},"CHANNEL.memChannel":{"EventPutSuccessCount":"22948908","Type":"CHANNEL","StopTime":"0","EventPutAttemptCount":"22948908","ChannelSize":"5","StartTime":"1344882209413","EventTakeSuccessCount":"22948900","ChannelCapacity":"100","EventTakeAttemptCount":"22948908"}}

$ bin/flume-ng agent --conf-file example.conf --name a1 -Dflume.monitoring.type=http -Dflume.monitoring.port=34545

然后, 可以在 http://<hostname>:<port>/metrics 網頁上獲得度量標準. 自定義組件可以報告上面 Ganglia 部分中提到的指標.

自定義報告

可以通過撰寫執行報告的服務器向其他系統報告指標. 任何報告類都必須實作 org.apache.flume.instrumentation.MonitorService 介面 . 這樣的類可以與 GangliaServer 用于報告的方式相同. 他們可以輪詢平臺 mbean 服務器以輪詢 mbeans 以獲取指標. 例如, 如果可以使用名為 HTTPReporting 的 HTTP 監視服務, 如下所示:

$ bin/flume-ng agent --conf-file example.conf --name a1 -Dflume.monitoring.type=com.example.reporting.HTTPReporting -Dflume.monitoring.node=com.example:332

從自定義組件報告指標

任何自定義 flume 組件都應繼承自 org.apache.flume.instrumentation.MonitoredCounterGroup 類. 然后, 該類應為其公開的每個度量標準提供 getter 方法. 請參閱下面的代碼. MonitoredCounterGroup 需要一個屬性串列, 其度量由此類公開. 截至目前, 此類僅支持將度量標準公開為長值.

public class SinkCounter extends MonitoredCounterGroup implementsSinkCounterMBean {private static final String COUNTER_CONNECTION_CREATED ="sink.connection.creation.count";private static final String COUNTER_CONNECTION_CLOSED ="sink.connection.closed.count";private static final String COUNTER_CONNECTION_FAILED ="sink.connection.failed.count";private static final String COUNTER_BATCH_EMPTY ="sink.batch.empty";private static final String COUNTER_BATCH_UNDERFLOW ="sink.batch.underflow";private static final String COUNTER_BATCH_COMPLETE ="sink.batch.complete";private static final String COUNTER_EVENT_DRAIN_ATTEMPT ="sink.event.drain.attempt";private static final String COUNTER_EVENT_DRAIN_SUCCESS ="sink.event.drain.sucess";private static final String[] ATTRIBUTES = {COUNTER_CONNECTION_CREATED, COUNTER_CONNECTION_CLOSED,COUNTER_CONNECTION_FAILED, COUNTER_BATCH_EMPTY,COUNTER_BATCH_UNDERFLOW, COUNTER_BATCH_COMPLETE,COUNTER_EVENT_DRAIN_ATTEMPT, COUNTER_EVENT_DRAIN_SUCCESS};public SinkCounter(String name) {super(MonitoredCounterGroup.Type.SINK, name, ATTRIBUTES);}@Overridepublic long getConnectionCreatedCount() {return get(COUNTER_CONNECTION_CREATED);}public long incrementConnectionCreatedCount() {return increment(COUNTER_CONNECTION_CREATED);}}

工具

檔案 channel 完整性工具

檔案 channel 完整性工具可驗證檔案 channel 中各個 event 的完整性, 并洗掉損壞的 event

這些工具可以按如下方式運行:

$bin/flume-ng tool --conf ./conf FCINTEGRITYTOOL -l ./datadir -e org.apache.flume.MyEventValidator -DmaxSize 2000

其中 datadir 是要驗證的資料目錄的逗號分隔串列.

以下是可用選項

event 驗證工具

event 驗證器工具可用于以特定于應用程式的方式驗證檔案 channelevent . 該工具在每個 event 上應用用戶提供程式驗證登錄, 并洗掉未向邏輯確認的 event

這些工具可以按如下方式運行:

$bin/flume-ng tool --conf ./conf FCINTEGRITYTOOL -l ./datadir -e org.apache.flume.MyEventValidator -DmaxSize 2000

其中 datadir 是要驗證的資料目錄的逗號分隔串列.

以下是可用選項

Event 驗證器實作必須實作 EventValidator 介面. 建議不要從實作中拋出任何例外, 因為它們被視為無效 event . 其他引數可以通過 - D 選項傳遞給 EventValitor 實作.

讓我們看一個基于簡單大小的 event 驗證器的示例, 它將拒絕大于指定的最大大小的 event

public static class MyEventValidator implements EventValidator {private int value = https://www.cnblogs.com/laoqing/p/0;private MyEventValidator(int val) {value = val;}@Overridepublic boolean validateEvent(Event event) {return event.getBody() <= value;}public static class Builder implements EventValidator.Builder {private int sizeValidator = 0;@Overridepublic EventValidator build() {return new DummyEventVerifier(sizeValidator);}@Overridepublic void configure(Context context) {binaryValidator = context.getInteger("maxSize");}}}

拓撲設計注意事項

Flume 非常靈活, 允許大量可能的部署方案. 如果您計劃在大型生產部署中使用 Flume, 那么花一些時間考慮如何根據 Flume 拓撲來表達您的問題是明智的. 本節介紹一些注意事項

Flume 是否適合您的問題?

如果您需要將文本日志資料提取到 Hadoop / HDFS 中, 那么 Flume 最適合您的問題, 完全停止. 對于其他用例, 以下是一些指導原則:

Flume 旨在通過相對穩定, 可能復雜的拓撲來傳輸和攝取定期生成的 event 資料."event 資料" 的概念定義非常廣泛. 對 Flume 來說, event 只是一個普通的位元組. event 的大小有一些限制 - 例如, 它不能大于您可以存盤在記憶體中或單個機器上的磁盤上 - 但實際上, 水槽 event 可以是從文本日志條目到影像檔案的所有內容. event 的關鍵屬性是它們以連續的流式方式生成. 如果您的資料沒有定期生成(即您嘗試將大量資料批量加載到 Hadoop 集群中), 那么 Flume 仍然可以正常作業, 但對您的情況來說可能有點過頭了. Flume 喜歡相對穩定的拓撲結構. 您的拓撲不需要是不可變的, 因為 Flume 可以處理拓撲中的更改而不會丟失資料, 并且還可以容忍由于故障轉移或配置而定期重新配置. 如果您每天都要更改拓撲結構, 那么它可能無法正常作業, 因為重新配置需要一些思考和開銷.

Flume 中的流量可靠性

Flume 流的可靠性取決于幾個因素. 通過調整這些因素, 您可以使用 Flume 實作各種可靠性選項.

您使用什么 type 的 channel .Flume 具有持久的 channel(將資料保存到磁盤的 channel)和非持久 channel(如果機器出現故障將丟失資料). 持久 channel 使用基于磁盤的存盤, 存盤在此類 channel 中的資料將在機器重啟或非磁盤相關故障中持續存在.

您的 channel 是否已為作業量充分配置. Flume 中的 channel 充當各種跳躍的緩沖區. 這些緩沖區具有固定容量, 一旦該容量已滿, 您將在流中的早期點創建背壓. 如果此壓力傳播到流量 source,Flume 將變得不可用并可能丟失資料.

您是否使用冗余拓撲. Flume 讓你在冗余拓撲中復制流. 這可以提供非常容易的容錯 source, 并且可以克服磁盤或機器故障.

考慮 Flume 拓撲結構中可靠性的最佳方法是考慮各種故障情況及其結果. 如果磁盤發生故障會怎么樣? 如果機器出現故障會怎樣? 如果您的終端 sink(例如 HDFS)停機一段時間并且您有背壓, 會發生什么? 可能的設計空間巨大, 但您需要提出的基本問題只是極少數.

Flow 拓撲設計

設計 Flume 拓撲的第一步是列舉資料的所有 source 和目標(終端 sink). 這些將定義拓撲的邊緣點. 下一個考慮因素是是否引入中間聚合層或 event 路由. 如果要從大量 source 中收集資料, 則聚合資料以簡化終端 sink 的提取可能會有所幫助. 聚合層還可以通過充當緩沖區來消除 source 的突發性或 sink 的不可用性. 如果要在不同位置之間路由資料, 您可能還希望在不同點分割流: 這會創建可能本身包含聚合點的子拓撲.

調整 Flume 部署的大小

一旦了解了拓撲的外觀, 下一個問題就是需要多少硬體和網路容量. 首先, 量化您生成的資料量. 這并不總是一項簡單的任務! 大多數資料流是突發性的 (例如, 由于晝夜模式) 并且可能是不可預測的. 一個好的起點是考慮每個拓撲層中的最大吞吐量, 包括每秒 event 數和每秒位元組數. 一旦知道給定層的所需吞吐量, 就可以計算該層所需節點數的下限. 要確定可達到的吞吐量, 最好使用合成或采樣 event 資料在硬體上試驗 Flume. 通常, 基于磁盤的 channel 應該獲得 10 的 MB / s, 而基于記憶體的 channel 應該達到 100 的 MB / s 或更多. 但性能會有很大差異, 具體取決于硬體和操作環境.

調整聚合吞吐量可以為每層提供所需節點數量的下限. 有幾個原因需要額外的節點, 例如增加冗余和更好地吸收負載中的突發.

故障排除

處理 agent 失敗

如果 Flume agent 程式關閉, 則該 agent 程式上托管的所有流程都將中止. 重新啟動 agent 后, 將恢復流程. 使用檔案 channel 或其他穩定 channel 的流將從中斷處繼續處理 event . 如果無法在同一硬體上重新啟動 agent , 則可以選擇將資料庫遷移到另一個硬體并設定新的 Flume agent , 該 agent 可以繼續處理 db 中保存的 event . 可以利用資料庫 HA 期貨將 Flume agent 移動到另一個主機.

兼容性

HDFS

目前 Flume 支持 HDFS 0.20.2 和 0.23.

組件摘要

別名約定

這些別名的約定在上面的組件特定示例中使用, 以使所有示例中的名稱保持簡短和一致.

標籤：大數據

上一篇：PB9 連接2008/2012 問題

下一篇：關于呼叫AVICAP32錄像問題.