👨🏽 ✅ 🛌🏼 Flume-データストリームを管理します。パート2 🎽 🛐 📿

こんにちは、Habr！ Apache Flumeシリーズの記事を続けます。前のパートでは、このツールを表面的に調べ、構成および実行する方法を見つけました。今回は、Flumeの主要なコンポーネントについて説明しますが、実際のデータを操作するのは怖くありません。

ファイルチャンネル

前回の記事では、メモリーチャンネルについて説明しました。明らかに、メモリを使用してデータを保存するチャネルは信頼できません。ノードを再起動すると、チャネルに保存されているすべてのデータが失われます。これにより、メモリチャネルが役に立たなくなることはありません。その速度のために使用が非常に正当化される場合があります。ただし、真に信頼できる輸送システムには、より堅牢なソリューションが必要です。

このソリューションは、ファイルチャネル-ファイルチャネルです。このチャンネルがデータをファイルに保存していると推測するのは簡単です。同時に、チャネルはランダムアクセスを使用してファイルを操作するため、イベントのシーケンスを維持しながらイベントを追加および収集できます。高速ナビゲーションのために、チャネルはラベル（チェックポイント）のシステムを使用し、WALメカニズムが実装されています。これはすべて、一般に、チャネルの「内部」に隠されており、次のパラメーターを使用して構成されます（太字-必須パラメーター）。

パラメータ	説明	デフォルトで
タイプ	チャネル実装、ファイルを指定する必要があります	-
checkpointDir	タグ付きのファイルを保存するためのフォルダー。指定しない場合、チャネルはFlumeホームフォルダーを使用します。	$ HOME / ...
useDualCheckpoints	タグ付きのバックアップフォルダを作成します。	偽
backupCheckpointDir	ラベル付きファイルのバックアップ用のフォルダー。useDualCheckpoints= trueの場合はもちろん指定する必要があります（もちろん、このバックアップは元のディスク（たとえば、別のディスク）から遠ざける必要があります）。	-
dataDirs	データファイルが配置されるコンマで区切られたフォルダーのリスト。パフォーマンスを向上させるには、異なるドライブに複数のフォルダーを指定することをお勧めします。フォルダーが指定されていない場合、チャネルはFlumeホームフォルダーも使用します。	$ HOME / ...
能力	チャネル容量。イベントの数を示します。	1,000,000
transactionCapacity	1つのトランザクションのイベントの最大数。トランスポートシステム全体のパフォーマンスが依存する非常に重要なパラメーター。詳細については、以下に説明します。	10,000
checkpointInterval	新しいタグの作成間隔（ミリ秒単位）。タグは再起動時に重要な役割を果たし、チャネル状態を復元しながらデータファイルのセクションを「ジャンプオーバー」できます。その結果、チャネルはデータファイル全体を再読み取りしないため、チャネルが「詰まっている」場合の起動が大幅に高速化されます。	30000
checkpointOnClose	チャンネルを閉じるときにラベルを記録します。トレーリングラベルを使用すると、再起動時にチャネルをできるだけ早く回復できますが、チャネルが閉じられると作成に時間がかかります（実際にはほとんどありません）。	本当
キープアライブ	チャネルへの追加操作のタイムアウト（秒単位）。つまり、チャネルが詰まっている場合、トランザクションはしばらく待ってから「チャンスを与えます」。また、チャネルに空きスペースがない場合、トランザクションはロールバックされます。	3
maxFileSize	チャネルファイルの最大サイズ（バイト単位）。このパラメーターの値は、チャネルが「バイト」できるスペースを決定するものではありません。1つのデータファイルのサイズを設定し、チャネルはこれらのファイルのいくつかを作成できます。	2146435071（2GB）
minimumRequiredSpace	ディスクの空き領域がこのパラメーターで指定されているよりも少ない場合、チャネルは新しいイベントを受信しません。データフォルダーが複数のディスクにある場合、Flumeは	524288000（500MB）

残りの設定は、チャネルファイルのデータ暗号化とリカバリプロセス（リプレイ）に関連しています。ここで、ファイルチャネルを操作する際に考慮すべき事項についていくつか説明します。

Flumeがデータをフォルダーに書き込む権限を持っていることを確認してください 。

または、より正確には、Flumeに代わって起動されるユーザーは、 チェックポイントとデータのフォルダーへの書き込み権限を持っています。

SSDはチャネルを大幅に高速化します 。

以下のグラフは、ファイルチャネルを使用して500イベントのパケットをFlumeノードに送信するのにかかった時間を示しています。ノードの1つはSSDを使用してチャネルデータを保存し、もう1つはSATAを使用します。違いは大きい。

単純な分割を実行すると、SSD上のファイルチャネルを持つFlumeノードは、1秒あたり最大500 / 0.025 = 20,000イベントをダイジェストできます（参照用に、この例のメッセージサイズは約1KBであり、チャネルはストレージに1つのディスクのみを使用します）。

チャネル容量は、変化に非常に敏感です。

