[Oracle Cloud] Fluentd を使って、Streaming から HDFS へデータを連携してみた

# はじめに

Oracle Cloud Infractructure(以下OCI)では、ストリーミングデータをリアルタイムに収集・処理が出来る Streaming というサービスが提供されています。Streaming は Apache Kafka 互換のAPIを持っているため、Kafka Client から接続して、データのProduce, Consume が出来ます。

今回は、Fluentd を使って、Streaming に流れるデータを、Hadoop の HDFS に格納する方法を確認していきます。



# 前提条件

Hadoop/HDFS は、[こちらの記事](https://qiita.com/sugimount/items/80e2b201c43856f74166)で構築したものを前提にしています。



# HDFS Configuration

Fluentd の [fluent-plugin-webhdfs](https://github.com/fluent/fluent-plugin-webhdfs/) を使って HDFS にデータを格納していきます。この Plugin は REST APT で HDFS を操作するための、WebHDFS インターフェースをつかっています。WebHDFS は Default で無効になっているので、Hadoop/HDFS 側の設定を変更して有効にします。

hdfs-site.xml を編集します。

```
vim $HADOOP_HOME/etc/hadoop/hdfs-site.xml
```



configuration に追加します。

```
<property>
  <name>dfs.webhdfs.enabled</name>
  <value>true</value>
</property>

<property>
  <name>dfs.support.append</name>
  <value>true</value>
</property>

<property>
  <name>dfs.support.broken.append</name>
  <value>true</value>
</property>
```



変更後のファイルの全体は以下のような感じになっています。

```
[opc@hivenew hadoop-3.2.1]$ cat $HADOOP_HOME/etc/hadoop/hdfs-site.xml
<?xml version="1.0" encoding="UTF-8"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<!--
  Licensed under the Apache License, Version 2.0 (the "License");
  you may not use this file except in compliance with the License.
  You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

  Unless required by applicable law or agreed to in writing, software
  distributed under the License is distributed on an "AS IS" BASIS,
  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  See the License for the specific language governing permissions and
  limitations under the License. See accompanying LICENSE file.
-->

<!-- Put site-specific property overrides in this file. -->

<configuration>
  <property>
    <name>dfs.name.dir</name>
    <value>/home/opc/hdfs/name</value>
  </property>
  <property>
    <name>dfs.data.dir</name>
    <value>/home/opc/hdfs/data</value>
  </property>
  <property>
    <name>dfs.replication</name>
    <value>1</value>
  </property>
  <property>
    <name>dfs.webhdfs.enabled</name>
    <value>true</value>
  </property>
  <property>
    <name>dfs.support.append</name>
    <value>true</value>
  </property>
  <property>
    <name>dfs.support.broken.append</name>
    <value>true</value>
  </property>
</configuration>
[opc@hivenew hadoop-3.2.1]$
```



Hadoop / HDFS を起動します。

```
cd $HADOOP_HOME
sbin/start-all.sh
```



jps コマンドを使って、起動されているか確認します。

```
[opc@hivenew hadoop-3.2.1]$ jps
3057 NameNode
4483 NodeManager
4805 Jps
4358 ResourceManager
3210 DataNode
3406 SecondaryNameNode
```



動作確認のため、browser上から NameNode に接続できるか確認します。

http://youripaddress:9870

![1589592361070.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/5206e4ba-0a55-5768-7deb-c1f0523be5f8.png)



# Auth Token生成

Streaming に、Kafka API で接続するためには、IAM ユーザーで Auth Token が必要です。自分の IAMユーザーの詳細画面で、Generate します。

[![1588474181611.png](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fd5e213cd-9d8d-61ca-51d9-d09a52f3bf37.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=440eb7a39c5e4d75b72a2412070b21ec)](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fd5e213cd-9d8d-61ca-51d9-d09a52f3bf37.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=440eb7a39c5e4d75b72a2412070b21ec)



適当に説明をいれます。

[![1588474211704.png](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2F826783fd-91d3-695a-fdf6-49292d958f01.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=89faeebd672917820612f9622b01c892)](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2F826783fd-91d3-695a-fdf6-49292d958f01.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=89faeebd672917820612f9622b01c892)



Token が表示されるので、メモっておきます。(画像のものは現在使えません)

[![1588474230416.png](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fb181dd69-3e36-ce09-c95e-3c6d1dd97f76.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=ce7a9621b385f852e537d5b5dc64abff)](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fb181dd69-3e36-ce09-c95e-3c6d1dd97f76.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=ce7a9621b385f852e537d5b5dc64abff)



# Stream の作成

OCI Console で、Analytics > Streaming を選択します。その後、Create Stream を選びます。

[![1588519801919.png](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fdedbe86d-6667-2f82-6f98-a18acbcc18f5.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=b84426187c4ac475c6b065c78c541a88)](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fdedbe86d-6667-2f82-6f98-a18acbcc18f5.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=b84426187c4ac475c6b065c78c541a88)



適当にパラメータを入れて作成します

[![1588519908911.png](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fa12f4d72-c6bd-b10d-533e-d52d94d7a12d.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=40aabbfc6e00eb15e6fb25ea7af419f0)](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fa12f4d72-c6bd-b10d-533e-d52d94d7a12d.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=40aabbfc6e00eb15e6fb25ea7af419f0)



Stream を作ったときに、Stream Pool が何もない場合は、Default の Stream Pool が自動的に作成されます。`teststream01`は、DefaultPool に自動配置されます。Stream Pool は、複数のStreamを管理する概念となっていて、Stream への Endpoint を Public にするのか Private にするのか、またデータの暗号に使う鍵は何を使用するのか、といった事をまとめて管理できます。

今回は自動作成にしたので、Endpoint は Public になっていて、暗号化用の鍵は Oracle 側で管理されているものを使う設定です。

[![1588520017422.png](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fd02f1ff7-aaed-1f31-b943-1ef534b77d7c.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=fd23906affa3a22a3c3b4577674e6b5b)](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fd02f1ff7-aaed-1f31-b943-1ef534b77d7c.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=fd23906affa3a22a3c3b4577674e6b5b)

上記の画面にある「View Kafka Connection Settings」を押します。

[![1588521024252.png](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2F9e75dd0e-7ce9-71f9-7962-fb86d32dacdf.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=d49a23af436f278045c15d5ae48fff7b)](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2F9e75dd0e-7ce9-71f9-7962-fb86d32dacdf.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=d49a23af436f278045c15d5ae48fff7b)

Copy All を押して、すべての設定値をメモっておきます。

[![1588521056591.png](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fc932646e-6fa7-f74f-8f74-59e40736f822.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=8dae892c965b4ffade83295244d47fb8)](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fc932646e-6fa7-f74f-8f74-59e40736f822.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=8dae892c965b4ffade83295244d47fb8)



# Fluentd Install

次のURLを確認して事前設定をします。
https://docs.fluentd.org/installation/before-install

Fluentdをインストールします。

```
$ curl -L https://toolbelt.treasuredata.com/sh/install-redhat-td-agent3.sh | sh
```



Systemd で管理されているので、起動します。

```
sudo systemctl start td-agent.service
sudo systemctl enable td-agent.service
```



# Fluentd Configration

Streaming(Kafka互換API) → Fluentd → WebHDFS の順番でデータが流れるようにします。
source には、`kafka_group` を指定しています。Stream Pool で取得した接続情報を使用して指定しています。
store には、`webhdfs` を指定しています。

詳細のパラメータを確認したい場合は、次の GitHub に書かれています。
https://github.com/fluent/fluent-plugin-kafka
https://github.com/fluent/fluent-plugin-webhdfs/

```
cat <<'EOF' > /etc/td-agent/td-agent.conf
<source>
  @type kafka_group
  @label @STREAMING

  brokers cell-1.streaming.ap-tokyo-1.oci.oraclecloud.com:9092
  consumer_group sugigroup
  topics teststream01
  format text
  
  # sasl
  username poc02/oracleidentitycloudservice/suguru.sugiyama@oracle.com/ocid1.streampool.oc1.ap-tokyo-1.amaaaaaaycetm7yawtz56lnnerap4r45y4vheekgvhdaevxf3clfpuew6mla
  password 8t[shwUN}I-d+{}8Nx_a
  sasl_over_ssl true
  ssl_ca_certs_from_system true
</source>

<label @STREAMING>
  <match **>
    @type copy
    <store>
      @type stdout
    </store>
    <store>
      @type webhdfs
      host 10.0.0.12
      port 9870
      path "/log/%Y%m%d_%H/streaming.log.#{Socket.gethostname}"
      flush_interval 10s
      username opc
    </store>
  </match>
</label>
EOF
```



Fluentd を再起動します。

```
systemctl restart td-agent
```



# 動作確認

これで、設定が終わったので、実際に動作確認をしてみましょう、OCI Streaming の管理画面上で、Stream に Produce が出来ます。`hello webhdfs!` という文字列でデータを格納してみましょう。

![1589582508141.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/98a2f7b8-9a41-b83f-b3ac-b7cd1d2de38f.png)



何も負荷を掛けていない無風状態の環境では、テストデータを格納してから、約30秒ほどで HDFS 上でデータが確認できました。もちろん負荷状況にもって格納までの時間は変わってくると思います。Fluentd に Config した内容にしたがって、次のように格納されています、

`/log/<日付>/streaming.log.<ホスト名>` 

```
[opc@hivenew hadoop-3.2.1]$ bin/hadoop fs -ls /log/20200516_00
Found 1 items
-rw-r--r--   1 opc supergroup       1325 2020-05-16 00:57 /log/20200516_00/streaming.log.hivenew
[opc@hivenew hadoop-3.2.1]$
```



cat で確認すると、`hello webhdfs!` という文字が末尾に見えています。Streaming に投げたデータが自動的に HDFS に格納されることを確認できました。

```
[opc@hivenew hadoop-3.2.1]$ bin/hadoop fs -cat /log/20200516_00/streaming.log.hivenew
2020-05-16 00:57:30,350 INFO sasl.SaslDataTransferClient: SASL encryption trust check: localHostTrusted = false, remoteHostTrusted = false
2020-05-16T00:06:25Z    test.test       {"I say":"hello1"}
2020-05-16T00:43:15Z    test.test       {"I say":"hello1"}
2020-05-16T00:52:05Z    teststream01    {"I say":"hello"}
2020-05-16T00:52:05Z    teststream01    {"I say":"hello"}
2020-05-16T00:52:05Z    teststream01    {"I say":"hello"}
2020-05-16T00:52:05Z    teststream01    {"I say":"hello"}
2020-05-16T00:52:05Z    teststream01    {"I say":"hello1"}
2020-05-16T00:52:05Z    teststream01    {"I say":"hello1"}
2020-05-16T00:52:05Z    teststream01    {"I say":"hello1"}
2020-05-16T00:52:05Z    teststream01    {"I say":"hello"}
2020-05-16T00:52:05Z    teststream01    {"I say":"hello"}
2020-05-16T00:52:05Z    teststream01    {"I say":"hello1"}
2020-05-16T00:52:05Z    teststream01    {"key":"a","value":"a"}
2020-05-16T00:56:16Z    teststream01    {"message":"message=1"}
2020-05-16T00:56:16Z    teststream01    {"message":"message=3"}
2020-05-16T00:56:16Z    teststream01    {"message":"message=5"}
2020-05-16T00:56:16Z    teststream01    {"message":"message=8"}
2020-05-16T00:56:16Z    teststream01    {"message":"message=4"}
2020-05-16T00:56:16Z    teststream01    {"message":"message=7"}
2020-05-16T00:56:18Z    teststream01    {"message":"message=0"}
2020-05-16T00:56:18Z    teststream01    {"message":"message=2"}
2020-05-16T00:56:18Z    teststream01    {"message":"message=6"}
2020-05-16T00:56:18Z    teststream01    {"message":"message=9"}
2020-05-16T00:57:14Z    teststream01    {"message":"hello webhdfs!"}
```





# 参考URL

https://docs.fluentd.org/how-to-guides/http-to-hdfs

https://inokara.hateblo.jp/entry/2016/09/17/093035

https://github.com/fluent/fluent-plugin-kafka

https://github.com/fluent/fluent-plugin-webhdfs/