ポイント: Apache Flumeは、高ボリュームのデータストリームを効率的に読み取り、データ損失なくストレージシステムに書き込むためのツールです。ログやTwitterフィードなど、さまざまなソースをサポートしています。
- FlumeはHDFS、Hive、ElasticSearchなどのシステムにデータを出力します。
- 設定は、データフローを管理するためにソース、シンク、チャネルを定義することを含みます。
Flume 101
Apache Flumeはデータソースからデータを読み込み、高ボリュームでストレージに書き込み、イベントを失うことなく処理します。これらのデータフィードには、ストリーミングログ、ネットワークトラフィック、Twitterフィードなどが含まれます（下記の例を参照）。
ただし、データ操作は行いません。代わりに、Hiveのようにそれを操作できるものに書き込むことができます。その場合、HiveのSQLに似たコマンド（HQL）を使用して処理します。または、ElasticSearchに出力することもでき、データはJSON形式で保存され、Kibanaでクエリできます。データを保存するための他の多くの選択肢もあります。
インストール
Flumeのインストールは簡単です。ダウンロードして解凍するだけです。設定するクラスパスなどはありません。Flumeを開始するのではなく、代わりに読み取りと保存を行いたいデータソースごとにエージェントを起動します。
Flumeのソースとシンク
習得すべき専門用語はあまりありません:
ソースは入力データです。シンクは出力です。そしてメモリチャネルはメモリ内にイベントを保存し、エージェントがダウンした場合にはそれらが失われるという条件があります。Flumeがサポートするシンク、ソース、メモリチャネルのいくつかには以下が含まれます:
ソース—Avro（スキーマを用いたデータシリアル化）、Thrift（異なるプログラミング言語間の通信を促進するためにFacebookによって書かれた抽象化）、Exec（シェルコマンド）、JMS（メッセージング）、Twitter、Kafka（LinkedInのストリーミング製品）、NetCat、Syslog、HTTP、カスタム（自分で書くことができ、サードパーティのものもたくさんあります。）
シンク—HDFS、Hive、Logger（Apache log4j）、ElasticSearch、Kafka、カスタム（Cassandra、HBase、MongoDBなどのサードパーティプラグイン）
メモリチャネル—JDBC（データベースを意味する）、Kafka、ファイル、カスタム
設定
Flumeエージェントの設定は以下に示す形式です。ソース（入力）、シンク（出力）、チャネル（ソースをシンクに接続）があります。エージェント名は自由に作成したものです。
色分けは、ソース、シンク、チャネルの名前を宣言し、それらを対応するオプションと関連付ける方法を示しています。
(agent name).sources = (source name)
(agent name).channels = (channel name)
(agent name).sinks = (sink name)
(agent name).sources.(source name).options = value
(agent name).sinks.(sink name).options = value
(agent name).channels.(channel name).options = value
(agent name).sinks.(sink name).channel = (channel name)
(agent name).sources.(sink name).channel = (channel name)
エージェントを起動するには、binフォルダーに移動し、実行します:
./flume-ng agent -n (agent name) -c conf -f ../conf/(agent config file name)
ファイルをテールする
ここにCentOSシステムでsyslogを読み取る方法の例があります。実際には、syslogを使用し、FlumeをTCPまたはUDPメッセージをリッスンするように設定することが最も一般的ですが、ここでは単にtailを使用します。
このデータをHadoopに保存したいので、ディレクトリを作成します:
hadoop fs -mkdir /flume
こちらが私たちの設定ファイルです。execおよびhdfsのタイプに注意してください。各タイプには異なるオプションがあります。execタイプはコマンドを実行する必要があります。また、hdfsタイプはパスが必要です。
cat ../conf/flume.conf
# give names to the agent, source, sinks, and channels
agent.sources=tail
agent.sinks=hdfs
agent.channels=ch1
# tell it what command to run
agent.sources.tail.type=exec
agent.sources.tail.command=tail -F /var/log/messages
# give Hadoop options
agent.sinks.hdfs.type=hdfs
agent.sinks.hdfs.hdfs.path=/flume
agent.sinks.hdfs.hdfs.filePrefix=syslog
agent.sinks.hdfs.hdfs.rollInterval=10
agent.sinks.hdfs.hdfs.rollSize=0
# give channel options
agent.channels.ch1.type=memory
agent.channels.ch1.capacity=1000
# connect the sources and sinks to the channel
agent.sources.tail.channels=ch1
agent.sinks.hdfs.channel=ch1
Flumeエージェントを起動します。エージェントの名前はflume.confに記載したエージェント名と一致させる必要があります。
./flume-ng agent -n (agent name) -c conf -f ../conf/(config file)
その後、stdoutがデータが書き込まれたことを示すメッセージを出力します:
17/03/08 16:42:44 INFO hdfs.HDFSEventSink: Closing /flume/syslog
17/03/08 16:42:44 INFO hdfs.BucketWriter: Closing /flume/syslog.1488987752411.tmp
17/03/08 16:42:44 INFO hdfs.BucketWriter: Renaming /flume/syslog.1488987752411.tmp to /flume/syslog.1488987752411
catを使用して書き込まれた内容を確認できます。必ずしも読みやすいテキストではないかもしれません。
hadoop fs -cat /flume/syslog.1488987752411
設定ファイルに誤りがある場合、FlumeはJavaスタックトレースを出力し、何が問題かを示す比較的フレンドリーなエラーメッセージを表示します。
ここでは、TwitterのストリーミングAPIを使用してTwitterのツイートを書き込み、それをHadoopに保存します。これを行うには、最初にhttps://apps.twitter.com/でアプリを作成する必要があり、接続に必要なキーが得られます。
ツイートはFlume内に構造を持ちません。したがって、それらを操作したい場合は、例えばHiveに書き込むことになります。しかし、それを行うにはHiveでTwitterスキーマを作成する必要があります。その方法についてはインターネットで指示を探すことができます。この例では、単にHadoopに書き込みます。
TwitterAgent.sources = Twitter
TwitterAgent.channels = t1
TwitterAgent.sinks = hdfs
TwitterAgent.sources.Twitter.type = org.apache.flume.source.twitter.TwitterSource
TwitterAgent.sources.Twitter.consumerKey = (obfuscated)
TwitterAgent.sources.Twitter.consumerSecret = (obfuscated)
TwitterAgent.sources.Twitter.accessToken = (obfuscated)
TwitterAgent.sources.Twitter.accessTokenSecret = (obfuscated)
TwitterAgent.sources.Twitter.keywords = health
TwitterAgent.sinks.console.type = logger
TwitterAgent.sinks.hdfs.type=hdfs
TwitterAgent.sinks.hdfs.hdfs.path=/flume/twitter
TwitterAgent.sinks.hdfs.hdfs.filePrefix=messages
TwitterAgent.sinks.hdfs.hdfs.rollInterval=10
TwitterAgent.sinks.hdfs.hdfs.rollSize=0
TwitterAgent.channels.t1.type=memory
TwitterAgent.channels.t1.capacity = 1000
TwitterAgent.sources.Twitter.channels = t1
TwitterAgent.sinks.hdfs.channel = t1
次に、Hadoopに書き込まれたファイルをリストします:
-rw-r--r-- 1 root supergroup 217127 2017-03-08 14:54
/flume/twitter/messages.1488981289896
-rw-r--r-- 1 root supergroup 232847 2017-03-08 14:55
/flume/twitter/messages.1488981289897
-rw-r--r-- 1 root supergroup 227083 2017-03-08 14:55
/flume/twitter/messages.1488981289898
コンソールにツイートをログするには:
TwitterAgent.sinks.console.type = logger
ただし、見ての通り、16進数で表示され、読みづらいです:
7/03/08 14:44:39 INFO sink.LoggerSink: Event: { headers:{} body: 4F 62 6A 01 02 16 61
76 72 6F 2E 73 63 68 65 6D Obj...avro.schem }
17/03/08 14:44:41 INFO sink.LoggerSink: Event: { headers:{} body: 4F 62 6A 01 02 16 61
76 72 6F 2E 73 63 68 65 6D Obj...avro.schem }
これがFlumeの基本的な紹介です。見ての通り、全く複雑な製品ではありません。演習として、syslogをFlumeに書き込んでみることをお勧めします。
Hadoopガイド
- 初めてのためのHadoopチュートリアル: Hadoopの基本
- Hadoop管理の入門
- Hadoopセキュリティの紹介
- Apache Flumeの紹介
- Hiveの入門
- HDFSの入門
- Hbaseの紹介
- Hadoopの例: 5つの実世界の使用例
- Hadoopキャリア: 雇用されるために
- Hadoop面接質問
- Hadoop分析の紹介
- Apache Pigの紹介
- Apache Yarnの入門
- Hadoopアーキテクチャの入門
- Hadoopの利点と利点
- Apache CassandraとのHadoopの利用
- Hadoop vs Kubernetes: K8sおよびクラウドネイティブはHadoopを終わらせるのか?
- Hadoop認定
- Hadoopクラスタ: 入門
- Hadoopエコシステムとコンポーネント
これらの投稿は私自身の意見であり、BMCの立場、戦略、または意見を必ずしも代表するものではありません。