すべてのプロダクト
Search

このプロダクト

    Simple Log Service:Flume によるデータ消費

    ドキュメントセンター

    Simple Log Service:Flume によるデータ消費

    最終更新日:Jun 16, 2026

    aliyun-log-flume プラグインを使用して Log Service と Apache Flume を統合し、ログデータの書き込みと消費を行います。

    背景情報

    aliyun-log-flume は、Log Service と Apache Flume を接続するプラグインであり、Log Service と HDFS や Kafka などの他のシステムとの間のデータ交換を可能にします。このプラグインは、シンクとソースを提供します。

    • シンク：Flume は他のデータソースからデータを読み取り、Log Service に書き込みます。

    • ソース：Flume は Log Service からログデータを消費し、他のシステムに書き込みます。

    詳細については、「aliyun-log-flume」をご参照ください。

    手順

    1. Apache Flume をダウンロードしてインストールします。

      詳細については、「Apache Flume」をご参照ください。

    2. aliyun-log-flume プラグインをダウンロードし、 /flume/lib ディレクトリに配置します。

      詳細については、「aliyun-log-flume-1.3.jar」をご参照ください。

    3. /flume/conf ディレクトリに、flumejob.conf という名前の設定ファイルを作成します。

      • シンクの設定と例については、「シンク」をご参照ください。

      • ソースの設定と例については、「ソース」をご参照ください。

    4. Flume を起動します。

    シンク

    シンクは、他のデータソースから Log Service にデータを書き込み、2 つの解析形式をサポートします。

    • SIMPLE：Flume イベント全体を単一のフィールドとして Log Service に書き込みます。

    • DELIMITED：Flume イベントを区切り文字形式のデータとして扱い、設定されたカラム名に基づいてフィールドに解析し、Log Service に書き込みます。

    シンクのパラメーターを次の表に示します。

    パラメーター

    必須

    説明

    type

    はい

    シンクのタイプです。このパラメーターを com.aliyun.loghub.flume.sink.LoghubSink に設定します。

    endpoint

    はい

    Log Service プロジェクトのエンドポイントです。例：http://cn-qingdao.log.aliyuncs.com。この値は、実際のエンドポイントに置き換えてください。詳細については、「エンドポイント」をご参照ください。

    project

    はい

    Log Service プロジェクトの名前です。

    Logstore

    はい

    Logstore の名前です。

    accessKeyId

    はい

    Alibaba Cloud アカウントの AccessKey ID です。RAM ユーザーの AccessKey ペアを使用することを推奨します。詳細については、「AccessKeyの作成」をご参照ください。

    accessKey

    はい

    Alibaba Cloud アカウントの AccessKey Secret です。RAM ユーザーの AccessKey ペアを使用することを推奨します。詳細については、「AccessKeyの作成」をご参照ください。

    batchSize

    いいえ

    各バッチで Log Service に書き込むデータエントリの数です。デフォルト値：1000。

    maxBufferSize

    いいえ

    バッファーキューのサイズです。デフォルト値：1,000 エントリ。

    serializer

    いいえ

    イベントのシリアル化形式です。有効な値：

    • DELIMITED：シリアル化形式を区切り文字モードに設定します。

    • SIMPLE：シリアル化形式をシンプルモードに設定します。これがデフォルトのモードです。

    • JSON：シリアル化形式を JSON モードに設定します。

    • カスタム シリアライザー：カスタムシリアル化クラスの完全修飾名です。

    columns

    いいえ

    このパラメーターは、 serializer が DELIMITED に設定されている場合に必須です。コンマ (,) で区切られたカラム名のリストを指定します。カラムの順序は、データ内の順序と一致している必要があります。

    separatorChar

    いいえ

    このパラメーターは、 serializer が DELIMITED に設定されている場合に使用します。データの区切り文字を指定します。区切り文字は単一の文字である必要があります。デフォルト値：コンマ (,)。

    quoteChar

    いいえ

    このパラメーターは、 serializer が DELIMITED に設定されている場合に使用します。引用符を指定します。デフォルト値：二重引用符 (")。

    escapeChar

    いいえ

    このパラメーターは、 serializer が DELIMITED に設定されている場合に使用します。エスケープ文字を指定します。デフォルト値：バックスラッシュ (\)。

    useRecordTime

    いいえ

    データの timestamp フィールドをログ時間として使用するかどうかを指定します。デフォルト値は false であり、その場合は現在の時刻が使用されます。

    シンクの設定例については、「GitHub」をご参照ください。

    ソース

    ソースは Log Service からログデータを消費し、他のシステムに送信します。2 つの出力形式をサポートしています。

    • DELIMITED：データを区切り文字形式で Flume に書き込みます。

    • JSON：データを JSON 形式で Flume に書き込みます。

    ソースのパラメーターを次の表に示します。

    パラメーター

    必須

    説明

    type

    はい

    ソースのタイプです。このパラメーターを com.aliyun.loghub.flume.source.LoghubSource に設定します。

    endpoint

    はい

    Log Service プロジェクトのエンドポイントです。例：http://cn-qingdao.log.aliyuncs.com。この値は、実際のエンドポイントに置き換えてください。詳細については、「エンドポイント」をご参照ください。

    project

    はい

    Log Service プロジェクトの名前です。

    Logstore

    はい

    Logstore の名前です。

    accessKeyId

    はい

    Alibaba Cloud アカウントの AccessKey ID です。RAM ユーザーの AccessKey ペアを使用することを推奨します。詳細については、「AccessKeyの作成」をご参照ください。

    accessKey

    はい

    Alibaba Cloud アカウントの AccessKey Secret です。RAM ユーザーの AccessKey ペアを使用することを推奨します。詳細については、「AccessKeyの作成」をご参照ください。

    heartbeatIntervalMs

    いいえ

    クライアントと Log Service 間のハートビート間隔です。デフォルト値：30,000 ms。

    fetchIntervalMs

    いいえ

    データ取得間隔です。デフォルト値：100 ms。

    fetchInOrder

    いいえ

    データを順次消費するかどうかを指定します。デフォルト値：false。

    batchSize

    いいえ

    各バッチで読み取るデータエントリの数です。デフォルト値：100。

    consumerGroup

    いいえ

    データの読み取りに使用されるコンシューマーグループの名前です。

    initialPosition

    いいえ

    データの読み取りを開始する初期位置です。有効な値：begin、end、timestamp。デフォルト値：begin。

    説明

    サーバー側にチェックポイントが存在する場合、それが優先されます。

    timestamp

    いいえ

    このパラメーターは、 initialPosition が timestamp に設定されている場合に必須です。UNIX タイムスタンプ形式で時刻を指定します。

    deserializer

    はい

    イベントのデシリアル化形式です。有効な値：

    • DELIMITED：デシリアル化形式を区切り文字モードに設定します。これがデフォルトのモードです。

    • JSON：デシリアル化形式を JSON モードに設定します。

    • カスタム デシリアライザー：カスタムデシリアル化クラスの完全修飾名です。

    columns

    いいえ

    このパラメーターは、 deserializer が DELIMITED に設定されている場合に必須です。コンマ (,) で区切られたカラム名のリストを指定します。カラムの順序は、データ内の順序と一致している必要があります。

    separatorChar

    いいえ

    このパラメーターは、 deserializer が DELIMITED に設定されている場合に使用します。データの区切り文字を指定します。区切り文字は単一の文字である必要があります。デフォルト値：コンマ (,)。

    quoteChar

    いいえ

    このパラメーターは、 deserializer が DELIMITED に設定されている場合に使用します。引用符を指定します。デフォルト値：二重引用符 (")。

    escapeChar

    いいえ

    このパラメーターは、 deserializer が DELIMITED に設定されている場合に使用します。エスケープ文字を指定します。デフォルト値：バックスラッシュ (\)。

    appendTimestamp

    いいえ

    このパラメーターは、 deserializer が DELIMITED に設定されている場合に使用します。各行の末尾にタイムスタンプをカラムとして自動的に追加するかどうかを指定します。デフォルト値：false。

    sourceAsField

    いいえ

    このパラメーターは、 deserializer が JSON に設定されている場合に使用します。ログソースを __source__ という名前のフィールドとして追加するかどうかを指定します。デフォルト値：false。

    tagAsField

    いいえ

    このパラメーターは、 deserializer が JSON に設定されている場合に使用します。ログタグをフィールドとして追加するかどうかを指定します。フィールド名は __tag__:{tag_name} の形式になります。デフォルト値：false。

    timeAsField

    いいえ

    このパラメーターは、 deserializer が JSON に設定されている場合に使用します。ログ時間を __time__ という名前のフィールドとして追加するかどうかを指定します。デフォルト値：false。

    useRecordTime

    いいえ

    Log Service からの元のログ時間を Flume イベントのタイムスタンプとして使用するかどうかを指定します。false の場合、現在のシステム時刻が使用されます。デフォルト値は false です。

    ソースの設定例については、「GitHub」をご参照ください。