問題タブ [flume-ng]

私は Apache Flume を使用するのが初めてで、それがどのように機能するかを正確に理解するのは困難です。私の問題を説明するために、私は自分の必要性と私がやったことを説明します。

csv ファイルのディレクトリ (これらのファイルは 5 分ごとに作成されます) と HDFS クラスターの間のストリームを構成したいと考えています。

「スプーリング ディレクトリ」ソースと HDFS シンクが必要であることがわかりました。それは私にこのflume.confファイルを与えます

その結果、ローカル ファイル システムで入力ファイルの名前が「.complete」に変更され、HDFS にデータがアップロードされ、Flume によって生成された一意の新しい名前が付けられます。

それはほとんど私が必要としていたものです。

しかし、アップロードする前に、ファイル固有の操作 (ヘッダーの削除、カンマのエスケープなど) を行いたいと考えています。方法がわかりません。インターセプターを使用することを考えています。しかし、データがフルームにある場合、イベントで変換され、ストリーミングされます。彼の時点では、ファイルの知識はありません。

そうしないと、元の時間イベントがファイル名に書き込まれるため、現在の日付ではなく、この時間をイベントに関連付けたいと思います。

また、元のファイル名を hdfs に保持したい (そこにはいくつかの有用な情報があります)。

誰でも私を助けるためのアドバイスがありますか?

Flume のドキュメントによると、イベント サイズ、イベント カウント、または期間に基づいてデータを HDFS に移動できます。ファイル全体をスプール ディレクトリから HDFS に単一のファイルとして移動する方法はありますか

ありがとう。

text_with_headers シリアライザー (HDFS シンク シリアライザー) を使用すると、Flume イベント ヘッダーを破棄せずに保存できます。出力形式は、ヘッダー、スペース、本文ペイロードで構成されます。本文を削除して、ヘッダーのみを保持したいと考えています。HBase シンクの場合、「RegexHbaseEventSerializer」を使用してイベントを変換できます。しかし、HDFS シンクのそのような規定を見つけることができません。

ログを継続的に生成する Tableau サーバー (Windows プラットフォーム) があります。リアルタイム分析のためにApache Sparkにストリーミングしたいと思います。次の解決策を見ましたが、要件を満たすものはないようです。

1) nxlog エージェントを使用します。将来、複数のタブロー サーバーからログが取得される可能性があるため、これはスケーラブルではありません。

2) 流暢、flume は Windows と互換性がありません。

3) Kafka は、ログ ファイルを追跡しないため、論外です。

そのような問題に対するスケーラブルな解決策は何ですか? 主な制限は、Tableau Server が Windows で実行されていることです。

HDFSシンクにデータをストリーミングする（同じファイルに追加する）フルームがあり、これを「猫」にしてHDFSから見ることができます。ただし、MapReduce ジョブは、フラッシュされた最初のバッチ (bacthSize = 100) のみを取得しています。残りは拾われていませんが、猫を飼って残りを見ることができました。ファイルがロールされた (閉じられた) 後に MapRecue ジョブを実行すると、すべてのデータが取得されます。残りのバッチが存在するにもかかわらず、MR ジョブがバッチの残りを見つけられない理由を知っていますか。

ログの収集にはflume-ng 1.5バージョンを使用しました。

データ フローには 2 つのエージェントがあり、それぞれ 2 つのホスト上にあります。

そして、データはエージェント 1からエージェント 2 に送信されます。

エージェントのコンポーネントは次のとおりです。

agent1: スプーリング ディレクトリ ソース --> ファイル チャネル --> avro シンク

agent2: avro ソース --> ファイル チャネル --> hdfs シンク

しかし、100 万分の 1 の割合のデータが失われているようです。 問題を解決するために、次の手順を試しました。

1. エージェント ログを検索します。エラーまたは例外が見つかりません。
2. エージェント モニター メトリクスの検索: チャネルから出入りするイベント数は常に等しい
3. hive クエリと hdfs ファイルによるデータ数の統計は、それぞれシェルを使用します。

agent1 の構成:

agent2 の構成

どんな提案も大歓迎です！

私はflumeを初めて使用します。私のflumeエージェントはhttpサーバーとしてソースを持ち、そこから定期的にzipファイル（圧縮されたxmlファイル）を取得します。このzipファイルは非常に小さく（10 MB未満）、zipファイルを置きたいですhdfsシンクに抽出されました。これを行う方法についていくつかのアイデアを共有してください。カスタムインターセプターを使用する必要がありますか。

ソースとしてflumeとspooldir、シンクとしてHDFSを使用して、csvファイル（6MB）をHDFSにロードしようとしています。これが私の構成ファイルです：

その後、これらのエラーが発生し、理由がわかりません:

誰でもこの問題で私を助けてもらえますか?

複数の XML ファイルを含む zip ファイルを読み取るために Apache Flume 1.6.0.SNAPSHOT を使用しています。各 XML をイベントとして表現したいと考えています。BlobDeserializer の例に従ってみましたが、XML の数に基づいてイベントを分割できません。誰かがこれについて私を助けてくれませんか。ありがとう