トピックの構成¶
このトピックでは、Confluent Platform で使用できる構成パラメーターについて説明します。パラメーターは、重要度の高いものから順に並べられています。
-
cleanup.policy
"delete" または "compact" あるいは両方の文字列。この文字列は、古いログセグメントに使用する保持ポリシーを指定します。デフォルトのポリシー("delete")では、保持時間またはサイズ制限に達したときに古いセグメントを破棄します。"compact" 設定は、トピックでのログ圧縮を有効にします。
型: リスト デフォルト: delete 指定可能な値: [compact、delete] サーバーのデフォルトプロパティ: log.cleanup.policy 重要度: 中 -
compression.type
特定のトピックの最終的な圧縮タイプを指定します。この構成には、標準の圧縮コーデック(「gzip」、「snappy」、「lz4」、「zstd」)を指定できます。他に使用できる値として、「uncompressed」を指定すると圧縮なしと同等になり、「producer」を指定するとプロデューサーで設定された元の圧縮コーデックが維持されます。
型: string デフォルト: プロデューサー 指定可能な値: [uncompressed、zstd、lz4、snappy、gzip、producer] サーバーのデフォルトプロパティ: compression.type 重要度: 中 -
confluent.key.schema.validation
レコードキーでのスキーマ検証がこのトピックで有効になっている場合、true です。
型: boolean デフォルト: false 指定可能な値: サーバーのデフォルトプロパティ: null 重要度: 中 -
confluent.key.subject.name.strategy
キースキーマをスキーマレジストリに登録する際に使用されるサブジェクト名の構築方法を決定します。デフォルトでは TopicNameStrategy が使用されます。
型: string デフォルト: io.confluent.kafka.serializers.subject.TopicNameStrategy 指定可能な値: サーバーのデフォルトプロパティ: null 重要度: 中 -
confluent.tier.enable
トピックの階層化を有効にします。これにより、構成されたリモートストレージとの間で、データの階層化とフェッチが有効になります。
型: boolean デフォルト: false 指定可能な値: サーバーのデフォルトプロパティ: confluent.tier.enable 重要度: 中 -
confluent.tier.local.hotset.bytes
階層化が有効になっている場合、この構成は、ブローカーローカルストレージで(ログセグメントから構成される)パーティションを拡張できる最大サイズを制御します。このサイズに達すると、空き領域を増やすために古いログセグメントが破棄されます。ブローカーローカルストレージに保持されるログセグメントは、"ホットセット" と呼ばれます。ローカルストアから破棄されたセグメントは、保持の構成によっては引き続き階層型ストレージに存在し、フェッチ可能な状態のまま残る可能性があります。デフォルトではサイズの制限はなく、時間の制限があるだけです。この制限はパーティションレベルで適用されるため、トピックのホットセットのバイト数を計算するには、パーティション数を乗算する必要があります。
型: long デフォルト: -1 指定可能な値: サーバーのデフォルトプロパティ: confluent.tier.local.hotset.bytes 重要度: 中 -
confluent.tier.local.hotset.ms
階層化が有効になっている場合、この構成は、ブローカーローカルストレージにログセグメントを保持する最大時間を制御します。この時間に達すると、空き領域を増やすために古いログセグメントが破棄されます。ローカルストアから破棄されたセグメントは、保持の構成によっては引き続き階層型ストレージに存在し、フェッチ可能な状態のまま残る可能性があります。-1 に設定されている場合、時間制限は適用されません。
型: long デフォルト: 86400000(1 日) 指定可能な値: サーバーのデフォルトプロパティ: confluent.tier.local.hotset.ms 重要度: 中 -
confluent.value.schema.validation
レコード値でのスキーマ検証がこのトピックで有効になっている場合、true です。
型: boolean デフォルト: false 指定可能な値: サーバーのデフォルトプロパティ: null 重要度: 中 -
confluent.value.subject.name.strategy
値スキーマをスキーマレジストリに登録する際に使用されるサブジェクト名の構築方法を決定します。デフォルトでは TopicNameStrategy が使用されます。
型: string デフォルト: io.confluent.kafka.serializers.subject.TopicNameStrategy 指定可能な値: サーバーのデフォルトプロパティ: null 重要度: 中 -
delete.retention.ms
ログ圧縮トピックの delete tombstone マーカーを保持する時間。この設定でも、オフセット 0 から始める場合に、コンシューマーが読み取りを完了する必要がある時間を制限します。これによって、最終段階の有効なスナップショットを確実に取得できます(そうでない場合、スキャンを完了する前に delete tombstone が収集される可能性があります)。
型: long デフォルト: 86400000(1 日) 指定可能な値: [0,...] サーバーのデフォルトプロパティ: log.cleaner.delete.retention.ms 重要度: 中 -
file.delete.delay.ms
ファイルシステムからファイルを削除するまでの待機時間
型: long デフォルト: 60000(1 分) 指定可能な値: [0,...] サーバーのデフォルトプロパティ: log.segment.delete.delay.ms 重要度: 中 -
flush.messages
この設定により、ログに書き込まれたデータの fsync を強制的に実行する間隔を指定できます。たとえば、これを 1 に設定した場合、メッセージ 1 件ごとに fsync を実行します。これが 5 の場合、メッセージ 5 件ごとに fsync を実行します。一般的には、これを設定せず、永続性のためのレプリケーションを使用して、より効率的なオペレーティングシステムのバックグランドフラッシュ機能を許可することをお勧めします。この設定は、トピックごとにオーバーライドできます(トピックごとの構成に関するセクションを参照)。
型: long デフォルト: 9223372036854775807 指定可能な値: [0,...] サーバーのデフォルトプロパティ: log.flush.interval.messages 重要度: 中 -
flush.ms
この設定により、ログに書き込まれたデータの fsync を強制的に実行する時間の間隔を指定できます。たとえば、これを 1000 に設定した場合、1000 ミリ秒が経過した後で fsync を実行します。一般的には、これを設定せず、永続性のためのレプリケーションを使用して、より効率的なオペレーティングシステムのバックグランドフラッシュ機能を許可することをお勧めします。
型: long デフォルト: 9223372036854775807 指定可能な値: [0,...] サーバーのデフォルトプロパティ: log.flush.interval.ms 重要度: 中 -
follower.replication.throttled.replicas
ログレプリケーションをフォロワー側でスロットルする必要があるレプリカのリスト。このリストには、レプリカのセットを [PartitionId]:[BrokerId],[PartitionId]:[BrokerId]:... の形式で指定する必要があります。または、ワイルドカード「*」を使用して、このトピックのすべてのレプリカをスロットルすることもできます。
型: リスト デフォルト: "" 指定可能な値: [partitionId]:[brokerId],[partitionId]:[brokerId],... サーバーのデフォルトプロパティ: follower.replication.throttled.replicas 重要度: 中 -
index.interval.bytes
この設定は、Kafka がインデックスエントリをオフセットインデックスに追加する頻度を制御します。デフォルト設定では、おおよそ 4096 バイトごとにメッセージのインデックスを作成します。インデックス作成をさらに頻繁にすると、ログ内で読み取りがジャンプする位置がさらに正確になりますが、インデックスのサイズが大きくなります。おそらく、これを変更する必要はないでしょう。
型: int デフォルト: 4096(4 キビバイト) 指定可能な値: [0,...] サーバーのデフォルトプロパティ: log.index.interval.bytes 重要度: 中 -
leader.replication.throttled.replicas
ログレプリケーションをリーダー側でスロットルする必要があるレプリカのリスト。このリストには、レプリカのセットを [PartitionId]:[BrokerId],[PartitionId]:[BrokerId]:... の形式で指定する必要があります。または、ワイルドカード「*」を使用して、このトピックのすべてのレプリカをスロットルすることもできます。
型: リスト デフォルト: "" 指定可能な値: [partitionId]:[brokerId],[partitionId]:[brokerId],... サーバーのデフォルトプロパティ: leader.replication.throttled.replicas 重要度: 中 -
max.compaction.lag.ms
ログ内でメッセージが圧縮対象外として保持される最大時間。圧縮中のログにのみ適用されます。
型: long デフォルト: 9223372036854775807 指定可能な値: [1,...] サーバーのデフォルトプロパティ: log.cleaner.max.compaction.lag.ms 重要度: 中 -
max.message.bytes
Kafka で許容される最大レコードバッチサイズ(圧縮が有効な場合は圧縮後のサイズを表します)。0.10.2 よりも前のコンシューマーが存在する環境でこの値を大きくする場合は、それだけの大きさのレコードバッチをフェッチできるように、コンシューマーのフェッチサイズも大きくする必要があります。最新バージョンのメッセージフォーマットでは、効率を上げるためにレコードが常にバッチにグループ化されます。前のメッセージフォーマットバージョンでは非圧縮のレコードはバッチにグループ化されず、その場合、この制限は 1 つのレコードにのみ適用されます。
型: int デフォルト: 1048588 指定可能な値: [0,...] サーバーのデフォルトプロパティ: message.max.bytes 重要度: 中 -
message.format.version
ブローカーがメッセージをログに追加するときに使用するメッセージフォーマットバージョンを指定します。値は有効な ApiVersion であることが必要です。例として、0.8.2、0.9.0.0、0.10.0 などがあります。詳細については ApiVersion を確認してください。特定のメッセージフォーマットバージョンを設定すると、ディスク上にあるすべての既存のメッセージが、指定のバージョン以前であると保証することになります。この値の設定が正しくない場合、以前のバージョンを使用するコンシューマーは、受け取ったメッセージのフォーマットに対応していないために動作しなくなります。
型: string デフォルト: 2.6-IV0 指定可能な値: [0.8.0, 0.8.1, 0.8.2, 0.9.0, 0.10.0-IV0, 0.10.0-IV1, 0.10.1-IV0, 0.10.1-IV1, 0.10.1-IV2, 0.10.2-IV0, 0.11.0-IV0, 0.11.0-IV1, 0.11.0-IV2, 1.0-IV0, 1.1-IV0, 2.0-IV0, 2.0-IV1, 2.1-IV0, 2.1-IV1, 2.1-IV2, 2.2-IV0, 2.2-IV1, 2.3-IV0, 2.3-IV1, 2.4-IV0, 2.4-IV1, 2.5-IV0, 2.6-IV0] サーバーのデフォルトプロパティ: log.message.format.version 重要度: 中 -
message.timestamp.difference.max.ms
ブローカーがメッセージを受信した時点のタイムスタンプと、メッセージ内に指定されたタイムスタンプとの間で許容される最大差異。message.timestamp.type=CreateTime の場合、タイムスタンプの相違がこのしきい値を超えると、メッセージは拒否されます。この構成は、message.timestamp.type=LogAppendTime の場合、無視されます。
型: long デフォルト: 9223372036854775807 指定可能な値: [0,...] サーバーのデフォルトプロパティ: log.message.timestamp.difference.max.ms 重要度: 中 -
message.timestamp.type
メッセージのタイムスタンプが、メッセージの作成時刻とログの追加時刻のどちらであるかを示します。使用できる値は `CreateTime` または `LogAppendTime` です。
型: string デフォルト: CreateTime 指定可能な値: [CreateTime、LogAppendTime] サーバーのデフォルトプロパティ: log.message.timestamp.type 重要度: 中 -
min.cleanable.dirty.ratio
この構成では、ログコンパクターがログのクリーニングを試行する頻度を制御します(ログ圧縮が有効になっていることが前提)。デフォルトでは、ログの 50% 以上が圧縮されている場合、ログのクリーニングは回避されます。この割合によって、重複により無駄になるログの最大スペースを制限します(50% の場合、ログの最大 50% まで重複にすることができます)。割合が高くなるほど、クリーニングの回数が減り、その効率が向上しますが、ログで無駄になるスペースは増えます。max.compaction.lag.ms 構成または min.compaction.lag.ms 構成も指定した場合、次のいずれかの状態になった時点で、ログコンパクターはそのログが圧縮の対象であるとみなします。(i) ダーティの比率のしきい値に達し、ログがダーティな(圧縮されていない)レコードを保持していた期間が min.compaction.lag.ms 以上である場合、または (ii) ログがダーティな(圧縮されていない)レコードを保持していた期間が max.compaction.lag.ms 以内である場合。
型: double デフォルト: 0.5 指定可能な値: [0,...,1] サーバーのデフォルトプロパティ: log.cleaner.min.cleanable.ratio 重要度: 中 -
min.compaction.lag.ms
ログ内でメッセージが圧縮されずに保持される最小時間。圧縮中のログにのみ適用されます。
型: long デフォルト: 0 指定可能な値: [0,...] サーバーのデフォルトプロパティ: log.cleaner.min.compaction.lag.ms 重要度: 中 -
min.insync.replicas
プロデューサーが acks を "all"(または "-1")に設定した場合、この構成パラメーターは、書き込みが正常に行われたとみなされるために、その書き込みを確認しなければならないレプリカの最小数を指定します。この最小数に達しない場合、プロデューサーは例外(NotEnoughReplicas または NotEnoughReplicasAfterAppend)を発します。
min.insync.replicas
とacks
を併用すると、より大きな永続性保証を適用できます。一般的には、トピックの作成時にレプリケーション係数を 3、min.insync.replicas
を 2 に設定し、生成時のacks
を "all" にします。これにより、大部分のレプリカが書き込みを受け取らなかった場合に、確実にプロデューサーから例外が生成されます。型: int デフォルト: 1 指定可能な値: [1,...] サーバーのデフォルトプロパティ: min.insync.replicas 重要度: 中 -
preallocate
新しいログセグメントの作成時にファイルをディスクに事前割り当てする必要がある場合は、true です。
型: boolean デフォルト: false 指定可能な値: サーバーのデフォルトプロパティ: log.preallocate 重要度: 中 -
retention.bytes
この構成は、"delete" 保持ポリシーを使用している場合に、古いログセグメントを破棄してスペースを空ける前に、(ログセグメントから構成される)パーティションが到達できる最大サイズを制御します。デフォルトではサイズの制限はなく、時間の制限があるだけです。この制限はパーティションレベルで適用されるため、これにパーティションの数を掛けてトピック保持サイズを計算します(単位: バイト)。
型: long デフォルト: -1 指定可能な値: サーバーのデフォルトプロパティ: log.retention.bytes 重要度: 中 -
retention.ms
この構成は、"delete" 保持ポリシーを使用している場合に、古いログセグメントを破棄してスペースを空ける前に、ログを保持する最長時間を制御します。これは、コンシューマーがいかに早くデータを読み取る必要があるかに関する SLA に該当します。-1 に設定されている場合、時間制限は適用されません。
型: long デフォルト: 604800000(7 日) 指定可能な値: [-1,...] サーバーのデフォルトプロパティ: log.retention.ms 重要度: 中 -
segment.bytes
この構成はログのセグメントファイルサイズを制御します。保持およびクリーニングは常に 1 回 1 ファイルずつ行われるため、セグメントサイズを大きくするとファイルの数は少なくなりますが、保持を細かく制御できなくなります。
型: int デフォルト: 1073741824(1 ギビバイト) 指定可能な値: [14,...] サーバーのデフォルトプロパティ: log.segment.bytes 重要度: 中 -
segment.index.bytes
この構成は、オフセットをファイルの位置にマップするインデックスのサイズを制御します。このインデックスファイルは事前割り当てし、縮小するのはログのロール後のみです。通常は、この設定を変更する必要はありません。
型: int デフォルト: 10485760(10 メビバイト) 指定可能な値: [0,...] サーバーのデフォルトプロパティ: log.index.size.max.bytes 重要度: 中 -
segment.jitter.ms
セグメントローリングの thundering herd を回避するために、スケジュール設定されたセグメントロール時間から減算する最大ランダムジッター。
型: long デフォルト: 0 指定可能な値: [0,...] サーバーのデフォルトプロパティ: log.roll.jitter.ms 重要度: 中 -
segment.ms
この構成は、保持で古いデータを確実に削除または圧縮できるように、セグメントファイルが完全でなくても Kafka がログのロールを強制するまでの時間を制御します。
型: long デフォルト: 604800000(7 日) 指定可能な値: [1,...] サーバーのデフォルトプロパティ: log.roll.ms 重要度: 中 -
unclean.leader.election.enable
最終手段として、ISR セットにないレプリカをリーダーに選出できるようにするかどうかを示します。ただしこの場合、データ喪失の可能性があります。
型: boolean デフォルト: false 指定可能な値: サーバーのデフォルトプロパティ: unclean.leader.election.enable 重要度: 中 -
confluent.placement.constraints
この構成は、ISR への参加が常に許可されるブローカーのセット(レプリカ)を制御する JSON オブジェクトです。また、ISR への参加が常に許可されないブローカーのセット(オブザーバー)も制御できます。JSON の形式は次のとおりです。
{
"version": 1,
"replicas": [
{
"count": 2,
"constraints": {"rack": "east-1"}
},
{
"count": 1,
"constraints": {"rack": "east-2"}
}
],
"observers":[
{
"count": 1,
"constraints": {"rack": "west-1"}
}
]
}型: string デフォルト: "" 指定可能な値: kafka.common.TopicPlacement$TopicPlacementValidator@4c15e7fd サーバーのデフォルトプロパティ: null 重要度: 低 -
message.downconversion.enable
この構成では、コンシューマーリクエストを満たすために、メッセージフォーマットのダウンコンバートを有効にするかどうかを制御します。
false
にすると、コンシューマーが以前のメッセージフォーマットを想定していても、ブローカーはダウンコンバートを実行しません。ブローカーは、このような以前のクライアントからのコンシューマーリクエストに対してUNSUPPORTED_VERSION
エラーで応答します。この構成は、レプリケーションで準拠する必要のあるメッセージフォーマット規則には適用されません。型: boolean デフォルト: true 指定可能な値: サーバーのデフォルトプロパティ: log.message.downconversion.enable 重要度: 低
注釈
このウェブサイトには、Apache License v2 の条件に基づいて Apache Software Foundation で開発されたコンテンツが含まれています。