スキーマを指定する

この使用例では、MongoDB Kafka ソースコネクタを構成して、カスタムスキーマをデータに適用する方法を示します。スキーマは、Apache Kafka トピック内のデータに関する構造とタイプ情報を指定する定義です。ソースコネクタによって設定されるトピックのデータが一貫した構造になっているようにする必要がある場合は、スキーマを使用します。

コネクタでスキーマを使用する方法の詳細については、「スキーマの適用」ガイドを参照してください。

例

アプリケーションが MongoDB コレクション内のカスタマーデータを追跡し、このデータを Kafka トピックに公開するとします。カスタマーデータのサブスクライブに、一貫した形式のデータを受信したい。データにスキーマを適用することを選択します。

要件とソリューションは次のとおりです。

要件	解決法
MongoDB コレクションからカスタマーデータを受信する	Configure a MongoDB source connector to receive updates to data from a specific database and collection. See Receive Data from a Collection.
カスタマーデータスキーマを提供する	Specify a schema that corresponds to the structure and data types of the customer data. See Create a Custom Schema.
カスタマーデータから Kafka メタデータを省略	Include only the data from the `fullDocument` field. See Omit Metadata from Published Records.

For the full configuration file that meets the requirements above, see Specify the Configuration.

コレクションからのデータの受信

MongoDB コレクションからデータを受信するようにソースコネクタを構成するには、データベース名とコレクション名を指定します。この例では、次のように、 customersデータベース内のpurchasesコレクションから読み取るようにコネクタを構成できます。

database=customers
collection=purchases

カスタムスキーマの作成

コレクションのサンプルカスタマーデータドキュメントには、次の情報が含まれています。

{
  "name": "Zola",
  "visits": [
    {
      "$date": "2021-07-25T17:30:00.000Z"
    },
    {
      "$date": "2021-10-03T14:06:00.000Z"
    }
  ],
  "goods_purchased": {
    "apples": 1,
    "bananas": 10
  }
}

サンプルドキュメントから、スキーマは次のデータ型を使用してフィールドを表示する必要があると判断します。

フィールド名	データ型	説明
name	string	顧客名
訪問	タイムスタンプの配列	カスタマーが訪問した日付
product_price	string（想定された型）を整数値にマップ	顧客が購入した商品の名前と数量

以下のサンプルスキーマに示すように、Apache Avro スキーマ形式を使用してデータを記述できます。

{
  "type": "record",
  "name": "Customer",
  "fields": [{
      "name": "name",
      "type": "string"
    },{
      "name": "visits",
      "type": {
        "type": "array",
        "items": {
          "type": "long",
          "logicalType": "timestamp-millis"
        }
      }
    },{
      "name": "goods_purchased",
      "type": {
        "type": "map",
        "values": "int"
      }
    }
  ]
}

重要

変換子

Atlas のバイナリエンコーディングで Apache Kafka を介してデータを送信する場合は、Avro 変換を使用する必要があります。詳細については、変換に関するガイドを参照してください。

公開レコードからメタデータを省略する

connectorは、カスタマーデータドキュメントと、そのドキュメントを記述するメタデータを Kafka トピックに公開します。次の設定を使用して、レコードのfullDocumentフィールドに含まれるドキュメントデータのみを含めるように connector を設定できます。

publish.full.document.only=true

fullDocumentフィールドの詳細については、 Change Streamsのガイドを参照してください。

構成を指定する

カスタムスキーマコネクターの構成は次のようになります。

connector.class=com.mongodb.kafka.connect.MongoSourceConnector
connection.uri=<your MongoDB connection URI>
database=customers
collection=purchases
publish.full.document.only=true
output.format.value=schema
output.schema.value={\"type\": \"record\", \"name\": \"Customer\", \"fields\": [{\"name\": \"name\", \"type\": \"string\"}, {\"name\": \"visits\", \"type\": {\"type\": \"array\", \"items\": {\"type\": \"long\", \"logicalType\": \"timestamp-millis\"}}}, {\"name\": \"goods_purchased\", \"type\": {\"type\": \"map\", \"values\": \"int\"}}]}
value.converter.schemas.enable=true
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter=org.apache.kafka.connect.storage.StringConverter

注意

埋め込みスキーマ

上記の構成では、 Kafka Connect JSON schema 変換はカスタムスキーマをメッセージに埋め込みます。 JSON schema 変換の詳細については、変換のガイドを参照してください。

スキーマの指定の詳細については、「スキーマの適用」ガイドを参照してください。

戻る

既存のデータのコピー

変更ストリーム

スキーマを指定する

例

コレクションからのデータの受信

カスタム スキーマの作成

公開レコードからメタデータを省略する

構成を指定する

カスタムスキーマの作成