突然チャンネルの容量を変更することに決めた場合、不愉快な驚きを感じるかもしれません-データを復元するためにチャンネルがリプレイを開始します。これは、チェックポイントファイルを使用してさらにナビゲーション/データを操作する代わりに、チャネルがすべてのデータファイルを完全に実行することを意味します。チャネルに大量のデータがある場合、プロセスにはかなりの時間がかかる可能性があります。

チャネルの異常停止は、データ損失につながる可能性があります 。

これは、Flumeプロセスを強制終了（またはハードリセット）した場合に発生する可能性があります。または、発生しない場合があります。私の記憶では、これは私たちに一度だけ起こりました-データファイルは「破損」し、チャンネルデータを含むすべてのファイルを手動で削除する必要がありました（残念ながら、チャンネルは詰まらず、損失は回避されました）。したがって、チャネルはまだ100％の信頼性を提供しません-誰かが「スイッチを引っ張る」可能性が常にあり、修復不可能なことが起こります。さて、これが発生し、チャンネルが開始を拒否した場合、あなたの行動は次のようになるでしょう：
1. チェックポイントを削除してみてください—この場合、チャネルはデータファイルからのみ回復を試みます。
2. 前の段落が役に立たず、チャネルが「チャネルからデータを読み取れません、チャネルが閉じられます」というスタイルで何かを書き込む場合、データファイルは破損しています。ここでは、すべてのデータフォルダーの完全なクリーニングのみが役立ちます。ああ。

File-Channelの代わりとして、Flumeはさらにいくつかのチャネルを提供します。特に、データベースをバッファとして使用するJDBC-channelとKafka-channelを提供します。もちろん、このようなチャネルを使用するには、データベースとKafkaを別々にデプロイする必要があります。

Avro SourceおよびAvro Sink

Avroはデータシリアル化ツールの 1つであり、そのおかげでソースとストックに名前が付けられました。これらのコンポーネントのネットワークは、Nettyを使用して実装されます。前の記事で説明したNetcat Sourceと比較して、Avro Sourceには次の利点があります。

イベントでヘッダーを使用できます（つまり、データとともにサポート情報を送信します）。
複数のクライアントから同時に情報を受信できます。 Netcatは通常のソケットで実行され、一連の着信接続を受け入れます。つまり、一度に1つのクライアントしか処理できません。

それでは、Avro Sourceが提供する設定を見てみましょう。

パラメータ	説明	デフォルトで
タイプ	ソース実装、 avroを指定する必要があります。	-
チャンネル	ソースがイベントを送信するチャネル（スペースで区切られます）。	-
縛る	ソースを修正するホスト/ IP。	-
港	ソースがクライアントからの接続を受け入れるポート。	-
スレッド	着信イベント（I / Oワーカー）を処理するスレッドの数。値を選択するときは、このソースにイベントを送信する潜在的な顧客の数をガイドする必要があります。少なくとも2つのスレッドを設定する必要があります。そうしないと、ソースが1つしかない場合でも、ソースが単に「ハング」する可能性があります。必要なスレッド数が不明な場合は、構成でこのパラメーターを指定しないでください。	限定されない
圧縮タイプ	データ圧縮、ここにはいくつかのオプションがあります-noneまたはdeflate 。クライアントがデータを圧縮形式で送信する場合にのみ指定する必要があります。圧縮によりトラフィックを大幅に節約でき、一度に送信するイベントが多いほど、大幅に節約できます。	なし

他のソースと同様に、Avro Sourceに次を指定できます。

selector.typeは、前の記事で言及したチャンネルセレクターです。いくつかのルールに従って、複数のチャネルでイベントを分割または複製できます。セレクターについては、以下で詳しく説明します。
インターセプター -スペースで区切られたインターセプターのリスト。イベントがチャネルに入る前にインターセプターが起動します。これらは、何らかの方法でイベントを変更するために使用されます（たとえば、ヘッダーを追加したり、イベントの内容を変更したりする）。それらについても以下で説明します。

また、このソースには、Nettyフィルターとデータ暗号化設定が用意されています。このコードを使用して、このソースにイベントを送信できます。

Avro SourceのプリミティブJavaクライアント

import java.util.HashMap; import java.util.Map; import org.apache.flume.Event; import org.apache.flume.EventDeliveryException; import org.apache.flume.api.RpcClient; import org.apache.flume.api.RpcClientFactory; import org.apache.flume.event.EventBuilder; import org.apache.flume.event.SimpleEvent; public class FlumeSender { public static void main(String[] args) throws EventDeliveryException { RpcClient avroClient = RpcClientFactory.getDefaultInstance("127.0.0.1", 50001); Map<String, String> headers = new HashMap<>(); headers.put("type", "common"); Event event = EventBuilder.withBody(" ".getBytes(), headers); avroClient.append(event); avroClient.close(); } }

次に、Avro-drainの構成を検討します。

