ApacheFlume-クイックガイド

Flumeとは何ですか？

Apache Flumeは、ログファイル、イベント（など）などの大量のストリーミングデータをさまざまなソースから一元化されたデータストアに収集して転送するためのツール/サービス/データ取り込みメカニズムです。

Flumeは、信頼性が高く、分散され、構成可能なツールです。これは主に、ストリーミングデータ（ログデータ）をさまざまなWebサーバーからHDFSにコピーするように設計されています。

Flumeのアプリケーション

eコマースWebアプリケーションが、特定の地域の顧客の行動を分析したいとします。そのためには、分析のために利用可能なログデータをHadoopに移動する必要があります。ここで、ApacheFlumeが助けになります。

Flumeは、アプリケーションサーバーによって生成されたログデータをより高速にHDFSに移動するために使用されます。

Flumeの利点

Flumeを使用する利点は次のとおりです-

• Apache Flumeを使用すると、データを任意の集中型ストア（HBase、HDFS）に保存できます。

• 着信データの速度が宛先にデータを書き込むことができる速度を超えると、Flumeはデータプロデューサーと中央ストアの間の仲介役として機能し、それらの間でデータの安定したフローを提供します。

• Flumeはの機能を提供します contextual routing。

• Flumeのトランザクションはチャネルベースであり、メッセージごとに2つのトランザクション（1つの送信者と1つの受信者）が維持されます。信頼性の高いメッセージ配信を保証します。

• Flumeは、信頼性が高く、フォールトトレラントで、スケーラブルで、管理しやすく、カスタマイズ可能です。

Flumeの特徴

Flumeの注目すべき機能のいくつかは次のとおりです-

• Flumeは、複数のWebサーバーから一元化されたストア（HDFS、HBase）にログデータを効率的に取り込みます。

• Flumeを使用すると、複数のサーバーからすぐにHadoopにデータを取得できます。

• Flumeは、ログファイルに加えて、FacebookやTwitterなどのソーシャルネットワーキングサイト、およびAmazonやFlipkartなどのeコマースWebサイトによって生成された大量のイベントデータをインポートするためにも使用されます。

• Flumeは、ソースと宛先のタイプの大規模なセットをサポートしています。

• Flumeは、マルチホップフロー、ファンインファンアウトフロー、コンテキストルーティングなどをサポートします。

• Flumeは水平方向にスケーリングできます。

Big Data,ご存知のように、これは従来のコンピューティング技術では処理できない大きなデータセットのコレクションです。ビッグデータを分析すると、貴重な結果が得られます。Hadoop は、単純なプログラミングモデルを使用して、コンピューターのクラスター全体の分散環境でビッグデータを保存および処理できるようにするオープンソースフレームワークです。

ストリーミング/ログデータ

一般に、分析されるデータのほとんどは、アプリケーションサーバー、ソーシャルネットワーキングサイト、クラウドサーバー、エンタープライズサーバーなどのさまざまなデータソースによって生成されます。このデータは次の形式になりますlog files そして events。

Log file −一般に、ログファイルは fileオペレーティングシステムで発生するイベント/アクションを一覧表示します。たとえば、Webサーバーは、サーバーに対して行われたすべての要求をログファイルに一覧表示します。

このようなログデータを収集すると、次の情報を取得できます。

• アプリケーションのパフォーマンスを確認し、さまざまなソフトウェアおよびハードウェアの障害を特定します。
• ユーザーの行動とより良いビジネス洞察を導き出します。

データをHDFSシステムに転送する従来の方法は、 putコマンド。使い方を見てみましょうput コマンド。

HDFSputコマンド

ログデータを処理する際の主な課題は、複数のサーバーによって生成されたこれらのログをHadoop環境に移動することです。

Hadoop File System ShellHadoopにデータを挿入してそこから読み取るコマンドを提供します。を使用してHadoopにデータを挿入できますput 以下に示すコマンド。

$ Hadoop fs –put /path of the required file  /path in HDFS where to save the file

putコマンドの問題

使用できます putこれらのソースからHDFSにデータを転送するためのHadoopのコマンド。ただし、次のような欠点があります。

• 使用する put コマンド、転送できます only one file at a time一方、データジェネレータははるかに高いレートでデータを生成します。古いデータで行われた分析は精度が低いため、データをリアルタイムで転送するためのソリューションが必要です。

• 使用する場合 putコマンドを実行する場合、データをパッケージ化する必要があり、アップロードの準備ができている必要があります。Webサーバーは継続的にデータを生成するため、これは非常に困難な作業です。

ここで必要なのは、次の欠点を克服できるソリューションです。 put 「ストリーミングデータ」をコマンドして、データジェネレーターから集中型ストア（特にHDFS）に遅延を抑えて転送します。

HDFSの問題

HDFSでは、ファイルはディレクトリエントリとして存在し、ファイルが閉じられるまでファイルの長さはゼロと見なされます。たとえば、ソースがHDFSにデータを書き込んでいて、操作の途中で（ファイルを閉じずに）ネットワークが中断された場合、ファイルに書き込まれたデータは失われます。

