翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。 # Firehose ストリームにデータを送信する このセクションでは、さまざまなデータソースを使用して Firehose ストリームにデータを送信する方法について説明します。Amazon Data Firehose を初めて使用する場合、時間を取って「[Amazon Data Firehose とは何ですか?](what-is-this-service.md)」で説明している概念と用語を理解してください。 **注記** 一部の AWS サービスは、同じリージョンにある Firehose ストリームにのみメッセージとイベントを送信できます。Amazon CloudWatch Logs、CloudWatch Events、または のターゲットを設定するときに Firehose ストリームがオプションとして表示されない場合は AWS IoT、Firehose ストリームが他の サービスと同じリージョンにあることを確認します。各リージョンのサービスエンドポイントの詳細については、「[Amazon Data Firehose endpoints](https://docs.aws.amazon.com/general/latest/gr/fh.html#fh_region)」を参照してください。 次のデータソースから Firehose ストリームにデータを送信できます。 **Topics** + [データを送信するように Kinesis エージェントを設定する](writing-with-agents.md) + [AWS SDK でデータを送信する](writing-with-sdk.md) + [CloudWatch Logs を Firehose に送信する](writing-with-cloudwatch-logs.md) + [CloudWatch Events を Firehose に送信する](writing-with-cloudwatch-events.md) + [Firehose にデータを送信する AWS IoT ように を設定する](writing-with-iot.md) # データを送信するように Kinesis エージェントを設定する Amazon Kinesis エージェントは、Firehose にデータを収集および送信する方法を示すリファレンス実装として機能する、スタンドアロン Java ソフトウェアアプリケーションです。このエージェントは一連のファイルを継続的に監視し、新しいデータを Firehose ストリームに送信します。エージェントは、ファイルのローテーション、チェックポイント、および障害時の再試行を処理する方法を示します。信頼性が高く、適時かつシンプルな態様でデータを配信する方法を示します。また、ストリーミングプロセスをより適切にモニタリングおよびトラブルシューティングするために、CloudWatch メトリクスを発行する方法も示します。詳細については、[awslabs/amazon-kinesis-agent](https://github.com/awslabs/amazon-kinesis-agent) を参照してください。 デフォルトでは、レコードは改行文字 (`'\n'`) に基づいて各ファイルから解析されます。しかし、複数行レコードを解析するよう、エージェントを設定することもできます ([エージェント構成設定を指定する](agent-config-settings.md)を参照)。 このエージェントは、ウェブサーバー、ログサーバーおよびデータベースサーバーなど、Linux ベースのサーバー環境にインストールできます。エージェントをインストールした後で、モニタリングするファイルとデータ用の Firehose ストリームを指定して設定します。エージェントが設定されると、ファイルから永続的にデータを収集して、Firehose ストリームに安全にデータを送信します。 ## 前提条件 Kinesis Agent の使用を開始する前に、次の前提条件を満たしていることを確認してください。 + オペレーティングシステムは Amazon Linux、または Red Hat Enterprise Linux バージョン 7 以降でなければなりません。 + エージェントバージョン 2.0.0 以降は JRE バージョン 1.8 以降を使用して実行されます。エージェントバージョン 1.1.x は JRE 1.7 以降を使用して実行されます。 + Amazon EC2 を使用してエージェントを実行している場合は、EC2 インスタンスを起動します。 + 指定する IAM ロールまたは AWS 認証情報には、エージェントが Firehose ストリームにデータを送信するために Amazon Data Firehose [PutRecordBatch](https://docs.aws.amazon.com/firehose/latest/APIReference/API_PutRecordBatch.html) オペレーションを実行するアクセス許可が必要です。エージェントの CloudWatch モニタリングを有効にしている場合は、CloudWatch [PutMetricData](https://docs.aws.amazon.com/AmazonCloudWatch/latest/APIReference/API_PutMetricData.html) オペレーションを実行する権限も必要になります。詳細については、「[Amazon Data Firehose によるアクセスの制御](controlling-access.md)」、「[Kinesis エージェントの状態をモニタリングする](agent-health.md)」、および「[CloudWatch Amazon CloudWatch に対する認証とアクセスコントロール](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/auth-and-access-control-cw.html)」を参照してください。 # AWS 認証情報の管理 次のいずれかの方法を使用して AWS 認証情報を管理します。 + カスタム認証情報プロバイダーを作成します。詳細については、「[カスタム認証プロバイダーを作成する](custom-cred-provider.md)」を参照してください。 + EC2 インスタンスを起動する際に IAM ロールを指定します。 + エージェントを設定するときに AWS 認証情報を指定します ( `awsSecretAccessKey`の設定テーブルの `awsAccessKeyId`および のエントリを参照[エージェント構成設定を指定する](agent-config-settings.md))。 + を編集`/etc/sysconfig/aws-kinesis-agent`して、 AWS リージョンと AWS アクセスキーを指定します。 + EC2 インスタンスが別の AWS アカウントにある場合は、IAM ロールを作成して Amazon Data Firehose サービスへのアクセスを提供します。エージェントを設定するときに、そのロールを指定します ([assumeRoleARN](agent-config-settings.md#assumeRoleARN) と [assumeRoleExternalId](agent-config-settings.md#assumeRoleExternalId) を参照してください)。前の方法のいずれかを使用して、このロールを引き受けるアクセス許可を持つ他のアカウントのユーザーの AWS 認証情報を指定します。 # カスタム認証プロバイダーを作成する カスタム認証情報プロバイダーを作成し、そのクラス名と jar パスを Kinesis エージェントに渡すことができます。これらを渡すための構成設定として `userDefinedCredentialsProvider.classname` と `userDefinedCredentialsProvider.location` を使用します。これら 2 つの構成設定の説明については、「[エージェント構成設定を指定する](agent-config-settings.md)」を参照してください。 カスタム認証情報プロバイダーを作成するには、次の例に示すように、`AWS CredentialsProvider` インターフェイスを実装するクラスを定義します。 ``` import com.amazonaws.auth.AWSCredentials; import com.amazonaws.auth.AWSCredentialsProvider; import com.amazonaws.auth.BasicAWSCredentials; public class YourClassName implements AWSCredentialsProvider { public YourClassName() { } public AWSCredentials getCredentials() { return new BasicAWSCredentials("key1", "key2"); } public void refresh() { } } ``` このクラスは、引数を取らないコンストラクターを必要とします。 AWS は更新メソッドを定期的に呼び出して、更新された認証情報を取得します。認証情報プロバイダーからその存続期間中に常に異なる認証情報を提供する場合は、このメソッドに認証情報を更新するためのコードを含めます。または、静的な (変わらない) 認証情報を提供する認証情報プロバイダーを必要とする場合は、このメソッドを空のままにすることもできます。 # エージェントをダウンロードおよびインストールする 最初に、インスタンスに接続します。詳細については、「*Amazon EC2 ユーザーガイド*」の「[Linux インスタンスへの接続](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-connect-to-instance-linux.html)」を参照してください。接続できない場合は、「*Amazon EC2 ユーザーガイド*」の「[インスタンスへの接続に関するトラブルシューティング](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/TroubleshootingInstancesConnecting.html)」を参照してください。 次に、次のいずれかの方法を使用して、エージェントをインストールします。 + **Amazon Linux リポジトリからエージェントをセットアップするには** このメソッドは Amazon Linux インスタンスでのみ機能します。以下のコマンドを使用します。 ``` sudo yum install –y aws-kinesis-agent ``` エージェント v 2.0.0 以降が、オペレーティングシステム Amazon Linux 2 (AL2) のコンピュータにインストールされます。このエージェントバージョンでは、Java 1.8 以降が必要です。必要な Java バージョンがまだ存在しない場合は、エージェントのインストールプロセスによってインストールされます。Amazon Linux 2 の詳細については、「[https://aws.amazon.com/amazon-linux-2/](https://aws.amazon.com/amazon-linux-2/)」を参照してください。 + **Amazon S3 リポジトリからエージェントをセットアップするには** このメソッドは、Red Hat Enterprise Linux と Amazon Linux 2 インスタンスでも機能します。これは、パブリックに利用可能なリポジトリからエージェントをインストールするからです。次のコマンドを使用して、エージェントバージョン 2.x.x の最新バージョンをダウンロードしてインストールします。 ``` sudo yum install –y https://s3.amazonaws.com/streaming-data-agent/aws-kinesis-agent-latest.amzn2.noarch.rpm ``` 特定バージョンのエージェントをインストールするには、そのバージョン番号をコマンドで指定します。たとえば、次のコマンドはエージェント v 2.0.1 をインストールします。 ``` sudo yum install –y https://streaming-data-agent.s3.amazonaws.com/aws-kinesis-agent-2.0.1-1.amzn1.noarch.rpm ``` Java 1.7 を持っていて、それをアップグレードしたくない場合は、Java 1.7 と互換性があるエージェントバージョン 1.x.x をダウンロードできます。たとえば、エージェント v1.1.6 をダウンロードするには、次のコマンドを使用します。 ``` sudo yum install –y https://s3.amazonaws.com/streaming-data-agent/aws-kinesis-agent-1.1.6-1.amzn1.noarch.rpm ``` 次のコマンドで最新のエージェントをダウンロードできます。 ``` sudo yum install -y https://s3.amazonaws.com/streaming-data-agent/aws-kinesis-agent-latest.amzn2.noarch.rpm ``` + **GitHub リポジトリからエージェントをセットアップするには** 1. まず、エージェントのバージョンに応じて、必要な Java バージョンがインストールされていることを確認します。 1. 「[awslabs/amazon-kinesis-agent](https://github.com/awslabs/amazon-kinesis-agent)」GitHub リポジトリからエージェントをダウンロードしてください。 1. ダウンロードしたディレクトリまで移動し、次のコマンドを実行してエージェントをインストールします。 ``` sudo ./setup --install ``` + **Docker コンテナにエージェントをセットアップするには** Kinesis Agent は、[amazonlinux](https://docs.aws.amazon.com/AmazonECR/latest/userguide/amazon_linux_container_image.html) コンテナベースを使ってコンテナで実行することもできます。次の Docker ファイルを使用し、`docker build` を実行します。 ``` FROM amazonlinux RUN yum install -y aws-kinesis-agent which findutils COPY agent.json /etc/aws-kinesis/agent.json CMD ["start-aws-kinesis-agent"] ``` # エージェントを設定および開始する **エージェントを設定して開始するには** 1. (デフォルトのファイルアクセス許可を使用している場合、スーパーユーザーとして) 設定ファイル (`/etc/aws-kinesis/agent.json`) を開き、編集します。 この設定ファイルで、エージェントがデータを集めるファイル (`"filePattern"`) とエージェントがデータを送信する Firehose ストリーム (`"deliveryStream"`) を指定します。ファイル名はパターンで、エージェントはファイルローテーションを確認します。1 秒あたりに一度だけ、ファイルを交替または新しいファイルを作成できます。エージェントはファイル作成タイムスタンプを使用して、追跡し、Firehose ストリームに送信するファイルを判断します。新しいファイルを作成したり、1 秒に 1 回を超える頻度でファイルをローテーションしたりしても、エージェント間で適切に区別することはできません。 ``` { "flows": [ { "filePattern": "/tmp/app.log*", "deliveryStream": "yourdeliverystream" } ] } ``` デフォルトの AWS リージョンは です`us-east-1`。別のリージョンを使用する場合は、設定ファイルに `firehose.endpoint` 設定を追加してリージョンのエンドポイントを指定します。詳細については、「[エージェント構成設定を指定する](agent-config-settings.md)」を参照してください。 1. エージェントを手動で開始する: ``` sudo service aws-kinesis-agent start ``` 1. (オプション) システムスタートアップ時にエージェントを開始するように設定します。 ``` sudo chkconfig aws-kinesis-agent on ``` エージェントは、システムのサービスとしてバックグラウンドで実行されます。継続的に指定ファイルをモニタリングし、指定された Firehose ストリームにデータを送信します。エージェント活動は、`/var/log/aws-kinesis-agent/aws-kinesis-agent.log` に記録されます。 # エージェント構成設定を指定する エージェントは、2 つの必須設定、`filePattern` と `deliveryStream`、さらに追加機能としてオプションの設定をサポートしています。必須およびオプションの設定を `/etc/aws-kinesis/agent.json` で指定できます。 設定ファイルを変更した場合は、次のコマンドを使用してエージェントを停止および起動する必要があります。 ``` sudo service aws-kinesis-agent stop sudo service aws-kinesis-agent start ``` または、次のコマンドを使用できます。 ``` sudo service aws-kinesis-agent restart ``` 全般設定は次のとおりです。 | 構成設定 | 説明 | | --- | --- | | assumeRoleARN | ユーザーが引き受けるロールの Amazon リソースネーム (ARN)。詳細については、IAM *ユーザーガイド*の[「IAM ロールを使用した AWS アカウント間のアクセスの委任](https://docs.aws.amazon.com/IAM/latest/UserGuide/tutorial_cross-account-with-roles.html)」を参照してください。 | | assumeRoleExternalId | ロールを引き受けることができるユーザーを決定するオプションの ID。詳細については、*IAM ユーザーガイド*の[外部 ID の使用方法](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-user_externalid.html)を参照してください。 | | awsAccessKeyId | AWS デフォルトの認証情報を上書きする アクセスキー ID。この設定は、他のすべての認証情報プロバイダーに優先されます。 | | awsSecretAccessKey | AWS デフォルトの認証情報を上書きするシークレットキー。この設定は、他のすべての認証情報プロバイダーに優先されます。 | | cloudwatch.emitMetrics | エージェントがメトリクスを CloudWatch に出力できるようにします (true に設定された場合)。 デフォルト: true | | cloudwatch.endpoint | CloudWatch のリージョンのエンドポイントです。 デフォルト: `monitoring.us-east-1.amazonaws.com` | | firehose.endpoint | Amazon Data Firehose のリージョンのエンドポイントです。 デフォルト: `firehose.us-east-1.amazonaws.com` | | sts.endpoint | AWS セキュリティトークンサービスのリージョンエンドポイント。 デフォルト: `https://sts.amazonaws.com` | | userDefinedCredentialsProvider.classname | カスタム認証情報プロバイダーを定義する場合、この設定を使用してその完全修飾クラス名を指定します。クラス名の末尾に .class を含めないでください。 | | userDefinedCredentialsProvider.location | カスタム認証情報プロバイダーを定義する場合、この設定を使用して、カスタム認証情報プロバイダーが含まれている jar の絶対パスを指定します。エージェントは、この jar ファイルを /usr/share/aws-kinesis-agent/lib/ でも検索します。 | フロー設定は次のとおりです。 | 構成設定 | 説明 | | --- | --- | | aggregatedRecordSizeBytes | 1 回のオペレーションでエージェントがレコードを集約し、Firehose ストリームに配置するには、この設定を指定します。エージェントが Firehose ストリームに配置する前の集約レコードのサイズに設定します。 デフォルト: 0 (集約なし) | | dataProcessingOptions | Firehose ストリームに送信される前に解析された各レコードに適用されるの処理オプションの一覧。処理オプションは指定した順序で実行されます。詳細については、「[エージェントを使用してデータを事前処理する](pre-processing.md)」を参照してください。 | | deliveryStream | [必須] Firehose ストリームの名前。 | | filePattern | [必須] エージェントによってモニタリングされる必要があるファイルの glob このパターンに一致するすべてのファイルは、エージェントによって自動的に選択され、モニタリングされます。このパターンに一致するすべてのファイルで、読み取りアクセス許可を `aws-kinesis-agent-user` に付与します。ファイルを含むディレクトリで、読み取りアクセス許可と実行アクセス許可を `aws-kinesis-agent-user` に付与します。 エージェントは、このパターンに一致するファイルをすべて取得します。エージェントが意図しないレコードを取得しないように、このパターンは慎重に選択してください。 | | initialPosition | ファイルの解析が開始される最初の位置。有効な値は、`START_OF_FILE` および `END_OF_FILE` です。 デフォルト: `END_OF_FILE` | | maxBufferAgeMillis | エージェントが Firehose ストリームに送信する前にデータをバッファリングする最大時間 (ミリ秒)。 値の範囲: 1,000～900,000 (1 秒～15 分) デフォルト: 60,000 (1 分) | | maxBufferSizeBytes | エージェントが Firehose ストリームに送信する前にデータをバッファリングする最大サイズ (バイト)。 値の範囲: 1～4,194,304 (4 MB) デフォルト: 4,194,304 (4 MB) | | maxBufferSizeRecords | エージェントが Firehose ストリームに送信する前にデータをバッファリングするレコードの最大数。 値の範囲: 1～500 デフォルト: 500 | | minTimeBetweenFilePollsMillis | エージェントが新しいデータのモニタリング対象ファイルをポーリングし、解析する時間間隔 (ミリ秒単位)。 値の範囲: 1 以上 デフォルト: 100 | | multiLineStartPattern | レコードの開始を識別するパターン。レコードは、パターンに一致する 1 行と、それに続くパターンに一致しない行で構成されます。有効な値は正規表現です。デフォルトでは、ログファイルのそれぞれの改行は 1 つのレコードとして解析されます。 | | skipHeaderLines | モニタリング対象ファイルの始めにエージェントが解析をスキップするの行数。 値の範囲: 0 以上 デフォルト: 0 (ゼロ) | | truncatedRecordTerminator | レコードのサイズが Amazon Data Firehose レコードの許容サイズを超えたときに解析されたレコードを切り捨てるために、エージェントが使用する文字列。(1,000 KB) デフォルト: `'\n'` (改行) | # 複数のファイルディレクトリとストリームを設定する 複数のフロー設定を指定することによって、エージェントが複数のファイルディレクトリを監視し、複数のストリームにデータを送信するように設定できます。次の設定例では、エージェントは 2 つのファイルディレクトリをモニタリングし、それぞれ Kinesis データストリームおよび Firehose ストリームにデータを送信します。Kinesis Data Streams と Amazon Data Firehose に異なるエンドポイントを指定できるため、データストリームと Firehose ストリームが同じリージョンに存在する必要はありません。 ``` { "cloudwatch.emitMetrics": true, "kinesis.endpoint": "https://your/kinesis/endpoint", "firehose.endpoint": "https://your/firehose/endpoint", "flows": [ { "filePattern": "/tmp/app1.log*", "kinesisStream": "yourkinesisstream" }, { "filePattern": "/tmp/app2.log*", "deliveryStream": "yourfirehosedeliverystream" } ] } ``` Amazon Kinesis Data Streams でのエージェントの使用の詳細については、「[Kinesis エージェントを使用した Amazon Kinesis Data Streams への書き込み](https://docs.aws.amazon.com/kinesis/latest/dev/writing-with-agents.html)」を参照してください。 # エージェントを使用してデータを事前処理する エージェントは Firehose ストリームにレコードを送信する前に、モニタリング対象ファイルから解析したレコードを事前処理できます。ファイルフローに `dataProcessingOptions` 設定を追加することで、この機能を有効にできます。処理オプションを 1 つ以上追加することができます。また、指定の順序で実行されます。 エージェントは、次の処理オプションに対応しています。エージェントはオープンソースであるため、処理オプションを開発および拡張できます。[Kinesis エージェント](https://github.com/awslabs/amazon-kinesis-agent)からエージェントをダウンロードできます。処理オプション `SINGLELINE` 改行文字、先頭のスペース、末尾のスペースを削除することで、複数行レコードを単一行レコードに変換します。 ``` { "optionName": "SINGLELINE" } ``` `CSVTOJSON` 区切り形式から JSON 形式にレコードを変換します。 ``` { "optionName": "CSVTOJSON", "customFieldNames": [ "field1", "field2", ... ], "delimiter": "yourdelimiter" } ``` `customFieldNames` [必須] 各 JSON キー値のペアでキーとして使用されるフィールド名。たとえば、`["f1", "f2"]` を指定した場合は、レコード「v1、v2」は `{"f1":"v1","f2":"v2"}` に変換されます。 `delimiter` レコードで区切り記号として使用する文字列。デフォルトはカンマ (,) です。 `LOGTOJSON` ログ形式から JSON 形式にレコードを変換します。サポートされているログ形式は、**Apache Common Log**、**Apache Combined Log**、**Apache Error Log**、および **RFC3164 Syslog** です。 ``` { "optionName": "LOGTOJSON", "logFormat": "logformat", "matchPattern": "yourregexpattern", "customFieldNames": [ "field1", "field2", … ] } ``` `logFormat` [必須] ログエントリ形式。以下の値を指定できます。 + `COMMONAPACHELOG` — Apache Common Log 形式。各ログエントリは、デフォルトで次のパターン`%{host} %{ident} %{authuser} [%{datetime}] \"%{request}\" %{response} %{bytes}`になります。 + `COMBINEDAPACHELOG` — Apache Combined Log 形式。各ログエントリは、デフォルトで次のパターン`%{host} %{ident} %{authuser} [%{datetime}] \"%{request}\" %{response} %{bytes} %{referrer} %{agent}`になります。 + `APACHEERRORLOG` — Apache Error Log 形式。各ログエントリは、デフォルトで次のパターン`[%{timestamp}] [%{module}:%{severity}] [pid %{processid}:tid %{threadid}] [client: %{client}] %{message}`になります。 + `SYSLOG` — FC3164 Syslog 形式。各ログエントリは、デフォルトで次のパターン`%{timestamp} %{hostname} %{program}[%{processid}]: %{message}`になります。 `matchPattern` 指定されたログ形式のデフォルトパターンを上書きします。カスタム形式を使用する場合は、この設定を使用してログエンティティから値を抽出します。`matchPattern` を指定する場合は、`customFieldNames` も指定する必要があります。 `customFieldNames` JSON キー値のペアでキーとして使用されるカスタムフィールド名。`matchPattern` から抽出した値のフィールド名を定義するために、または事前定義されたログ形式のデフォルトのフィールド名を上書きするために、この設定を使用できます。 **Example : LOGTOJSON 設定** JSON形式に変換された Apache Common Log エントリの `LOGTOJSON` 設定の一つの例を次に示します。 ``` { "optionName": "LOGTOJSON", "logFormat": "COMMONAPACHELOG" } ``` 変換前: ``` 64.242.88.10 - - [07/Mar/2004:16:10:02 -0800] "GET /mailman/listinfo/hsdivision HTTP/1.1" 200 6291 ``` 変換後: ``` {"host":"64.242.88.10","ident":null,"authuser":null,"datetime":"07/Mar/2004:16:10:02 -0800","request":"GET /mailman/listinfo/hsdivision HTTP/1.1","response":"200","bytes":"6291"} ``` **Example : カスタムフィールドがある LOGTOJSON 設定** こちらは `LOGTOJSON` 設定の別の例です。 ``` { "optionName": "LOGTOJSON", "logFormat": "COMMONAPACHELOG", "customFieldNames": ["f1", "f2", "f3", "f4", "f5", "f6", "f7"] } ``` この設定では、前の例からの同じ Apache Common Log エントリは、次のように JSON 形式に変換されます。 ``` {"f1":"64.242.88.10","f2":null,"f3":null,"f4":"07/Mar/2004:16:10:02 -0800","f5":"GET /mailman/listinfo/hsdivision HTTP/1.1","f6":"200","f7":"6291"} ``` **Example : Apache Common Log エントリの変換** 次のフロー設定は Apache Common Log エントリを JSON 形式の単一行レコードに変換します。 ``` { "flows": [ { "filePattern": "/tmp/app.log*", "deliveryStream": "my-delivery-stream", "dataProcessingOptions": [ { "optionName": "LOGTOJSON", "logFormat": "COMMONAPACHELOG" } ] } ] } ``` **Example : 複数行レコードの変換** 次のフロー設定は、最初の行が`[SEQUENCE=`で開始している複数行レコードを解析します。各レコードはまず単一行レコードに変換されます。次に、値はタブの区切り記号に基づいたレコードから取得されます。取得された値は指定された `customFieldNames` 値にマッピングされ、JSON 形式の単一行レコードを形成します。 ``` { "flows": [ { "filePattern": "/tmp/app.log*", "deliveryStream": "my-delivery-stream", "multiLineStartPattern": "\\[SEQUENCE=", "dataProcessingOptions": [ { "optionName": "SINGLELINE" }, { "optionName": "CSVTOJSON", "customFieldNames": [ "field1", "field2", "field3" ], "delimiter": "\\t" } ] } ] } ``` **Example : 一致パターンで LOGTOJSON 設定** こちらは、最後のフィールド (バイト) が省略された JSON 形式に変換された Apache Common Log エントリの `LOGTOJSON` 設定の一例です。 ``` { "optionName": "LOGTOJSON", "logFormat": "COMMONAPACHELOG", "matchPattern": "^([\\d.]+) (\\S+) (\\S+) \\[([\\w:/]+\\s[+\\-]\\d{4})\\] \"(.+?)\" (\\d{3})", "customFieldNames": ["host", "ident", "authuser", "datetime", "request", "response"] } ``` 変換前: ``` 123.45.67.89 - - [27/Oct/2000:09:27:09 -0400] "GET /java/javaResources.html HTTP/1.0" 200 ``` 変換後: ``` {"host":"123.45.67.89","ident":null,"authuser":null,"datetime":"27/Oct/2000:09:27:09 -0400","request":"GET /java/javaResources.html HTTP/1.0","response":"200"} ``` # 一般的なエージェント CLI コマンドを使用する 次の表は、 AWS Kinesis エージェントを操作するための一般的なユースケースと対応するコマンドのセットを示しています。 | ユースケース | コマンド | | --- | --- | | システムスタートアップ時にエージェントを自動的に開始する | 
sudo chkconfig aws-kinesis-agent on
| | エージェントのステータスを確認する | 
sudo service aws-kinesis-agent status
| | エージェントを停止する | 
sudo service aws-kinesis-agent stop
| | この場所からエージェントのログファイルを読み取る | 
/var/log/aws-kinesis-agent/aws-kinesis-agent.log
| | エージェントのアンインストール | 
sudo yum remove aws-kinesis-agent
| # Kinesis エージェントから送信する際の問題をトラブルシューティングする この表は、Amazon Kinesis エージェントを使用する際に直面する一般的な問題のトラブルシューティングに関する情報と解決方法を示しています。 | 問題 | ソリューション | | --- | --- | | Kinesis エージェントが Windows で機能しないのはなぜですか? | [Windows 用 Kinesis Agent](https://docs.aws.amazon.com/kinesis-agent-windows/latest/userguide/what-is-kinesis-agent-windows.html) は Linux プラットフォーム用 Kinesis Agent とは異なるソフトウェアです。 | | Kinesis Agent の速度が低下したり、RecordSendErrors が増加したりするのはなぜですか? | 通常、これは Kinesis のスロットリングが原因です。Kinesis Data Streams の `WriteProvisionedThroughputExceeded` メトリクス、または Firehose ストリームの `ThrottledRecords` メトリクスをチェックします。これらのメトリクスが 0 を超えている場合は、ストリームの上限を引き上げる必要があることを意味します。詳細については、「[Kinesis Data Stream limits](https://docs.aws.amazon.com/streams/latest/dev/service-sizes-and-limits.html)」と「[Firehose streams](https://docs.aws.amazon.com/firehose/latest/dev/limits.html)」を参照してください。 スロットリングが原因ではないことがわかったら、Kinesis Agent が大量の小規模ファイルをテーリングするように設定されているかどうかを確認してください。Kinesis Agent が新しいファイルをテーリングするときには遅延が発生するため、少量の大きなファイルをテーリングするようにします。ログファイルを大きなファイルに統合してみてください。 | | java.lang.OutOfMemoryError 例外を解決するにはどうすればよいですか? | これは、Kinesis Agent に、現在のワークロードを処理するための十分なメモリがない場合に発生します。/usr/bin/start-aws-kinesis-agent で JAVA\$1START\$1HEAP と JAVA\$1MAX\$1HEAP を増やしてエージェントを再起動してみてください。 | | IllegalStateException : connection pool shut down 例外を解決するにはどうすればよいですか? | Kinesis エージェントに、現在のワークロードを処理するための十分な接続がないためです。/etc/aws-kinesis/agent.json の一般的なエージェント設定で maxConnections と maxSendingThreads を増やしてみてください。これらのフィールドのデフォルト値は、使用可能なランタイムプロセッサの 12 倍です。高度なエージェント設定については、「[AgentConfiguration.java](https://github.com/awslabs/amazon-kinesis-agent/blob/master/src/com/amazon/kinesis/streaming/agent/config/AgentConfiguration.java)」を参照してください。 | | Kinesis Agent に関する別の問題をデバッグする方法を教えてください。 | DEBUG レベルログは /etc/aws-kinesis/log4j.xml で有効にできます。 | | Kinesis Agent はどのように設定するとよいですか? | maxBufferSizeBytes の値が小さいほど、Kinesis Agent がデータを送信する頻度が高くなります。そのため、レコードの配信時間が短縮されますが、Kinesis への 1 秒あたりのリクエスト数も増えます。 | | Kinesis Agent が重複レコードを送信するのはなぜですか? | これはファイルテーリングの設定ミスが原因です。各 fileFlow’s filePattern が それぞれ 1 つのファイルのみと一致するようにします。また、使用されている logrotate モードが copytruncate モードになっている場合にも発生することがあります。重複を避けるため、モードをデフォルトか作成モードに変更してみてください。重複レコードの処理に関する詳細は、「[Handling Duplicate Records](https://docs.aws.amazon.com/streams/latest/dev/kinesis-record-processor-duplicates.html)」を参照してください。 | # AWS SDK でデータを送信する [Amazon Data Firehose API](https://docs.aws.amazon.com/firehose/latest/APIReference/) を使用して Firehose ストリームにデータを送信するには、[AWS SDK for Java](https://aws.amazon.com/sdk-for-java/)、[.NET](https://aws.amazon.com/sdk-for-net/)、[Node.js](https://aws.amazon.com/sdk-for-javascript/)、[Python](https://aws.amazon.com/sdk-for-python/)、または [Ruby](https://aws.amazon.com/sdk-for-ruby/) を使用します。Amazon Data Firehose を初めて使用する場合、時間を取って「[Amazon Data Firehose とは何ですか?](what-is-this-service.md)」で説明している概念と用語を理解してください。詳細については、「[Amazon Web Services を使用した開発の開始](https://aws.amazon.com/developers/getting-started/)」を参照してください。 これらのサンプルは、すべての例外を確認しているわけではなく、すべてのセキュリティやパフォーマンスの側面を考慮しているわけでもない点で、本稼働環境に使用できるコードを表すものではありません。 Amazon Data Firehose API には、Firehose ストリームにデータを送信するための 2 つのオペレーション ([PutRecord](https://docs.aws.amazon.com/firehose/latest/APIReference/API_PutRecord.html) および [PutRecordBatch](https://docs.aws.amazon.com/firehose/latest/APIReference/API_PutRecordBatch.html)) が用意されています。`PutRecord()` が 1 回の呼び出し内で 1 つのデータレコードを送信し、`PutRecordBatch()` は 1 回の呼び出し内で複数のデータレコードを送信できます。 ## PutRecord を使用した単一の書き込みオペレーション データの出力では、Firehose ストリーム名とバイトバッファリング (<=1000 KB) のみが必要です。Amazon Data Firehose バッチはファイルを Amazon S3 にロードする前に、複数のレコードをバッチ処理するため、レコードの区切り文字を追加することをお勧めします。レコードを一度にひとつずつ Firehose ストリームに出力するには、次のコードを使用します。 ``` PutRecordRequest putRecordRequest = new PutRecordRequest(); putRecordRequest.setDeliveryStreamName(deliveryStreamName); String data = line + "\n"; Record record = new Record().withData(ByteBuffer.wrap(data.getBytes())); putRecordRequest.setRecord(record); // Put record into the DeliveryStream firehoseClient.putRecord(putRecordRequest); ``` コードコンテキストの詳細については、 AWS SDK に含まれているサンプルコードを参照してください。リクエストとレスポンスの構文の詳細については、「[Firehose API Operations](https://docs.aws.amazon.com/firehose/latest/APIReference/API_Operations.html)」の関連するトピックを参照してください。 ## PutRecordBatch を使用したバッチ書き込みオペレーション データの入力では、Firehose ストリーム名とレコードのリストのみが必要です。Amazon Data Firehose バッチはファイルを Amazon S3 にロードする前に、複数のレコードをバッチ処理するため、レコードの区切り文字を追加することをお勧めします。データレコードをバッチで Firehose ストリームに出力するには、次のコードを使用します。 ``` PutRecordBatchRequest putRecordBatchRequest = new PutRecordBatchRequest(); putRecordBatchRequest.setDeliveryStreamName(deliveryStreamName); putRecordBatchRequest.setRecords(recordList); // Put Record Batch records. Max No.Of Records we can put in a // single put record batch request is 500 firehoseClient.putRecordBatch(putRecordBatchRequest); recordList.clear(); ``` コードコンテキストの詳細については、 AWS SDK に含まれているサンプルコードを参照してください。リクエストとレスポンスの構文の詳細については、「[Firehose API Operations](https://docs.aws.amazon.com/firehose/latest/APIReference/API_Operations.html)」の関連するトピックを参照してください。 # CloudWatch Logs を Firehose に送信する CloudWatch Logs イベントは、CloudWatch サブスクリプションフィルターを使用して Firehose に送信できます。詳細については、「[Subscription filters with Amazon Data Firehose](https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/SubscriptionFilters.html#FirehoseExample)」を参照してください。 CloudWatch Logs イベントは、圧縮 gzip 形式で Firehose に送信されます。解凍されたログイベントを Firehose の宛先に配信する場合は、Firehose の解凍機能を使用して CloudWatch Logs を自動的に解凍できます。 **重要** Firehose は現在、Amazon OpenSearch Service の宛先への CloudWatch Logs の配信をサポートしていません。Amazon CloudWatch が複数のログイベントを 1 つの Firehose レコードにまとめても、Amazon OpenSearch Service は 1 つのレコードで複数のログイベントを受け入れることができないためです。代わりの方法として、[CloudWatch Logs で Amazon OpenSearch Service のサブスクリプションフィルターを使用する](https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/CWL_OpenSearch_Stream.html)こともできます。 # CloudWatch Logs を解凍する Firehose を使用して CloudWatch Logs を配信し、解凍されたデータを Firehose ストリームの宛先に配信する場合は、Firehose の[データ形式転換](https://docs.aws.amazon.com/firehose/latest/dev/data-transformation.html) (Parquet、ORC) または[動的パーティショニング](https://docs.aws.amazon.com/firehose/latest/dev/dynamic-partitioning.html)を使用します。Firehose ストリームのために解凍を有効にする必要があります。 解凍を有効にするには AWS マネジメントコンソール、、、 AWS Command Line Interface または AWS SDKsを使用します。 **注記** ストリームで解凍機能を有効にする場合は、そのストリームを CloudWatch Logs サブスクリプションフィルターにのみ使用し、Vended Logs には使用しません。CloudWatch Logs と Vended Logs の両方を取り込むために使用されるストリームで解凍機能を有効にした場合、Firehose への Vended Logs の取り込みは失敗します。この解凍機能は CloudWatch Logs 専用です。 # CloudWatch Logs の解凍後にメッセージを抽出する 解凍を有効にすると、メッセージ抽出も有効にするオプションを使用できます。メッセージ抽出を使用する場合、Firehose は、解凍された CloudWatch Logs レコードから所有者、ロググループ、ログストリームなど、すべてのメタデータを除外し、メッセージフィールド内のコンテンツのみを配信します。データを Splunk の宛先に配信する場合は、Splunk がデータを解析できるように、メッセージ抽出をオンにする必要があります。解凍後の出力例を次に示します (メッセージ抽出あり/なし)。 図 1: メッセージ抽出なしの解凍後の出力例: ``` { "owner": "111111111111", "logGroup": "CloudTrail/logs", "logStream": "111111111111_CloudTrail/logs_us-east-1", "subscriptionFilters": [ "Destination" ], "messageType": "DATA_MESSAGE", "logEvents": [ { "id": "31953106606966983378809025079804211143289615424298221568", "timestamp": 1432826855000, "message": "{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root1\"}" }, { "id": "31953106606966983378809025079804211143289615424298221569", "timestamp": 1432826855000, "message": "{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root2\"}" }, { "id": "31953106606966983378809025079804211143289615424298221570", "timestamp": 1432826855000, "message": "{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root3\"}" } ] } ``` 図 2: メッセージ抽出ありの解凍後の出力例: ``` {"eventVersion":"1.03","userIdentity":{"type":"Root1"} {"eventVersion":"1.03","userIdentity":{"type":"Root2"} {"eventVersion":"1.03","userIdentity":{"type":"Root3"} ``` # コンソールから新しい Firehose ストリームでの解凍を有効にする **を使用して新しい Firehose ストリームで解凍を有効にするには AWS マネジメントコンソール** 1. にサインイン AWS マネジメントコンソール し、[https://console.aws.amazon.com/kinesis](https://console.aws.amazon.com/kinesis) で Kinesis コンソールを開きます。 1. ナビゲーションペインで **[Amazon Data Firehose]** を選択します。 1. **[Firehose ストリームを作成]** を選択します。 1. **[ソースと送信先を選択]** で次のように操作します。 ****ソース**** Firehose ストリームのソース。次のソースのいずれかを選択します: + **Direct PUT** – プロデューサーアプリケーションが直接書き込む Firehose ストリームを作成するときは、こちらを選択します。Firehose で Direct PUT と統合される AWS サービス、エージェント、オープンソースサービスのリストについては、[こちら](create-name.md)のセクションを参照してください。 + **Kinesis ストリーム:** Kinesis データストリームをデータソースとして使用する Firehose ストリームを設定するには、このオプションを選択します。これで、Firehose を使用して、既存の Kinesis データストリームからデータを簡単に読み取り、宛先にロードすることができるようになります。詳細については、「[Writing to Firehose Using Kinesis Data Streams](https://docs.aws.amazon.com/firehose/latest/dev/writing-with-kinesis-streams.html)」を参照してください。 ****送信先**** Firehose ストリームの宛先。次のいずれかを選択します。 + Amazon S3 + Splunk 1. **[Firehose ストリーム名]** で、ストリームの名前を入力します。 1. (オプション) **[レコードを変換]** で次のように操作します。 + **[Amazon CloudWatch Logs からソースレコードを解凍]** セクションで、**[解凍をオンにする]** を選択します。 + 解凍後にメッセージ抽出を使用する場合は、**[メッセージ抽出をオンにする]** を選択します。 # 既存の Firehose ストリームでの解凍を有効にする このセクションでは、既存の Firehose ストリームで解凍を有効にする手順について説明します。Lambda 処理が無効になっているストリームと、Lambda 処理がすでに有効になっているストリームの 2 つのシナリオについて説明します。以下のセクションでは、Lambda 関数の作成または変更、Firehose 設定の更新、組み込みの Firehose 解凍機能を正常に実装するための CloudWatch メトリクスのモニタリングなど、ケース別に手順をステップごとに説明します。 ## Lambda 処理が無効になっている場合の解凍の有効化 Lambda 処理が無効になっている既存の Firehose ストリームで解凍を有効にするには、まず Lambda 処理を有効にする必要があります。この条件は、既存のストリームに対してのみ有効です。次の手順は、Lambda 処理が有効になっていない既存のストリームで解凍を有効にする方法を示しています。 1. Lambda 関数を作成する。ダミーレコードのパススルーを作成することも、この[ブループリント](https://github.com/aws-samples/aws-kinesis-firehose-resources/tree/main/blueprints/kinesis-firehose-cloudwatch-logs-processor)を使用して新しい Lambda 関数を作成することもできます。 1. 現在の Firehose ストリームを更新して Lambda 処理を有効にし、処理用に作成した Lambda 関数を使用します。 1. 新しい Lambda 関数でストリームを更新したら、Firehose コンソールに戻り、解凍を有効にします。 1. ステップ 1 で有効にした Lambda 処理を無効にします。これで、ステップ 1 で作成した関数を削除できます。 ## Lambda 処理が有効になっている場合の解凍の有効化 Lambda 関数を含む Firehose ストリームがすでにある場合、その関数を Firehose の解凍機能に置き換えると解凍を実行できます。続行する前に、Lambda 関数コードが解凍またはメッセージ抽出のみを実行するようになっていることを確認します。Lambda 関数の出力は、[図 1 または図 2](Message_extraction.md) に示す例に類似しているはずです。出力が類似している場合は、次のステップを実行して Lambda 関数を置き換えることができます。 1. 現在の Lambda 関数を、この[ブループリント](https://github.com/aws-samples/aws-kinesis-firehose-resources/tree/main/blueprints/kinesis-firehose-cloudwatch-logs-processor) に置き換えます。新しいブループリント Lambda 関数は、着信データが圧縮されているか、解凍されているかを自動的に検出します。入力データが圧縮されている場合にのみ、解凍が実行されます。 1. 解凍のために、組み込み Firehose オプションを使用して解凍をオンにします。 1. Firehose ストリームのために CloudWatch メトリクスを有効にします (まだ有効になっていない場合)。メトリクス `CloudWatchProcessorLambda_IncomingCompressedData` をモニタリングし、このメトリクスがゼロに変わるまで待機します。これにより、Lambda 関数に送信された入力データがすべて解凍され、Lambda 関数が不要になります。 1. ストリームを解凍する必要がなくなったため、Lambda データ変換を削除します。 # Firehose ストリームでの解凍を無効にする **** を使用してデータストリームの解凍を無効にするには AWS マネジメントコンソール 1. にサインイン AWS マネジメントコンソール し、[https://console.aws.amazon.com/kinesis](https://console.aws.amazon.com/kinesis) で Kinesis コンソールを開きます。 1. ナビゲーションペインで **[Amazon Data Firehose]** を選択します。 1. 編集する Firehose ストリームを選択します。 1. **[Firehose ストリームの詳細]** ページで、**[設定]** タブを選択します。 1. **[レコードを変換および転換]** セクションで、**[編集]** を選択します。 1. **[Amazon CloudWatch Logs からソースレコードを解凍]** で、**[解凍をオンにする]** をクリアし、**[変更を保存]** を選択します。 # Firehose での解凍をトラブルシューティングする 次の表は、Firehose がデータの解凍および処理中にエラーを処理する方法を示しています。これには、エラー S3 バケットへのレコードの配信、エラーのログ記録、メトリクスの出力が含まれます。また、不正なデータプットオペレーションについて返されるエラーメッセージに関しても説明します。 | 問題 | ソリューション | | --- | --- | | 解凍中にエラーが発生した場合、ソースデータはどうなりますか? | Amazon Data Firehose がレコードを解凍できない場合、レコードはそのまま (圧縮形式で) Firehose ストリームの作成時に指定したエラー S3 バケットに配信されます。レコードとともに、配信されるオブジェクトにはエラーコードとエラーメッセージも含まれ、これらのオブジェクトは `decompression-failed` という S3 バケットプレフィックスに配信されます。Firehose は、レコードの解凍に失敗した後も、他のレコードの処理を継続します。 | | 解凍が成功した後に処理パイプラインにエラーが発生した場合、ソースデータはどうなりますか? | 動的パーティショニングやデータ形式転換などの解凍後に、処理ステップにおいて Amazon Data Firehose でエラーが発生した場合、レコードは、Firehose ストリームの作成時に指定したエラー S3 バケットに圧縮形式で配信されます。レコードとともに、配信されるオブジェクトにはエラーコードとエラーメッセージも含まれます。 | | エラーや例外が発生した場合はどのように通知されますか? | 解凍中にエラーまたは例外が発生した場合、CloudWatch Logs を設定すると、Firehose は、エラーメッセージを CloudWatch Logs にログ記録します。さらに、Firehose は、モニタリングできる CloudWatch メトリクスにメトリクスを送信します。また、オプションで、Firehose によって発行されたメトリクスに基づいてアラームを作成することもできます。 | | put オペレーションが CloudWatch Logs から取得されない場合はどうなりますか? | お客様の puts が CloudWatch Logs からのものではない場合、次のエラーメッセージが返されます: 
Put to Firehose failed for AccountId: , FirehoseName:   because the request is not originating from allowed source types.
| | Firehose は解凍機能についてどのようなメトリクスを出力しますか? | Firehose は、あらゆるレコードの解凍についてメトリクスを出力します。DecompressedRecords の失敗もしくは成功、または DecompressedBytes の失敗もしくは成功の数を取得するために、期間 (1 分)、統計 (合計)、日付範囲を選択する必要があります。詳細については、「[CloudWatch Logs の解凍メトリクス](monitoring-with-cloudwatch-metrics.md#decompression-metrics-cw)」を参照してください。 | # CloudWatch Events を Firehose に送信する CloudWatch Events ルールにターゲットを追加することで、Amazon CloudWatch が Firehose ストリームにイベントを送信するよう設定できます。 **既存の Firehose ストリームにイベントを送信するターゲットを CloudWatch Events ルールで作成するには** 1. にサインイン AWS マネジメントコンソール し、[https://console.aws.amazon.com/cloudwatch/](https://console.aws.amazon.com/cloudwatch/) で CloudWatch コンソールを開きます。 1. **[‬ルールを作成]‭** を選択します。 1. **[ステップ 1: ルールを作成]** ページの **[ターゲット]** で、**[ターゲットの追加]**、**[Firehose ストリーム]** の順に選択します。 1. 既存の **Firehose ストリーム**を選択します。 CloudWatch Events ルールの作成の詳細については、「[Amazon CloudWatch Events の開始方法](https://docs.aws.amazon.com/AmazonCloudWatch/latest/events/CWE_GettingStarted.html)」を参照してください。 # Firehose にデータを送信する AWS IoT ように を設定する アクションを追加することで、Firehose ストリームに情報を送信する AWS IoT ように を設定できます。 **既存の Firehose ストリームにイベントを送信するアクションを作成するには** 1. AWS IoT コンソールでルールを作成するときは、**ルールの作成**ページの**「1 つ以上のアクションを設定する**」で、**「アクションの追加**」を選択します。 1. [**Amazon Kinesis Firehose ストリームにメッセージを送信する**] を選択します。 1. **アクションの設定**を選択します。 1. **[ストリーム名]** で、既存の Firehose ストリームを選択します。 1. [**Separator**] には、レコード間に挿入するための区切り記号を選択します。 1. [**IAM role name**] には、既存の IAM ロールを選択するか、または [**Create a new role**] を選択します。 1. **[アクションを追加]** を選択します。 AWS IoT ルールの作成の詳細については、[AWS 「 IoT ルールのチュートリアル](https://docs.aws.amazon.com/iot/latest/developerguide/iot-rules-tutorial.html)」を参照してください。