Sink Connector ポストプロセッサ

カスタムドキュメント ID 戦略の作成

組み込みのドキュメント ID 追加戦略でユースケースをカバーしない場合は、以下の手順に従ってカスタムのドキュメント ID 戦略を定義できます。

1. インターフェース IdStratey を実装し、カスタム構成ロジックを含むJavaクラスを作成します。

2. クラスを JAR ファイルにコンパイルします。

3. コンパイルされた JAR を、すべてのKafkaワーカーのクラスパスまたはプラグインパスに追加します。プラグインパスの詳細については、 Confluent のドキュメントを参照してください。

4. すべての Kafka ワーカーのカスタム クラスの完全なクラス名にdocument.id.strategy設定を更新します。

IdStrategyインターフェースの実装例については、 コネクタにパッケージ化された ID 戦略の実装を含むソースコードディレクトリを参照してください。

許可リスト ワイルドカードの例

複数のフィールド名を一致させるには、 *ワイルドカードを使用します。 次のサンプル構成では、次のフィールドが一致しています。

• 最上位のフィールドには「City」

• "wind_step" という名前で始まる最上位フィールドのサブドキュメントである "average" という名前のフィールド。

post.processor.chain=com.mongodb.kafka.connect.sink.processor.AllowListValueProjector
value.projection.type=AllowList
value.projection.list=city,wind_speed*.average

ポストプロセッサが許可リストプロジェクションを適用した後、次のレコードを出力します。

{
  "city": "Springfield",
  "wind_speed_10m": {
    "average": 3,
  },
  "wind_speed_80m": {
    "average": 8,
  }
}

ワイルドカードを指定したレベルから任意のレベルのオブジェクトに一致する**ワイルドカードを使用できます。 次のワイルドカードに一致する例では、「low」という名前のフィールドを含む任意のドキュメントをプロジェクションします。

post.processor.chain=com.mongodb.kafka.connect.sink.processor.AllowListValueProjector
value.projection.type=AllowList
value.projection.list=**.low

プロジェクションを適用する ポストプロセッサ は、次のレコードを出力します。

{
  "temperature": {
    "high": 28,
    "low": 24,
    "units": "C"
  },
  "soil_conditions": {
    "temperature": {
      "high": 22,
      "low": 17,
      "units": "C"
    }
  }
}

ブロック リスト ワイルドカードの例

次のブロックリスト構成例に示すように、ワイルドカード パターンを使用して、特定のドキュメント レベルのフィールドを一致させることができます。

post.processor.chain=com.mongodb.kafka.connect.sink.processor.BlockListValueProjector
value.projection.type=BlockList
value.projection.list=*.*.temperature

{
  "city": "Springfield",
  "temperature": {
    "high": 28,
    "low": 24,
    "units": "C"
  },
  "wind_speed_10m": {
    "average": 3,
    "units": "km/h"
  },
  "wind_speed_80m": {
    "average": 8,
    "units": "km/h"
  },
  "soil_conditions": {
    "moisture": {
      "average": 340,
      "units": "mm"
    }
  }
}

マッピング例による名前の変更

RenameByMappingポストプロセッサ 設定では、string に一致するフィールドを新しい名前に割り当てる 1 つ以上の JSON オブジェクトを指定します。 各オブジェクトには、以下の表に示すように、 oldName要素に一致するテキストと、 newName要素内の置換テキストが含まれています。

次のサンプル プロパティは、キー ドキュメントの「location」フィールドを照合し、その名前を「country」に変更します。

field.renamer.mapping=[{"oldName":"key.location", "newName":"country"}]

この設定は、 RenameByMappingポストプロセッサ に元のキー ドキュメントを次のドキュメントに変換するように指示します。

{
  "country": "Provence",
  "date_month": "October",
  "date_day": 17
}

次のように、 oldNameフィールドにフィールド名が追加された値のドキュメントを指定することで、値のドキュメントでも同様のフィールド名の割り当てを実行できます。

field.renamer.mapping=[{"oldName":"value.flapjacks", "newName":"crepes"}]

この設定は、 RenameByMappingポストプロセッサ に、元の値ドキュメントを次のドキュメントに変換するように指示します。

{
  "crepes": {
    "purchased": 598,
    "size": "large"
  }
}

次の設定に示すように、string 形式の JSON 配列を使用して、 field.renamer.mappingプロパティで 1 つ以上のマッピングを指定することもできます。

field.renamer.mapping=[{ "oldName":"key.location", "newName":"city" }, { "oldName":"value.crepes", "newName":"flapjacks" }]

正規表現による名前の変更

RenameByRegexポストプロセッサ 設定では、一致する必要があるフィールド名とテキスト パターンと、一致したテキストの置換値を指定します。 次の表で説明されているフィールドを含む JSON オブジェクトに、1 つ以上の名前変更式を指定できます。

次の例の設定では、ポストプロセッサに次の処理を実行するように指示します。

• キー ドキュメント内の「date」で始まる任意のフィールド名と一致します。 一致するフィールドのセットで、パターン_に一致するすべてのテキストを-文字に置き換えます。

• crepesのサブドキュメントである、値ドキュメント内の任意のフィールド名と一致します。 一致するフィールドのセットで、パターンpurchasedに一致するすべてのテキストをquantityに置き換えます。

field.renamer.regexp=[{"regexp":"^key\\.date.*$","pattern":"_","replace":"-"}, {"regexp":"^value\\.crepes\\..*","pattern":"purchased","replace":"quantity"}]

コネクタがサンプル キー ドキュメントとサンプル 値ドキュメントに ポストプロセッサ を適用すると、次の内容を出力します。

キー ドキュメント

{
  "location": "Provence",
  "date-month": "October",
  "date-day": 17
}

値ドキュメント

{
  "crepes": {
    "quantity": 598,
    "size": "large"
  }
}