パラメータ	説明	デフォルトで
タイプ	流出の実装はavroで示される必要があります。	-
チャンネル	株式がイベントを描画するチャンネル。	-
ホスト名	株式がイベントを送信するホスト/ IP。	-
港	指定されたマシン（ hostname ）がクライアントの接続を待機しているポート。	-
バッチサイズ	非常に重要なパラメーター：1回の要求でクライアントに送信されるイベントの「パケット」のサイズ。同時に、同じ値を使用してチャネルを空にします。つまり、これは、1つのトランザクションでチャネルから読み取られたイベントの数でもあります。	100
接続タイムアウト	接続タイムアウト（ハンドシェイク）、ミリ秒単位。	20000
リクエストタイムアウト	要求のタイムアウト（イベントのパケットの送信）（ミリ秒単位）。	20000
リセット接続間隔	「ホスト変更」の間隔。バランサーがサービスを提供するいくつかのホストは、指定されたホスト名の後ろに隠れている場合があることが理解されています。このパラメーターは、指定された時間間隔で強制的にランオフをマシン間で切り替えます。ランオフの作成者が考える便利さは、新しいマシンがバランサーの責任ゾーンに追加された場合、Flumeノードを再起動する必要がないことです。ランオフ自体は別の「宛先」があることを認識します。デフォルトでは、ストックはホストの変更を実行しません。	-1
maxIoWorkers	Avro Sourceのアナログスレッド。	2 * PROC_CORES
圧縮タイプ	Avro Sourceと同じです。違いは、シンクがデータを圧縮するのに対して、ソースは逆にアンパックすることです。したがって、Avro SinkがAvro Sourceにイベントを送信する場合、両方の圧縮タイプは同じである必要があります。	なし
圧縮レベル	compression-type = deflateの場合のみ圧縮レベル（0-圧縮しない、9-最大圧縮）。	6

次に、これらのコンポーネントを構成する際に考慮すべき重要事項について説明します。

バッチサイズを慎重に選択します 。

私が言ったように、これは非常に重要なパラメーターであり、その選択が不適切であると、人生を著しく損なう可能性があります。まず、バッチサイズはトランザクショントランザクション容量（transactionCapacity）以下でなければなりません。これは明示的にAvro Sinkに適用され、暗黙的にAvro Sourceに適用されます。例を考えてみましょう：

ここで、TCはtransactionCapacity、BSはバッチサイズです。通常の動作の条件は、BS <= TC1およびBS <= TC2です。つまり、在庫が機能するチャネルのトランザクション容量だけでなく、受信Avro Sourceが動作するチャネルのトランザクション容量も考慮する必要があります。そうしないと、シンクはチャンネルを空にできず、ソースはチャンネルにイベントを追加できません。そのような場合、Flumeはエラーメッセージをログに集中的にアップロードし始めます。

練習からのケース 。 ドレインの1つでは、バッチサイズ= 10000に設定し、受信ノードではチャネルにTC = 5000を設定しました。 データ量は少ないものの、在庫は一度に10,000件のイベントを一度にチャネルから引き出しなかったため、多くのイベントがチャネルに蓄積できませんでした。 しかし、しばらくすると、データ量が増加し、問題が発生し始めました。 受信ノードは、データの大きなパケットを拒否し始めました。 エラーが時間内に通知され、パラメータが変更され、チャネルに蓄積されたデータがいたずらなストリームによって宛先に到達しました。

大きなバンドルでイベントを送信します 。

トランザクションは、リソースの点で非常に高価な操作です。トランザクションの削減-パフォーマンスの向上。繰り返しますが、多数のイベントを送信するときの圧縮は、はるかに効率的に機能します。したがって、バッチサイズに加えて、チャネルのtransactionCapacityを増やす必要があります。

ノードのnetty依存関係をオーバーライドします 。

Flumeが古いnetty 3.6.2 Finalをプルアップしている間に、 nettyバージョン3.10.5 Finalを使用しています。古いバージョンの問題は小さなバグです。そのため、Avro Sink / Avro Sourceは定期的に相互に接続できません。これにより、データ転送中に数分間、ダウンタイムが定期的に発生します（その後、すべてが正常に戻ります）。データをできるだけ早く到着させる必要がある場合、そのような「交通渋滞」が問題になる可能性があります。

JavaでFlumeを起動する場合、Mavenで依存関係をオーバーライドできます。 Clouderaを使用して、またはサービスとしてFlumeを構成する場合、Netty依存関係を手動で変更する必要があります。それらは次のフォルダーにあります。
- Cloudera- / opt / cloudera / parcels / CDH-$ {バージョン} / lib / flume-ng / lib ;
- サービス（スタンドアローン）- $ FLUME_HOME / lib

ファイルロールシンク

そこで、Avro Source / Sinkとファイルチャネルに基づいてトランスポートノードを構成する方法を見つけました。現在、トランスポートネットワークを閉じるコンポーネント（つまり、Flumeの担当エリアからのデータを表示するコンポーネント）を処理する必要があります。

考慮すべき最初のドレインは、File-Roll Sinkです。私はこれが怠け者のドレインだと言うでしょう。最小限の設定をサポートし、1つのことしかできません-ファイルへのイベントの書き込み。