したがって、ログデータをHDFSに転送するには、信頼性が高く、構成可能で、保守可能なシステムが必要です。

Note− POSIXファイルシステムでは、ファイルにアクセスしているとき（たとえば、書き込み操作を実行しているとき）はいつでも、他のプログラムがこのファイル（少なくともファイルの保存された部分）を読み取ることができます。これは、ファイルを閉じる前にファイルがディスク上に存在するためです。

利用可能なソリューション

さまざまなソースからHDFSにストリーミングデータ（ログファイル、イベントなど）を送信するために、次のツールを自由に使用できます。

Facebookのスクライブ

Scribeは、ログデータの集約とストリーミングに使用される非常に人気のあるツールです。非常に多数のノードに拡張できるように設計されており、ネットワークおよびノー​​ドの障害に対して堅牢です。

Apache Kafka

KafkaはApacheSoftwareFoundationによって開発されました。オープンソースのメッセージブローカーです。Kafkaを使用すると、高スループットで低レイテンシのフィードを処理できます。

Apache Flume

Apache Flumeは、ログデータ、イベント（など）などの大量のストリーミングデータを収集して、さまざまなWebサービスから一元化されたデータストアに転送するためのツール/サービス/データ取り込みメカニズムです。

これは、主にストリーミングデータをさまざまなソースからHDFSに転送するように設計された、信頼性が高く、分散され、構成可能なツールです。

このチュートリアルでは、いくつかの例を使用してFlumeの使用方法について詳しく説明します。

次の図は、Flumeの基本的なアーキテクチャを示しています。図に示すように、data generators （Facebook、Twitterなど）個々のFlumeによって収集されるデータを生成します agentsそれらの上で実行されます。その後、data collector （エージェントでもあります）は、エージェントからデータを収集します。データは集約され、HDFSやHBaseなどの中央ストアにプッシュされます。

Flumeイベント

アン event 内部で転送されるデータの基本単位です Flume。これには、オプションのヘッダーを伴って送信元から宛先に転送されるバイト配列のペイロードが含まれています。典型的なFlumeイベントは次の構造になります-

Flumeエージェント

アン agentFlumeの独立したデーモンプロセス（JVM）です。クライアントまたは他のエージェントからデータ（イベント）を受信し、次の宛先（シンクまたはエージェント）に転送します。Flumeには複数のエージェントが含まれる場合があります。次の図は、Flume Agent

図に示すように、Flume Agentには、次の3つの主要コンポーネントが含まれています。 source、 channel、および sink。

ソース

A source は、データジェネレータからデータを受信し、Flumeイベントの形式で1つ以上のチャネルに転送するエージェントのコンポーネントです。

Apache Flumeはいくつかのタイプのソースをサポートし、各ソースは指定されたデータジェネレーターからイベントを受信します。

Example − Avroソース、Thriftソース、Twitter 1％ソースなど。

チャネル

A channelは、ソースからイベントを受信し、シンクによって消費されるまでイベントをバッファリングする一時ストアです。これは、ソースとシンクの間のブリッジとして機能します。

これらのチャネルは完全にトランザクション型であり、任意の数のソースとシンクで機能します。

Example − JDBCチャネル、ファイルシステムチャネル、メモリチャネルなど。

シンク

A sinkデータをHBaseやHDFSなどの集中型ストアに保存します。チャネルからのデータ（イベント）を消費し、宛先に配信します。シンクの宛先は、別のエージェントまたは中央ストアである可能性があります。

Example −HDFSシンク

Note−水路エージェントは、複数のソース、シンク、およびチャネルを持つことができます。このチュートリアルのFlume構成の章に、サポートされているすべてのソース、シンク、チャネルをリストしました。

FlumeAgentの追加コンポーネント

上で説明したのは、エージェントの基本的なコンポーネントです。これに加えて、データジェネレーターから中央ストアにイベントを転送する際に重要な役割を果たすコンポーネントがいくつかあります。

インターセプター

インターセプターは、ソースとチャネル間で転送される水路イベントを変更/検査するために使用されます。

チャンネルセレクター

これらは、複数のチャネルの場合にデータを転送するために選択するチャネルを決定するために使用されます。チャネルセレクタには2つのタイプがあります-

• Default channel selectors −これらは、各チャネルのすべてのイベントを複製する複製チャネルセレクターとも呼ばれます。

• Multiplexing channel selectors −これらは、イベントのヘッダー内のアドレスに基づいて、イベントを送信するチャネルを決定します。

シンクプロセッサ

これらは、選択したシンクのグループから特定のシンクを呼び出すために使用されます。これらは、シンクのフェイルオーバーパスを作成したり、チャネルから複数のシンクにまたがるイベントを負荷分散したりするために使用されます。

Flumeは、ログデータをHDFSに移動するために使用されるフレームワークです。通常、イベントとログデータはログサーバーによって生成され、これらのサーバーではFlumeエージェントが実行されています。これらのエージェントは、データジェネレーターからデータを受け取ります。

これらのエージェントのデータは、次のような中間ノードによって収集されます。 Collector。エージェントと同じように、Flumeには複数のコレクターが存在する可能性があります。

