Extended Log File Format Source Connector for Confluent Platform¶
このコネクターは、拡張ログファイルフォーマット のファイルを、ディレクトリからストリーミングしながらデータを厳密に型指定されたスキーマに変換するために使用されます。
このコネクターを使用するには、connector.class
構成プロパティでこのコネクタークラスの名前を指定するコネクター構成を使用します。
connector.class=com.github.jcustenborder.kafka.connect.spooldir.SpoolDirELFSourceConnector
コネクター固有の他の構成プロパティについて、以降で説明します。
構成プロパティ¶
全般¶
topic
データの書き込み先 Kafka トピック。
- Importance: HIGH
- Type: STRING
batch.size
各バッチで返されるレコード数。
- Importance: LOW
- 型: INT
- デフォルト値: 1000
empty.poll.wait.ms
ポーリングで空のレコードリストが返された場合の待ち時間。
- Importance: LOW
- Type: LONG
- デフォルト値: 500
- バリデーター: [1,…,9223372036854775807]
メタデータ¶
metadata.field
値においてメタデータが格納されるフィールドの名前。
- Importance: LOW
- Type: STRING
- デフォルト値: metadata
metadata.location
入力ファイルに関するメタデータの格納先。FIELD - 入力ファイルに関するメタデータは、レコードの値に含まれるフィールドに格納されます。HEADERS - 入力ファイルに関するメタデータは、レコードのヘッダーとして格納されます。NONE - 入力ファイルに関するメタデータは格納されません。
- Importance: LOW
- Type: STRING
- デフォルト値: HEADERS
- 指定可能な値:
NONE
、HEADERS
、FIELD
トピックの自動作成¶
トピックの自動作成の詳細については、「ソースコネクターのトピックの自動作成の構成 」を参照してください。
注釈
構成プロパティには、Java regex として定義されている正規表現(regex)を使用できます。
topic.creation.groups
一致するトピックにグループ別のトピック構成を定義するために使用するグループ別名のリスト。
default
グループは常に存在し、すべてのトピックに一致します。- 型: String 型のリスト
- デフォルト: 空
- 指定可能な値: このプロパティの値には、任意の追加グループを指定できます。トピックの構成には常に
default
グループが定義されています。
topic.creation.$alias.replication.factor
コネクターで作成する新規トピックのレプリケーション係数。この値は、Kafka クラスターのブローカーの数を超えてはなりません。この値が Kafka ブローカーの数よりも大きい場合、コネクターがトピックの作成を試行するとエラーが発生します。
default
グループの場合、これは 必須のプロパティ です。topic.creation.groups
で定義されている他のグループの場合、このプロパティは省略可能です。他のグループでは、Kafka ブローカーのデフォルト値が使用されます。- 型: int
- デフォルト: なし
- 指定可能な値 : 具体的な有効値を指定する場合は
>= 1
で指定し、Kafka ブローカーのデフォルト値を使用する場合は-1
を指定します。
topic.creation.$alias.partitions
このコネクターによって作成されるトピックのパーティションの数。
default
グループの場合、これは 必須のプロパティ です。topic.creation.groups
で定義されている他のグループの場合、このプロパティは省略可能です。他のグループでは、Kafka ブローカーのデフォルト値が使用されます。- 型: int
- デフォルト: なし
- 指定可能な値 : 具体的な有効値を指定する場合は
>= 1
で指定し、Kafka ブローカーのデフォルト値を使用する場合は-1
を指定します。
topic.creation.$alias.include
トピック名に一致する正規表現を表す文字列のリスト。このリストは、一致する値を持つトピックを対象に含め、このグループの特定の構成を一致するトピックに適用するために使用します。
topic.creation.groups
で定義されているすべてのグループに、$alias
が適用されます。default
グループには、このプロパティは適用されません。- 型: String 型のリスト
- デフォルト: 空
- 指定可能な値: 正確なトピック名または正規表現のコンマ区切りのリスト。
topic.creation.$alias.exclude
トピック名に一致する正規表現を表す文字列のリスト。このリストは、一致する値を持つトピックを、グループの特定の構成の適用から除外するために使用します。
topic.creation.groups
で定義されているすべてのグループに、$alias
が適用されます。default
グループには、このプロパティは適用されません。トピックの除外ルールは、すべての包含ルールをオーバーライドすることに注意してください。- 型: String 型のリスト
- デフォルト: 空
- 指定可能な値: 正確なトピック名または正規表現のコンマ区切りのリスト。
topic.creation.$alias.${kafkaTopicSpecificConfigName}
レコードの書き込み先の Kafka ブローカーのバージョンに対するすべての ブローカー構成の動的な変更 を行います。ルールに対して構成が指定されていない場合、ブローカーのトピックレベルの構成値が使用されます。
default
グループ、およびtopic.creation.groups
で定義されているすべてのグループに、$alias
が適用されます。- 型: プロパティ値
- デフォルト : Kafka ブローカー値
ファイルシステム¶
error.path
エラーを含むファイルを配置するディレクトリ。このディレクトリが存在し、Kafka Connect を実行しているユーザーによる書き込みが可能である必要があります。
- Importance: HIGH
- Type: STRING
- バリデーター: 存在し、書き込み可能であるディレクトリへの絶対パス。
input.file.pattern
入力ファイル名と照合する正規表現。ファイル名全体と一致する正規表現を使用してください。
Matcher.matches()
と同等。有効な構文の定義については、「Class Pattern」を参照してください。- Importance: HIGH
- Type: STRING
input.path
Kafka Connect が処理対象のファイルを読み取るディレクトリ。このディレクトリが存在し、Connect を実行しているユーザーによる書き込みが可能である必要があります。
- Importance: HIGH
- Type: STRING
- バリデーター: 存在し、書き込み可能であるディレクトリへの絶対パス。
finished.path
Connect が正常に処理されたファイルを配置するディレクトリ。このディレクトリが存在し、Connect を実行しているユーザーによる書き込みが可能である必要があります。
- Importance: HIGH
- Type: STRING
halt.on.error
エラー発生時に、タスクを中止するか次のファイルに進むかを設定します。
- Importance: HIGH
- Type: BOOLEAN
- デフォルト値: true
cleanup.policy
正常に処理されたファイルをコネクターでクリーンアップする方法を決定します。NONE を指定するとファイルがそのまま残されますが、コネクターが再起動された場合に再処理される原因になる可能性があります。DELETE を指定すると、ファイルシステムからファイルが削除されます。MOVE を指定すると、完了したファイル用のディレクトリにファイルが移動されます。MOVEBYDATE を指定すると、完了したファイル用のディレクトリに日付ごとのサブディレクトリが作成されてファイルが移動されます。
- Importance: MEDIUM
- Type: STRING
- デフォルト値: MOVE
- 指定可能な値:
NONE
、DELETE
、MOVE
、MOVEBYDATE
task.partitioner
複数のタスクを使用するようコネクターを構成した場合は、タスクパーティショナーの実装が使用されます。タスクで処理されるファイルを特定するために各タスクで使用されます。これにより、各ファイルの割り当て先タスクが 1 つに限定されます。
- Importance: MEDIUM
- Type: STRING
- デフォルト値: ByName
- バリデーター : 一致文字列:
ByName
file.buffer.size.bytes
BufferedInputStream のバッファサイズ。ファイルシステムとのやり取りで使用されます。
- Importance: LOW
- 型: INT
- デフォルト値: 131072
- バリデーター: [1,…]
file.minimum.age.ms
ファイルへの最後の書き込みが終わってからファイルを処理できるようになるまでの待ち時間(ミリ秒)。
- Importance: LOW
- Type: LONG
- デフォルト値: 0
- バリデーター: [0,…]
files.sort.attributes
各ファイルによってソート順序の決定に使用される属性。Name はファイルの名前です。Length はファイルの長さで、長いファイルほど先になります。LastModified はファイルの LastModified 属性で、古いファイルほど先になります。
- Importance: LOW
- Type: LIST
- デフォルト値: [NameAsc]
- バリデーター : 一致文字列:
NameAsc
、NameDesc
、LengthAsc
、LengthDesc
、LastModifiedAsc
、LastModifiedDesc
processing.file.extension
ファイルは、処理される前に、現在処理されていることを示す名前に変更されます。この設定は、ファイルの末尾に追加されます。
- Importance: LOW
- Type: STRING
- デフォルト値: .PROCESSING
- バリデーター : 一致文字列: regex( ^.*..+$ )
スキーマ¶
key.schema
Kafka に書き込まれるキーのスキーマ。
- Importance: HIGH
- Type: STRING
value.schema
Kafka に書き込まれる値のスキーマ。
- Importance: HIGH
- Type: STRING
スキーマ生成¶
schema.generation.enabled
スキーマが動的に生成されるかどうかを決定するフラグ。true を設定した場合は、
key.schema
とvalue.schema
を省略できますが、schema.generation.key.name
とschema.generation.value.name
は設定する必要があります。- Importance: MEDIUM
- Type: BOOLEAN
schema.generation.key.fields
キースキーマのビルドに使用されるフィールド。これは、スキーマ生成時にのみ使用されます。
- Importance: MEDIUM
- Type: LIST
schema.generation.key.name
生成されるキースキーマの名前。
- Importance: MEDIUM
- Type: STRING
- デフォルト値: com.github.jcustenborder.kafka.connect.model.Key
schema.generation.value.name
生成される値スキーマの名前。
- Importance: MEDIUM
- Type: STRING
- デフォルト値: com.github.jcustenborder.kafka.connect.model.Value
timestamp.field
値スキーマ内で、解析対象となるレコードのタイムスタンプを含むフィールド。このフィールドは、省略可能とすることはできません。また、タイムスタンプ であることが必要です。
- Importance: MEDIUM
- Type: STRING
タイムスタンプ¶
timestamp.mode
コネクターで ConnectRecord にタイムスタンプがどのように設定されるかを決定します。
Field
を設定した場合、タイムスタンプは値のフィールドから読み取られます。このフィールドは、省略可能とすることはできません。また、タイムスタンプ であることが必要です。フィールドは、timestamp.field
で指定します。FILE_TIME
を設定した場合、ファイルの最終更新時刻が使用されます。PROCESS_TIME
(デフォルト)を設定した場合、レコードが読み取られた時刻が使用されます。- Importance: MEDIUM
- Type: STRING
- デフォルト値: PROCESS_TIME
- 指定可能な値:
FIELD
、FILE_TIME
、PROCESS_TIME
timestamp.field
値スキーマ内で、解析対象となるレコードのタイムスタンプを含むフィールド。このフィールドは、省略可能とすることはできません。また、タイムスタンプ であることが必要です。
- Importance: MEDIUM
- Type: STRING
parser.timestamp.date.formats
ファイルで想定される日付フォーマット。これは、日付フィールドを順番に解析するために使用される文字列のリストです。最も正確な日付フォーマットをリストの先頭にする必要があります。詳細については、Java のドキュメント を参照してください。
- Importance: LOW
- Type: LIST
- デフォルト値 : [yyyy-MM-dd'T'HH:mm:ss、yyyy-MM-dd' 'HH:mm:ss]
parser.timestamp.timezone
すべての解析済み日付に使用するタイムゾーン。
- Importance: LOW
- Type: STRING
- デフォルト値: UTC