パラメータ	説明	デフォルトで
タイプ	ドレインの実装、 file_rollを指定する必要があります。	-
チャンネル	株式がイベントを描画するチャンネル。	-
ディレクトリ	ファイルが保存されるフォルダー。	-
rollInterval	新しいファイルの作成間隔（0-1つのファイルにすべてを書き込む）、秒単位。	30
シリアライザー	イベントのシリアル化。 TEXT、HEADER_AND_TEXT、AVRO_EVENT、またはEventSerializer.Builderインターフェイスを実装する独自のクラスを指定できます。	テキスト
バッチサイズ	Avro Sinkと同様に、チャネルからトランザクションごとに取得されるイベントのパケットのサイズ。	100

怠け者のドレインと考えるのはなぜですか？絶対に何も設定できないからです。圧縮も、ファイル名（作成のタイムスタンプが名前として使用されます）も、サブフォルダーによるグループ化もありません-何もありません。ファイルサイズも制限できません。この在庫は、おそらく「説明する時間がない-緊急にデータの受信を開始する必要がある」場合にのみ適しています。

ご注意 ファイルにデータを書き込む必要があるため、ファイルドレインを使用するよりもファイルドレインを実装する方が望ましいという結論に達しました。 Flumeのすべてのソースが公開されていることを考えると、作成するのは難しくありませんでした。 2日目に、マイナーなバグが修正されました。ストックは1年以上適切に機能しており、データをフォルダに分解してきちんとしたアーカイブにしています。 サイクルの3番目の部分の後に、この株式をGitHubに投稿します。

HDFSシンク

この株はすでにより深刻です-それは多くの設定をサポートしています。 File-Roll Sinkが同様の方法で作成されていないことは少し驚きです。

パラメータ	説明	デフォルトで
タイプ	ドレイン、 hdfsの実装を示す必要があります。	-
チャンネル	株式がイベントを描画するチャンネル。	-
hdfs.path	ファイルが書き込まれるフォルダー。このフォルダーに正しいアクセス許可が設定されていることを確認してください。 Clouderaを使用してフローを構成すると、ユーザーflumeに代わってデータが書き込まれます。	-
hdfs.filePrefix	ファイル名のプレフィックス。 File-Rollの場合のベースファイル名は、作成時のタイムスタンプです。したがって、 my-dataを指定すると、最終的なファイル名はmy-data1476318264182になります。	Flumedata
hdfs.fileSuffix	ファイル名の接尾辞。ファイル名の最後に追加されます。 .gzなどの拡張子を指定するために使用できます。	-
hdfs.inUsePrefix	filePrefixに似ていますが、データがまだ書き込まれている一時ファイル用です。	-
hdfs.inUseSuffix	fileSuffixに似ていますが、一時ファイル用です。本質的に一時的な拡張。	.tmp
hdfs.rollInterval	新しいファイルを作成する期間（秒単位）。この基準でファイルを閉じる必要がない場合は、0を設定します。	30
hdfs.rollSize	ボリュームでファイルを閉じるトリガーはバイト単位で示されます。この基準が適切でない場合も0を設定します。	1024
hdfs.rollCount	イベントの数でファイルを閉じるトリガー。 0を指定することもできます。	10
hdfs.idleTimeout	アクティビティがないためにファイルを閉じるトリガー（秒単位）。つまり、しばらくの間ファイルに何も書き込まれないと、ファイルは閉じます。このトリガーはデフォルトで無効になっています。	0
hdfs.batchSize	他の排水管と同じです。ドレインのドキュメントには、これがHDFSにリセットされる前にファイルに書き込まれるイベントの数であると書かれていますが。選択する際には、チャネルのトランザクション量にも注目します。	100
hdfs.fileType	ファイルタイプ-SequenceFile （キーと値のペアを持つHadoopファイル、原則として、「タイムスタンプ」ヘッダーからのタイムスタンプまたは現在の時刻がキーとして使用されます）、 DataStream （テキストデータ、実際には、File- Roll Sink）またはCompressedStream （DataStreamに似ていますが、圧縮あり）。	SequenceFile
hdfs.writeFormat	記録形式はTextまたはWritableです。 SequenceFileのみ。違いは、テキスト（ TextWritable ）またはバイト（ BytesWritable ）のいずれかが値として書き込まれることです。	5000
シリアライザー	File-Roll Sinkに似たDataStreamおよびCompressedStreamの構成可能。	テキスト
hdfs.codeC	CompressedStreamファイルタイプを使用している場合、このパラメーターを指定する必要があります。そのような圧縮オプションが提供されています： gzip、bzip2、lzo、lzop、snappy 。	-
hdfs.maxOpenFiles	同時に開くことができるファイルの最大数。このしきい値を超えると、最も古いファイルが閉じられます。	5000
hdfs.minBlockReplicas	重要なパラメータ。 HDFSブロックごとのレプリカの最小数。指定されていない場合、起動時にクラスパスで指定されたHadoop構成（つまり、クラスター設定）から取得されます。正直なところ、このパラメーターに関連するFlumeの動作の理由を説明することはできません。一番下の行は、このパラメーターの値が1と異なる場合、ストックは他のトリガーを見ずにファイルを閉じ始め、記録的な時間で多くの小さなファイルを生成することです。	-
hdfs.maxOpenFiles	同時に開くことができるファイルの最大数。このしきい値を超えると、最も古いファイルが閉じられます。	5000
hdfs.callTimeout	HDFSへのアクセスのタイムアウト（ファイルのオープン/クローズ、データのリセット）、ミリ秒単位。	10,000
hdfs.closeTries	ファイルを閉じようとする回数（最初に機能しなかった場合）。 0-最後まで試行します。	0
hdfs.retryInterval	失敗した場合にファイルを閉じようとする頻度（秒単位）。	180
hdfs.threadsPoolSize	HDFSでIO操作を実行するスレッドの数。多くのファイルにパッケージ化されたイベントの「寄せ集め」がある場合、この数を増やすことをお勧めします。	10
hdfs.rollTimerPoolSize	前のプールとは異なり、このスレッドプールはスケジュールされたタスクを実行します（ファイルを閉じます）。さらに、 rollIntervalとretryIntervalの 2つのパラメーターに基づいて機能します。つまりこのプールは、トリガーによるスケジュールされたシャットダウンと、ファイルを閉じるための定期的な繰り返し試行の両方を実行します。 1つのスレッドで十分です。	1
hdfs.useLocalTimeStamp	HDFSストックでは、生成されたファイルの名前に日付要素を使用します（たとえば、 hdfs.path = / logs /％Y-％m-％dにより、ファイルを日ごとにグループ化できます）。日付を使用すると、どこかから受信する必要があることが示唆されます。このパラメーターには2つのオプションがあります。イベントが処理された時間を使用する（ true ）か、イベントで指定された時間を使用する（つまり、「タイムスタンプ」ヘッダーで使用する（false））。タイムスタンプイベントを使用する場合は、メッセージにこのヘッダーがあることを確認してください。そうしないと、HDFSに記録されません。	偽
hdfs.round	タイムスタンプを値に丸めます。	偽
hdfs.roundValue	タイムスタンプを丸める方法。	1
hdfs.roundUnit	丸める単位（ second 、 minutesまたはhour ）。	第二