最後に、これらすべてのコレクターからのデータが集約され、HBaseやHDFSなどの中央ストアにプッシュされます。次の図は、Flumeのデータフローを説明しています。

マルチホップフロー

Flume内には複数のエージェントが存在する可能性があり、最終目的地に到達する前に、イベントが複数のエージェントを経由する場合があります。これはとして知られていますmulti-hop flow。

ファンアウトフロー

1つのソースから複数のチャネルへのデータフローは、 fan-out flow。2種類あります-

• Replicating −設定されたすべてのチャネルでデータが複製されるデータフロー。

• Multiplexing −イベントのヘッダーに記載されている選択されたチャネルにデータが送信されるデータフロー。

ファンインフロー

データが多くのソースから1つのチャネルに転送されるデータフローは、次のように知られています。 fan-in flow。

障害処理

Flumeでは、イベントごとに2つのトランザクションが発生します。1つは送信者で、もう1つは受信者で行われます。送信者は受信者にイベントを送信します。データを受信するとすぐに、受信者は独自のトランザクションをコミットし、「受信した」信号を送信者に送信します。シグナルを受信した後、送信者はトランザクションをコミットします。（送信者は、受信者から信号を受信するまでトランザクションをコミットしません。）

Flumeのアーキテクチャについては、前の章ですでに説明しました。この章では、ApacheFlumeをダウンロードしてセットアップする方法を見てみましょう。

先に進む前に、システムにJava環境が必要です。したがって、まず、システムにJavaがインストールされていることを確認してください。このチュートリアルのいくつかの例では、Hadoop HDFSを（シンクとして）使用しました。したがって、Javaと一緒にHadoopをインストールすることをお勧めします。詳細情報を収集するには、リンクをたどってください-http://www.tutorialspoint.com/hadoop/hadoop_enviornment_setup.htm

Flumeのインストール

まず、WebサイトからApacheFlumeソフトウェアの最新バージョンをダウンロードします。 https://flume.apache.org/。

ステップ1

ウェブサイトを開きます。クリックしてくださいdownloadホームページの左側にあるリンク。ApacheFlumeのダウンロードページに移動します。

ステップ2

ダウンロードページでは、ApacheFlumeのバイナリファイルとソースファイルへのリンクを確認できます。リンクapache-flume-1.6.0-bin.tar.gzをクリックします

これらのミラーのいずれかをクリックしてダウンロードを開始できるミラーのリストにリダイレクトされます。同様に、apache-flume-1.6.0-src.tar.gzをクリックすると、ApacheFlumeのソースコードをダウンロードできます。

ステップ3

のインストールディレクトリと同じディレクトリにFlumeという名前のディレクトリを作成します。 Hadoop、 HBase、およびその他のソフトウェアがインストールされました（すでにインストールされている場合）。

$ mkdir Flume

ステップ4

以下に示すように、ダウンロードしたtarファイルを抽出します。

$ cd Downloads/ 
$ tar zxvf apache-flume-1.6.0-bin.tar.gz  
$ tar zxvf apache-flume-1.6.0-src.tar.gz

ステップ5

apacheのコンテンツを移動する-flume-1.6.0-bin.tar ファイルに Flume以下に示すように、以前に作成されたディレクトリ。（Hadoopという名前のローカルユーザーにFlumeディレクトリを作成したと仮定します。）