HDFS-drainの設定の膨大なリストを以下に示します。このストックを使用すると、ほとんど好きなようにデータをファイルに切り分けることができます。特に、日付要素を使用できるというのは素晴らしいことです。この株の公式文書は同じページにあります。

HDFSドレイン構成の興味深い機能に気づいたかもしれません-HDFSのアドレスを示すパラメーターはありません。ドレインの作成者は、このドレインをHDFSと同じマシンで使用することを提案しています。

したがって、このドレインを設定するときに考慮すべきことは何ですか。

大きなバッチサイズとtransactionCapacityを使用します。

一般に、ここでのすべては他のドレインに似ています-トランザクションはリソースの点で非常に高価なので、大量に注ぐ方が良いです。

ファイルの命名にマクロを乱用しないでください 。

ファイル/フォルダー内の日付要素またはヘッダーのプレースホルダー名を使用することは、確かに便利なツールです。しかし、それほど高速ではありません。作成者が日付の置換をより最適化できるように思えます-ソースを見ると、これらの行をフォーマットするために実行される操作の数に驚くでしょう。このようなフォルダー構造をセットアップすることにしたと仮定します。
```
  hdfs.path = / logs /％{dir}
 hdfs.filePrefix =％{src} /％Y-％m-％d /％Y-％m-％d-％H-％M-％S.％{host}。％{src} 
```
ここで、dirとsrcは、それぞれイベントヘッダーの値です。キー。結果のファイルは/logs/web/my-source/2016-04-15/2016-04-15-12-00-00.my-host.my-source.gzのようになります。私のコンピューターでは、100万のイベントに対してこの名前を生成するのに20秒近くかかります！つまり 10,000イベントの場合、約200ミリ秒かかります。結論：毎秒10,000イベントの記録速度を主張している場合、ファイル名の生成に20％の時間を費やす準備をしてください。これはひどいです。これは、クライアント側でファイル名を生成する責任を負うことで解決できます。はい、これのためにいくつかのコードを書く必要がありますが、これらにフロー設定を変更することは可能です：
```
  hdfs.path = / logs
 hdfs.filePrefix =％{ファイル名} 
```
生成されたファイル名をfile-nameヘッダーに渡すと、リソースと時間を節約できます。このようなヘッダーによるファイルパスの形成には20秒かかりませんが、100万イベントに対して500〜600ミリ秒かかります。つまり、ほぼ40倍高速です。

イベントを組み合わせます 。

ドレーンのパフォーマンスを大幅に向上させるもう1つの小さなハック。イベントを1行ずつファイルに書き込む場合、クライアント側でそれらを結合できます。たとえば、サービスは同じファイルに送信する必要のあるログを生成します。それでは、\ nを区切り文字として使用して、複数の行を1つにまとめてみませんか？ HDFSまたはファイルシステムへのデータの書き込み自体は、データに関するこのすべての「デジタルの官僚主義」よりもはるかに短い時間で済みます。

少なくとも5対1の比率でイベントを組み合わせると、生産性が大幅に向上します。当然、ここで注意する必要があります。クライアントのイベントが一度に1つずつ生成される場合、イベントを結合するためにバッファを埋めるのに時間がかかる場合があります。この間ずっと、イベントはメモリに保存され、グループの形成が結合するのを待ちます。そのため、データを失う可能性が高くなります。要約：
1. 少量のデータの場合、クライアントが一度に1つずつイベントをFlumeに送信することをお勧めします。イベントを失う可能性は低くなります。
2. 大量のデータの場合、イベント集約を使用することをお勧めします。イベントが集中的に生成される場合、5〜10個のイベントのバッファが十分にすばやくダイヤルされます。この場合、排水のパフォーマンスが大幅に向上します。

HDFSクラスター上の複数のマシンにドレインをデプロイします 。

Clouderaを介してFlumeを構成する場合、クラスターの各ノードで個別のFlumeノードを実行することができます。そして、この機会は使用するほうが良いです-このようにして、すべてのクラスターマシン間で負荷が分散されるためです。ただし、同じ構成（つまり、すべてのマシンで同じ構成ファイル）を使用する場合は、ファイル名の競合が発生しないようにしてください。これを行うには、ヘッダーにホスト名を追加するイベントフックを使用します。したがって、ファイル名テンプレートでこのヘッダーを指定するだけで済みます（以下を参照）。

ご注意 実際、このような決定を下す際には、検討する価値があります。結局のところ、各株式は同種のデータをファイルに書き込みます。 その結果、HDFSで多数の小さなファイルを取得できます。 決定のバランスを取る必要があります-データ量が少ない場合は、HDFSでの記録用にFlumeノードを1つに制限できます。 これは、いわゆるデータ統合と呼ばれます。複数のソースからのデータが最終的に同じ在庫になる場合です。 ただし、データが流れている場合、1つのノードでは不十分な場合があります。 このシリーズの次の記事では、トランスポートネットワーク全体の設計について詳しく説明します。

イベントインターセプター

私はこれらの不思議なインターセプターについて何度も言及しましたが、おそらく今がそれが何であるかについて話す時です。インターセプターは、ソースでイベントを受信してからチャネルに送信するまでのフェーズで機能するイベントハンドラーです。インターセプターは、イベントを変換、変更、またはフィルターできます。

Flume はデフォルトで多くのインターセプターを提供します。

静的ヘッダー（定数、タイムスタンプ、ホスト名）を追加します。
ヘッダーにランダムなUUIDを生成します。
イベント（正規表現）の本体から値を抽出し、ヘッダーとして使用します。
イベントの内容を変更します（再び正規表現を使用）。
コンテンツに基づいてイベントをフィルタリングします。

さまざまなインターセプターの構成例

 # ============================ Avro-   ============================ # #    Vvro- my-agent.sources.avro-source.type = avro my-agent.sources.avro-source.bind = 0.0.0.0 my-agent.sources.avro-source.port = 50001 my-agent.sources.avro-source.channels = my-agent-channel #    ,    (   ) my-agent.sources.avro-source.interceptors = ts directory host replace group-replace filter extractor # ------------------------------------------------------------------------------ # #        . #    "dir",   — "test-folder". my-agent.sources.avro-source.interceptors.directory.type = static my-agent.sources.avro-source.interceptors.directory.key = dir my-agent.sources.avro-source.interceptors.directory.value = test-folder #      —   (  — false) my-agent.sources.avro-source.interceptors.directory.preserveExisting = true # ------------------------------------------------------------------------------ # #     "timestamp"       ,   my-agent.sources.avro-source.interceptors.ts.type = timestamp my-agent.sources.avro-source.interceptors.ts.preserveExisting = true # ------------------------------------------------------------------------------ # #      /IP   my-agent.sources.avro-source.interceptors.host.type = host my-agent.sources.avro-source.interceptors.host.useIP = true #   ( directory.key) my-agent.sources.avro-source.interceptors.host.hostHeader = host my-agent.sources.avro-source.interceptors.host.preserveExisting = true # ------------------------------------------------------------------------------ # #        ;    my-agent.sources.avro-source.interceptors.replace.type = search_replace my-agent.sources.avro-source.interceptors.replace.searchPattern = \t my-agent.sources.avro-source.interceptors.replace.replaceString = ; #    byte[],     (  — UTF-8) my-agent.sources.avro-source.interceptors.replace.charset = UTF-8 # ------------------------------------------------------------------------------ # #  ""   my-agent.sources.avro-source.interceptors.group-replace.type = search_replace # ,      2014-01-20        20/01/2014 #     .  ""   4  ()    , #        my-agent.sources.avro-source.interceptors.group-replace.searchPattern = (\\d{4})-(\\d{2})-(\\d{2})(.*) my-agent.sources.avro-source.interceptors.group-replace.replaceString = $3/$2/$1$4 # ------------------------------------------------------------------------------ # # -,      my-agent.sources.avro-source.interceptors.filter.type = regex_filter my-agent.sources.avro-source.interceptors.filter.regex = error$ #  true —   ,      , #    —  ,      my-agent.sources.avro-source.interceptors.filter.excludeEvents = true # ------------------------------------------------------------------------------ # # ,          my-agent.sources.avro-source.interceptors.extractor.type = regex_extractor # ,    : "2016-04-15;WARINING;- " my-agent.sources.avro-source.interceptors.extractor.regex = (\\d{4}-\\d{2}-\\d{2});(.*); #   —        , #   .     # (\\d{4}-\\d{2}-\\d{2}) -> $1 -> ts # (.*) -> $2 -> loglevel my-agent.sources.avro-source.interceptors.extractor.serializers = ts loglevel #      ,     TS my-agent.sources.avro-source.interceptors.extractor.serializers.ts.type = org.apache.flume.interceptor.RegexExtractorInterceptorMillisSerializer my-agent.sources.avro-source.interceptors.extractor.serializers.ts.name = timestamp my-agent.sources.avro-source.interceptors.extractor.serializers.ts.pattern = yyyy-MM-dd #     as is my-agent.sources.avro-source.interceptors.extractor.serializers.loglevel.name = level