$ mv apache-flume-1.6.0-bin.tar/* /home/Hadoop/Flume/

Flumeの構成

Flumeを構成するには、次の3つのファイルを変更する必要があります。 flume-env.sh, flumeconf.properties, そして bash.rc。

パス/クラスパスの設定

の中に .bashrc ファイルに、Flumeのホームフォルダ、パス、クラスパスを以下のように設定します。

confフォルダ

あなたが開くと conf Apache Flumeのフォルダーには、次の4つのファイルがあります-

• flume-conf.properties.template,
• flume-env.sh.template,
• flume-env.ps1.template、および
• log4j.properties.

名前を変更します

• flume-conf.properties.template としてファイル flume-conf.properties そして

• flume-env.sh.template なので flume-env.sh

flume-env.sh

開いた flume-env.sh ファイルを作成し、 JAVA_Home システムにJavaがインストールされているフォルダに移動します。

インストールの確認

を参照して、ApacheFlumeのインストールを確認します。 bin フォルダと次のコマンドを入力します。

$ ./flume-ng

Flumeが正常にインストールされると、以下に示すようにFlumeのヘルププロンプトが表示されます。

Flumeをインストールした後、Javaプロパティファイルである構成ファイルを使用して構成する必要があります。 key-value pairs。ファイル内のキーに値を渡す必要があります。

Flume構成ファイルでは、次のことを行う必要があります。

• 現在のエージェントのコンポーネントに名前を付けます。
• ソースを説明/構成します。
• シンクの説明/構成。
• チャネルを説明/構成します。
• ソースとシンクをチャネルにバインドします。

通常、Flumeには複数のエージェントを含めることができます。一意の名前を使用して、各エージェントを区別できます。そして、この名前を使用して、各エージェントを構成する必要があります。

コンポーネントの命名

まず、以下に示すように、エージェントのソース、シンク、チャネルなどのコンポーネントに名前を付けてリストする必要があります。

agent_name.sources = source_name 
agent_name.sinks = sink_name 
agent_name.channels = channel_name

Flumeは、さまざまなソース、シンク、およびチャネルをサポートします。それらは以下の表にリストされています。

ソース	チャンネル	シンク
アブロソース スリフトソース Execソース JMSソース スプーリングディレクトリソース Twitter 1％消防ホース出典 カフカソース NetCatソース シーケンスジェネレータソース Syslogソース SyslogTCPソース マルチポートSyslogTCPソース SyslogUDPソース HTTPソース ストレス源 レガシーソース スリフトレガシーソース カスタムソース スクライブソース	メモリチャネル JDBCチャネル カフカチャンネル ファイルチャネル こぼれるメモリチャネル 疑似トランザクションチャネル	HDFSシンク ハイブシンク ロガーシンク アブロシンク スリフトシンク IRCシンク ファイルロールシンク ヌルシンク HBaseSink AsyncHBaseSink MorphlineSolrSink ElasticSearchSink Kiteデータセットシンク カフカシンク

あなたはそれらのどれでも使うことができます。たとえば、Twitterソースを使用してメモリチャネルを介してHDFSシンクにTwitterデータを転送している場合、エージェント名IDはTwitterAgent、その後

TwitterAgent.sources = Twitter 
TwitterAgent.channels = MemChannel 
TwitterAgent.sinks = HDFS

エージェントのコンポーネントを一覧表示した後、プロパティに値を指定して、ソース、シンク、およびチャネルを説明する必要があります。

ソースの説明

各ソースには、プロパティの個別のリストがあります。「type」という名前のプロパティはすべてのソースに共通であり、使用しているソースのタイプを指定するために使用されます。

プロパティ「type」とともに、すべての値を提供する必要があります required 以下に示すように、特定のソースのプロパティを構成します。

agent_name.sources. source_name.type = value 
agent_name.sources. source_name.property2 = value 
agent_name.sources. source_name.property3 = value

たとえば、 twitter source、以下は、それを構成するために値を提供する必要があるプロパティです。

TwitterAgent.sources.Twitter.type = Twitter (type name) 
TwitterAgent.sources.Twitter.consumerKey =  
TwitterAgent.sources.Twitter.consumerSecret = 
TwitterAgent.sources.Twitter.accessToken =   
TwitterAgent.sources.Twitter.accessTokenSecret =

シンクの説明

ソースと同様に、各シンクには個別のプロパティのリストがあります。「type」という名前のプロパティはすべてのシンクに共通であり、使用しているシンクのタイプを指定するために使用されます。プロパティ「type」とともに、すべてのに値を提供する必要がありますrequired 以下に示すように、特定のシンクのプロパティを構成します。

agent_name.sinks. sink_name.type = value 
agent_name.sinks. sink_name.property2 = value 
agent_name.sinks. sink_name.property3 = value

たとえば、 HDFS sink、以下は、それを構成するために値を提供する必要があるプロパティです。

TwitterAgent.sinks.HDFS.type = hdfs (type name)  
TwitterAgent.sinks.HDFS.hdfs.path = HDFS directory’s Path to store the data

チャネルの説明

Flumeは、ソースとシンク間でデータを転送するためのさまざまなチャネルを提供します。したがって、ソースとチャネルとともに、エージェントで使用されるチャネルを記述する必要があります。

各チャネルを説明するには、以下に示すように、必要なプロパティを設定する必要があります。

agent_name.channels.channel_name.type = value 
agent_name.channels.channel_name. property2 = value 
agent_name.channels.channel_name. property3 = value

たとえば、 memory channel、以下は、それを構成するために値を提供する必要があるプロパティです。

TwitterAgent.channels.MemChannel.type = memory (type name)

ソースとシンクをチャネルにバインドする

チャネルはソースとシンクを接続するため、以下に示すように、両方をチャネルにバインドする必要があります。

agent_name.sources.source_name.channels = channel_name 
agent_name.sinks.sink_name.channels = channel_name

次の例は、ソースとシンクをチャネルにバインドする方法を示しています。ここでは、twitter source, memory channel, そして HDFS sink。

TwitterAgent.sources.Twitter.channels = MemChannel
TwitterAgent.sinks.HDFS.channels = MemChannel

Flumeエージェントの開始

構成後、Flumeエージェントを起動する必要があります。それは次のように行われます-

$ bin/flume-ng agent --conf ./conf/ -f conf/twitter.conf 
Dflume.root.logger=DEBUG,console -n TwitterAgent

ここで−

• agent −Flumeエージェントを開始するコマンド

• --conf ,-c<conf> −confディレクトリの設定ファイルを使用する

• -f<file> −欠落している場合は、構成ファイルのパスを指定します

• --name, -n <name> −Twitterエージェントの名前

• -D property =value −Javaシステムのプロパティ値を設定します。

Flumeを使用すると、さまざまなサービスからデータをフェッチして、中央のストア（HDFSおよびHBase）に転送できます。この章では、Twitterサービスからデータをフェッチし、ApacheFlumeを使用してHDFSに保存する方法について説明します。

Flumeアーキテクチャーで説明されているように、Webサーバーはログデータを生成し、このデータはFlumeのエージェントによって収集されます。チャネルはこのデータをシンクにバッファリングし、シンクは最終的にデータを集中型ストアにプッシュします。

この章で提供される例では、アプリケーションを作成し、ApacheFlumeが提供する実験的なTwitterソースを使用してアプリケーションからツイートを取得します。メモリチャネルを使用してこれらのツイートをバッファリングし、HDFSシンクを使用してこれらのツイートをHDFSにプッシュします。

Twitterデータを取得するには、以下の手順に従う必要があります-

• Twitterアプリケーションを作成する
• HDFSのインストール/開始
• Flumeを構成する

Twitterアプリケーションの作成

Twitterからツイートを取得するには、Twitterアプリケーションを作成する必要があります。以下の手順に従って、Twitterアプリケーションを作成します。

ステップ1

Twitterアプリケーションを作成するには、次のリンクをクリックしてください https://apps.twitter.com/。Twitterアカウントにサインインします。Twitterアプリを作成、削除、管理できるTwitterアプリケーション管理ウィンドウが表示されます。

ステップ2

クリックしてください Create New Appボタン。アプリを作成するために詳細を入力する必要があるアプリケーションフォームを取得するウィンドウにリダイレクトされます。Webサイトのアドレスを入力するときに、たとえば、完全なURLパターンを指定します。http://example.com.

ステップ3

詳細を入力し、同意します Developer Agreement 終了したら、をクリックします Create your Twitter application buttonこれはページの下部にあります。すべてがうまくいけば、アプリは以下に示すように与えられた詳細で作成されます。

ステップ4

下 keys and Access Tokens ページ下部のタブでは、という名前のボタンを確認できます Create my access token。それをクリックして、アクセストークンを生成します。

ステップ5

最後に、をクリックします Test OAuthページの右上にあるボタン。これはあなたを表示するページにつながりますConsumer key, Consumer secret, Access token, そして Access token secret。これらの詳細をコピーします。これらは、Flumeでエージェントを構成するのに役立ちます。

HDFSの開始

データをHDFSに保存しているため、Hadoopをインストール/検証する必要があります。Hadoopを起動し、Flumeデータを保存するためのフォルダーを作成します。Flumeを構成する前に、以下の手順に従ってください。

ステップ1：Hadoopをインストール/検証する

Hadoopをインストールします。Hadoopがすでにシステムにインストールされている場合は、以下に示すように、Hadoopバージョンコマンドを使用してインストールを確認します。

$ hadoop version

システムにHadoopが含まれていて、パス変数を設定している場合は、次の出力が得られます-

Hadoop 2.6.0 
Subversion https://git-wip-us.apache.org/repos/asf/hadoop.git -r 
e3496499ecb8d220fba99dc5ed4c99c8f9e33bb1 
Compiled by jenkins on 2014-11-13T21:10Z 
Compiled with protoc 2.5.0 
From source with checksum 18e43357c8f927c0695f1e9522859d6a 
This command was run using /home/Hadoop/hadoop/share/hadoop/common/hadoop-common-2.6.0.jar

ステップ2：Hadoopを開始する

を閲覧する sbin 以下に示すように、Hadoopのディレクトリと開始yarnおよびHadoop dfs（分散ファイルシステム）。

cd /$Hadoop_Home/sbin/ 
$ start-dfs.sh 
localhost: starting namenode, logging to
   /home/Hadoop/hadoop/logs/hadoop-Hadoop-namenode-localhost.localdomain.out 
localhost: starting datanode, logging to 
   /home/Hadoop/hadoop/logs/hadoop-Hadoop-datanode-localhost.localdomain.out 
Starting secondary namenodes [0.0.0.0] 
starting secondarynamenode, logging to 
   /home/Hadoop/hadoop/logs/hadoop-Hadoop-secondarynamenode-localhost.localdomain.out
  
$ start-yarn.sh 
starting yarn daemons 
starting resourcemanager, logging to 
   /home/Hadoop/hadoop/logs/yarn-Hadoop-resourcemanager-localhost.localdomain.out 
localhost: starting nodemanager, logging to 
   /home/Hadoop/hadoop/logs/yarn-Hadoop-nodemanager-localhost.localdomain.out

ステップ3：HDFSでディレクトリを作成する

Hadoop DFSでは、コマンドを使用してディレクトリを作成できます mkdir。それを参照して、名前でディレクトリを作成しますtwitter_data 以下に示すように、必要なパスで。

$cd /$Hadoop_Home/bin/ 
$ hdfs dfs -mkdir hdfs://localhost:9000/user/Hadoop/twitter_data

Flumeの構成

の構成ファイルを使用して、ソース、チャネル、およびシンクを構成する必要があります。 confフォルダ。この章の例では、ApacheFlumeが提供する実験的なソースを使用しています。Twitter 1% Firehose メモリチャネルとHDFSシンク。

Twitter 1％Firehose Source

このソースは非常に実験的です。ストリーミングAPIを使用して1％のサンプルTwitter Firehoseに接続し、ツイートを継続的にダウンロードしてAvro形式に変換し、AvroイベントをダウンストリームのFlumeシンクに送信します。

Flumeのインストールとともに、デフォルトでこのソースを取得します。ザ・jar このソースに対応するファイルは、 lib 以下に示すようなフォルダ。

クラスパスの設定

をセットする classpath 変数 lib Flumeのフォルダ Flume-env.sh 以下に示すようにファイルします。

export CLASSPATH=$CLASSPATH:/FLUME_HOME/lib/*

このソースには、次のような詳細が必要です Consumer key, Consumer secret, Access token, そして Access token secretTwitterアプリケーションの。このソースを構成するときに、次のプロパティに値を指定する必要があります-

• Channels

• Source type : org.apache.flume.source.twitter.TwitterSource

• consumerKey −OAuthコンシューマキー

• consumerSecret −OAuthコンシューマーシークレット

• accessToken −OAuthアクセストークン

• accessTokenSecret −OAuthトークンシークレット

• maxBatchSize−Twitterバッチに含める必要のあるTwitterメッセージの最大数。デフォルト値は1000（オプション）です。

• maxBatchDurationMillis−バッチを閉じる前に待機する最大ミリ秒数。デフォルト値は1000（オプション）です。

チャネル

メモリチャネルを使用しています。メモリチャネルを設定するには、チャネルのタイプに値を指定する必要があります。

• type−チャネルのタイプを保持します。この例では、タイプはMemChannel。

• Capacity−チャネルに保存されているイベントの最大数です。デフォルト値は100（オプション）です。

• TransactionCapacity−チャネルが受け入れるまたは送信するイベントの最大数です。デフォルト値は100（オプション）です。

HDFSシンク

このシンクはデータをHDFSに書き込みます。このシンクを構成するには、次の詳細を指定する必要があります。

• Channel

• type − hdfs

• hdfs.path −データが保存されるHDFS内のディレクトリのパス。

また、シナリオに基づいていくつかのオプション値を提供できます。以下に示すのは、アプリケーションで構成しているHDFSシンクのオプションのプロパティです。

• fileType −これはHDFSファイルに必要なファイル形式です。 SequenceFile, DataStream そして CompressedStreamこのストリームで使用できる3つのタイプです。この例では、DataStream。

• writeFormat −テキストまたは書き込み可能のいずれかです。

• batchSize−HDFSにフラッシュされる前にファイルに書き込まれたイベントの数です。デフォルト値は100です。

• rollsize−ロールをトリガーするファイルサイズです。デフォルト値は100です。

• rollCount−ファイルがロールされる前にファイルに書き込まれたイベントの数です。デフォルト値は10です。

例–構成ファイル

以下に、構成ファイルの例を示します。このコンテンツをコピーして名前を付けて保存twitter.conf Flumeのconfフォルダーにあります。

# Naming the components on the current agent. 
TwitterAgent.sources = Twitter 
TwitterAgent.channels = MemChannel 
TwitterAgent.sinks = HDFS
  
# Describing/Configuring the source 
TwitterAgent.sources.Twitter.type = org.apache.flume.source.twitter.TwitterSource
TwitterAgent.sources.Twitter.consumerKey = Your OAuth consumer key
TwitterAgent.sources.Twitter.consumerSecret = Your OAuth consumer secret 
TwitterAgent.sources.Twitter.accessToken = Your OAuth consumer key access token 
TwitterAgent.sources.Twitter.accessTokenSecret = Your OAuth consumer key access token secret 
TwitterAgent.sources.Twitter.keywords = tutorials point,java, bigdata, mapreduce, mahout, hbase, nosql
  
# Describing/Configuring the sink 

TwitterAgent.sinks.HDFS.type = hdfs 
TwitterAgent.sinks.HDFS.hdfs.path = hdfs://localhost:9000/user/Hadoop/twitter_data/
TwitterAgent.sinks.HDFS.hdfs.fileType = DataStream 
TwitterAgent.sinks.HDFS.hdfs.writeFormat = Text 
TwitterAgent.sinks.HDFS.hdfs.batchSize = 1000
TwitterAgent.sinks.HDFS.hdfs.rollSize = 0 
TwitterAgent.sinks.HDFS.hdfs.rollCount = 10000 
 
# Describing/Configuring the channel 
TwitterAgent.channels.MemChannel.type = memory 
TwitterAgent.channels.MemChannel.capacity = 10000 
TwitterAgent.channels.MemChannel.transactionCapacity = 100
  
# Binding the source and sink to the channel 
TwitterAgent.sources.Twitter.channels = MemChannel
TwitterAgent.sinks.HDFS.channel = MemChannel

実行

Flumeホームディレクトリを参照し、以下に示すようにアプリケーションを実行します。

$ cd $FLUME_HOME 
$ bin/flume-ng agent --conf ./conf/ -f conf/twitter.conf 
Dflume.root.logger=DEBUG,console -n TwitterAgent

すべてがうまくいくと、HDFSへのツイートのストリーミングが開始されます。以下に、ツイートの取得中のコマンドプロンプトウィンドウのスナップショットを示します。

HDFSの検証

以下のURLを使用して、Hadoop管理WebUIにアクセスできます。

http://localhost:50070/

名前の付いたドロップダウンをクリックします Utilitiesページの右側にあります。以下のスナップショットに示すように、2つのオプションが表示されます。

クリック Browse the file systemツイートを保存したHDFSディレクトリのパスを入力します。この例では、パスは次のようになります/user/Hadoop/twitter_data/。次に、HDFSに保存されているTwitterログファイルのリストを以下のように表示します。

前の章では、TwitterソースからHDFSにデータをフェッチする方法を見てきました。この章では、からデータをフェッチする方法について説明しますSequence generator。

前提条件

この章で提供されている例を実行するには、インストールする必要があります HDFS に加えて Flume。したがって、先に進む前に、Hadoopのインストールを確認し、HDFSを開始してください。（HDFSを開始する方法については、前の章を参照してください）。

Flumeの構成

の構成ファイルを使用して、ソース、チャネル、およびシンクを構成する必要があります。 confフォルダ。この章の例では、sequence generator source、 memory channel、 と HDFS sink。

シーケンスジェネレータソース

イベントを継続的に生成するソースです。0から始まり、1ずつ増加するカウンターを維持します。テスト目的で使用されます。このソースを構成するときは、次のプロパティに値を指定する必要があります-

• Channels

• Source type − seq

チャネル

私たちは使用しています memoryチャネル。メモリチャネルを設定するには、チャネルのタイプに値を指定する必要があります。以下に示すのは、メモリチャネルの構成中に提供する必要のあるプロパティのリストです。

• type−チャネルのタイプを保持します。この例では、タイプはMemChannelです。

• Capacity−チャネルに保存されているイベントの最大数です。デフォルト値は100です。（オプション）

• TransactionCapacity−チャネルが受け入れるまたは送信するイベントの最大数です。デフォルトは100です。（オプション）。

HDFSシンク

このシンクはデータをHDFSに書き込みます。このシンクを構成するには、次の詳細を指定する必要があります。

• Channel

• type − hdfs

• hdfs.path −データが保存されるHDFS内のディレクトリのパス。

また、シナリオに基づいていくつかのオプション値を提供できます。以下に示すのは、アプリケーションで構成しているHDFSシンクのオプションのプロパティです。

• fileType −これはHDFSファイルに必要なファイル形式です。 SequenceFile, DataStream そして CompressedStreamこのストリームで使用できる3つのタイプです。この例では、DataStream。

• writeFormat −テキストまたは書き込み可能のいずれかです。

• batchSize−HDFSにフラッシュされる前にファイルに書き込まれたイベントの数です。デフォルト値は100です。

• rollsize−ロールをトリガーするファイルサイズです。デフォルト値は100です。

• rollCount−ファイルがロールされる前にファイルに書き込まれたイベントの数です。デフォルト値は10です。

例–構成ファイル

以下に、構成ファイルの例を示します。このコンテンツをコピーして名前を付けて保存seq_gen .conf Flumeのconfフォルダーにあります。

# Naming the components on the current agent 

SeqGenAgent.sources = SeqSource   
SeqGenAgent.channels = MemChannel 
SeqGenAgent.sinks = HDFS 
 
# Describing/Configuring the source 
SeqGenAgent.sources.SeqSource.type = seq
  
# Describing/Configuring the sink
SeqGenAgent.sinks.HDFS.type = hdfs 
SeqGenAgent.sinks.HDFS.hdfs.path = hdfs://localhost:9000/user/Hadoop/seqgen_data/
SeqGenAgent.sinks.HDFS.hdfs.filePrefix = log 
SeqGenAgent.sinks.HDFS.hdfs.rollInterval = 0
SeqGenAgent.sinks.HDFS.hdfs.rollCount = 10000
SeqGenAgent.sinks.HDFS.hdfs.fileType = DataStream 
 
# Describing/Configuring the channel 
SeqGenAgent.channels.MemChannel.type = memory 
SeqGenAgent.channels.MemChannel.capacity = 1000 
SeqGenAgent.channels.MemChannel.transactionCapacity = 100 
 
# Binding the source and sink to the channel 
SeqGenAgent.sources.SeqSource.channels = MemChannel
SeqGenAgent.sinks.HDFS.channel = MemChannel

実行

Flumeホームディレクトリを参照し、以下に示すようにアプリケーションを実行します。

$ cd $FLUME_HOME 
$./bin/flume-ng agent --conf $FLUME_CONF --conf-file $FLUME_CONF/seq_gen.conf 
   --name SeqGenAgent

すべてがうまくいくと、ソースはシーケンス番号の生成を開始し、ログファイルの形式でHDFSにプッシュされます。

以下に示すのは、シーケンスジェネレーターによって生成されたデータをHDFSにフェッチするコマンドプロンプトウィンドウのスナップショットです。

HDFSの検証

次のURLを使用してHadoop管理WebUIにアクセスできます-

http://localhost:50070/

名前の付いたドロップダウンをクリックします Utilitiesページの右側にあります。次の図に示すように、2つのオプションが表示されます。

クリック Browse the file system シーケンスジェネレーターによって生成されたデータを保存したHDFSディレクトリのパスを入力します。

この例では、パスは次のようになります /user/Hadoop/ seqgen_data /。次に、シーケンスジェネレーターによって生成され、以下に示すようにHDFSに保存されているログファイルのリストを確認できます。

ファイルの内容を確認する

これらのログファイルにはすべて、連続した形式の番号が含まれています。これらのファイルの内容は、ファイルシステムで次のコマンドを使用して確認できます。cat 以下に示すコマンド。

この章では、イベントを生成してコンソールにログインする方法を説明する例を取り上げます。このために、私たちはを使用していますNetCat ソースと logger シンク。

前提条件

この章で提供されている例を実行するには、インストールする必要があります Flume。

Flumeの構成

の構成ファイルを使用して、ソース、チャネル、およびシンクを構成する必要があります。 confフォルダ。この章の例では、NetCat Source, Memory channel、および logger sink。

NetCatソース

NetCatソースを構成するときに、ソースを構成するときにポートを指定する必要があります。これで、ソース（NetCatソース）は指定されたポートをリッスンし、そのポートに入力した各行を個別のイベントとして受信し、指定されたチャネルを介してシンクに転送します。

このソースを構成するときに、次のプロパティに値を指定する必要があります-

• channels

• Source type − netcat

• bind −バインドするホスト名またはIPアドレス。

• port −ソースがリッスンするポート番号。

チャネル

私たちは使用しています memoryチャネル。メモリチャネルを設定するには、チャネルのタイプに値を指定する必要があります。以下に示すのは、メモリチャネルの構成中に提供する必要のあるプロパティのリストです。

• type−チャネルのタイプを保持します。この例では、タイプはMemChannel。

• Capacity−チャネルに保存されているイベントの最大数です。デフォルト値は100です。（オプション）

• TransactionCapacity−チャネルが受け入れるまたは送信するイベントの最大数です。デフォルト値は100です。（オプション）。

ロガーシンク

このシンクは、渡されたすべてのイベントをログに記録します。通常、テストまたはデバッグの目的で使用されます。このシンクを構成するには、次の詳細を指定する必要があります。

• Channel

• type −ロガー

構成ファイルの例

以下に、構成ファイルの例を示します。このコンテンツをコピーして名前を付けて保存netcat.conf Flumeのconfフォルダーにあります。

# Naming the components on the current agent
NetcatAgent.sources = Netcat   
NetcatAgent.channels = MemChannel 
NetcatAgent.sinks = LoggerSink  

# Describing/Configuring the source 
NetcatAgent.sources.Netcat.type = netcat 
NetcatAgent.sources.Netcat.bind = localhost
NetcatAgent.sources.Netcat.port = 56565  

# Describing/Configuring the sink 
NetcatAgent.sinks.LoggerSink.type = logger  

# Describing/Configuring the channel 
NetcatAgent.channels.MemChannel.type = memory 
NetcatAgent.channels.MemChannel.capacity = 1000 
NetcatAgent.channels.MemChannel.transactionCapacity = 100 
 
# Bind the source and sink to the channel 
NetcatAgent.sources.Netcat.channels = MemChannel
NetcatAgent.sinks. LoggerSink.channel = MemChannel

実行

Flumeホームディレクトリを参照し、以下に示すようにアプリケーションを実行します。

$ cd $FLUME_HOME
$ ./bin/flume-ng agent --conf $FLUME_CONF --conf-file $FLUME_CONF/netcat.conf 
   --name NetcatAgent -Dflume.root.logger=INFO,console

すべてがうまくいくと、ソースは指定されたポートのリッスンを開始します。この場合、56565。以下に示すのは、ポート56565を開始してリッスンしているNetCatソースのコマンドプロンプトウィンドウのスナップショットです。

ソースへのデータの受け渡し

NetCatソースにデータを渡すには、構成ファイルで指定されたポートを開く必要があります。別の端末を開き、を使用してソース（56565）に接続します。curlコマンド。接続が成功すると、「」というメッセージが表示されます。connected以下に示すように」。

$ curl telnet://localhost:56565 
connected

これで、データを1行ずつ入力できます（各行の後で、Enterキーを押す必要があります）。NetCatソースは、各行を個別のイベントとして受信し、「受信メッセージ」を受け取ります。OK」。

データの受け渡しが終了したら、（（）を押してコンソールを終了できます。Ctrl+C）。以下に示すのは、を使用してソースに接続したコンソールのスナップショットです。curl コマンド。

上記のコンソールに入力された各行は、ソースによって個別のイベントとして受信されます。使用しているのでLogger シンクの場合、これらのイベントは、指定されたチャネル（この場合はメモリチャネル）を介してコンソール（ソースコンソール）にログオンします。

次のスナップショットは、イベントがログに記録されるNetCatコンソールを示しています。