標準のインターセプターの中には、残念ながらヘッダーフィルターが見つかりませんでした。ただし、必要に応じて、このようなインターセプターを自分で作成できます。Flumeトランスポートを完全に構成するには、別のタイプのFlumeコンポーネントであるセレクターを検討する必要があります。

チャンネルセレクター

どのイベントをどのチャネルに送信するかを理解するには、チャネルにセレクタが必要です。合計2種類のセレクターがあります。

複製 -セレクタ。これにより、ソースは関連するすべてのチャネルでイベントを複製します。Flumeをデフォルトで使用するのは彼です。同時に、このセレクタを使用すると、「オプションの」チャネルを選択できます。メインのものとは異なり、ソースはそのようなチャネルへの失敗したイベントの追加を無視します。
多重化 -いくつかのルールに従ってチャネル間でイベントを分配するセレクター。標準の多重化セレクターの実装により、ヘッダー値に基づいてチャネル間でイベントを配信できます。

多重化セレクターの構成例

 # ============================ Avro-   ============================ # my-source.sources.avro-source.type = avro my-source.sources.avro-source.port = 50002 my-source.sources.avro-source.bind = 127.0.0.1 my-source.sources.avro-source.channels = hdfs-channel file-roll-channel null-channel #   — multiplexing,    # ,       ""  ""  , #         HDFS,   —    my-source.sources.avro-source.selector.type = multiplexing #   ,      my-source.sources.avro-source.selector.header = type #  type = important,      HDFS,     my-source.sources.avro-source.selector.mapping.important = hdfs-channel file-roll-channel #  type = common,      my-source.sources.avro-source.selector.mapping.common = file-roll-channel #   type     - ,     # ( ,     memchannel  null-sink) my-source.sources.avro-source.selector.mapping.default = hdfs-null-channel

セレクターはインターセプターの後のイベントを処理します。これは、イベントでインターセプターのいくつかの操作（たとえば、さまざまなヘッダーの抽出）を実行し、これらの操作の結果をセレクターで既に使用できることを意味します。

おわりに

この記事は予想外に大規模であることが判明したため、このシリーズの記事の次のパートで、約束されたサイトの監視を検討することにしました。結論として、HDFSで機能するFlume構成の1つを示したいと思います。これは、ノードあたり毎秒最大約2000イベントまでの少量のデータの配信と編成に適しています。このノードには、イベントにロールヘッダー（「15m」または「60m」）、dir、およびsrcが必要です。これらを使用して、2レベルのフォルダー階層が取得されます。

HDFSのFlume構成

 flume-hdfs.sources = hdfs-source flume-hdfs.channels = hdfs-15m-channel hdfs-60m-channel hdfs-null-channel flume-hdfs.sinks = hdfs-15m-sink hdfs-60m-sink # =========== Avro-,      host ============ # flume-hdfs.sources.hdfs-source.type = avro flume-hdfs.sources.hdfs-source.port = 50002 flume-hdfs.sources.hdfs-source.bind = 0.0.0.0 flume-hdfs.sources.hdfs-source.interceptors = hostname flume-hdfs.sources.hdfs-source.interceptors.hostname.type = host flume-hdfs.sources.hdfs-source.interceptors.hostname.hostHeader = host flume-hdfs.sources.hdfs-source.channels = hdfs-null-channel hdfs-15m-channel flume-hdfs.sources.hdfs-source.selector.type = multiplexing flume-hdfs.sources.hdfs-source.selector.header = roll flume-hdfs.sources.hdfs-source.selector.mapping.15m = hdfs-15m-channel flume-hdfs.sources.hdfs-source.selector.mapping.60m = hdfs-60m-channel flume-hdfs.sources.hdfs-source.selector.mapping.default = hdfs-null-channel # ============================  , 15  ============================ # flume-hdfs.channels.hdfs-15m-channel.type = file flume-hdfs.channels.hdfs-15m-channel.maxFileSize = 1073741824 flume-hdfs.channels.hdfs-15m-channel.capacity = 10000000 flume-hdfs.channels.hdfs-15m-channel.transactionCapacity = 10000 flume-hdfs.channels.hdfs-15m-channel.dataDirs = /flume/flume-hdfs/hdfs-60m-channel/data1,/flume/flume-hdfs/hdfs-60m-channel/data2 flume-hdfs.channels.hdfs-15m-channel.checkpointDir = /flume/flume-hdfs/hdfs-15m-channel/checkpoint # ============================  , 60  ============================ # flume-hdfs.channels.hdfs-60m-channel.type = file flume-hdfs.channels.hdfs-60m-channel.maxFileSize = 1073741824 flume-hdfs.channels.hdfs-60m-channel.capacity = 10000000 flume-hdfs.channels.hdfs-60m-channel.transactionCapacity = 10000 flume-hdfs.channels.hdfs-60m-channel.dataDirs =/flume/flume-hdfs/hdfs-60m-channel/data1,/flume/flume-hdfs/hdfs-60m-channel/data2 flume-hdfs.channels.hdfs-60m-channel.checkpointDir = /flume/flume-hdfs/hdfs-60m-channel/checkpoint # ===========   ,   15  (5 . ) =========== # flume-hdfs.sinks.hdfs-15m-sink.type = hdfs flume-hdfs.sinks.hdfs-15m-sink.channel = hdfs-15m-channel flume-hdfs.sinks.hdfs-15m-sink.hdfs.filePrefix = %{src}/%Y-%m-%d/%Y-%m-%d-%H-%M-%S.%{src}.%{host}.log flume-hdfs.sinks.hdfs-15m-sink.hdfs.path = /logs/%{dir} flume-hdfs.sinks.hdfs-15m-sink.hdfs.fileSuffix = .gz flume-hdfs.sinks.hdfs-15m-sink.hdfs.writeFormat = Text flume-hdfs.sinks.hdfs-15m-sink.hdfs.codeC = gzip flume-hdfs.sinks.hdfs-15m-sink.hdfs.fileType = CompressedStream flume-hdfs.sinks.hdfs-15m-sink.hdfs.minBlockReplicas = 1 flume-hdfs.sinks.hdfs-15m-sink.hdfs.rollInterval = 0 flume-hdfs.sinks.hdfs-15m-sink.hdfs.rollSize = 0 flume-hdfs.sinks.hdfs-15m-sink.hdfs.rollCount = 0 flume-hdfs.sinks.hdfs-15m-sink.hdfs.idleTimeout = 300 flume-hdfs.sinks.hdfs-15m-sink.hdfs.round = true flume-hdfs.sinks.hdfs-15m-sink.hdfs.roundValue = 15 flume-hdfs.sinks.hdfs-15m-sink.hdfs.roundUnit = minute flume-hdfs.sinks.hdfs-15m-sink.hdfs.threadsPoolSize = 8 flume-hdfs.sinks.hdfs-15m-sink.hdfs.batchSize = 10000 # ===========   ,   60  (20 . ) =========== # flume-hdfs.sinks.hdfs-60m-sink.type = hdfs flume-hdfs.sinks.hdfs-60m-sink.channel = hdfs-60m-channel flume-hdfs.sinks.hdfs-60m-sink.hdfs.filePrefix = %{src}/%Y-%m-%d/%Y-%m-%d-%H-%M-%S.%{src}.%{host}.log flume-hdfs.sinks.hdfs-60m-sink.hdfs.path = /logs/%{dir} flume-hdfs.sinks.hdfs-60m-sink.hdfs.fileSuffix = .gz flume-hdfs.sinks.hdfs-60m-sink.hdfs.writeFormat = Text flume-hdfs.sinks.hdfs-60m-sink.hdfs.codeC = gzip flume-hdfs.sinks.hdfs-60m-sink.hdfs.fileType = CompressedStream flume-hdfs.sinks.hdfs-60m-sink.hdfs.minBlockReplicas = 1 flume-hdfs.sinks.hdfs-60m-sink.hdfs.rollInterval = 0 flume-hdfs.sinks.hdfs-60m-sink.hdfs.rollSize = 0 flume-hdfs.sinks.hdfs-60m-sink.hdfs.rollCount = 0 flume-hdfs.sinks.hdfs-60m-sink.hdfs.idleTimeout = 1200 flume-hdfs.sinks.hdfs-60m-sink.hdfs.round = true flume-hdfs.sinks.hdfs-60m-sink.hdfs.roundValue = 60 flume-hdfs.sinks.hdfs-60m-sink.hdfs.roundUnit = minute flume-hdfs.sinks.hdfs-60m-sink.hdfs.threadsPoolSize = 8 flume-hdfs.sinks.hdfs-60m-sink.hdfs.batchSize = 10000 # ================ NULL- +     =============== # flume-hdfs.channels.hdfs-null-channel.type = memory flume-hdfs.channels.hdfs-null-channel.capacity = 30000 flume-hdfs.channels.hdfs-null-channel.transactionCapacity = 10000 flume-hdfs.channels.hdfs-null-channel.byteCapacityBufferPercentage = 20 flume-hdfs.sinks.hdfs-null-sink.channel = hdfs-null-channel flume-hdfs.sinks.hdfs-null-sink.type = null

サイクルの次の最終記事では、以下を検討します。

Flumeに基づいた本格的なデータトランスポートを構築するプロセス。
独自のコンポーネントを開発する例。
この記事には含まれていなかった、約束されたノード監視。

Flume-データストリームを管理します。 パート2