AWS CDK で Amazon OpenSearch Service のパイプラインは YAML の代わりに JSON で定義できる！

はじめに

この記事は
「Amplify Gen2 × Amazon OpenSearch Service 実践解説 — OSISでDynamoDB連携をCDK構築」
の補足解説記事です。前編で YAML 文字列として示した OSIS パイプライン定義を、JSON オブジェクトで記述する場合の具体例を整理します。

前編では、OSIS（OpenSearch Ingestion Service）のパイプライン本体を YAML 文字列 で定義する公式コードをベースに、目的や注意点を整理しました。

その中で、

• 「パイプライン本体の定義は YAML だけでなく JSON でも指定可能」

と補足しましたが、本記事ではその具体例として、

• TypeScript のオブジェクトで pipelineConfiguration を構築
• JSON.stringify() して pipelineConfigurationBody に渡す

という形で OSIS パイプライン本体を JSON で定義する方法 を、前編の公式コードをベースに紹介します。

前提

• Amplify Gen2（Node.js ベースのバックエンド）
• AWS CDK v2 を利用して OpenSearch / OSIS を構築
• DynamoDB Streams → OSIS → OpenSearch というイベント連携
• 前編記事の構成（ドメイン、IAM ロール、インデックスマッピングなど）は既に理解している前提

本記事では、前編の 「5) OSIS パイプラインのテンプレート定義」 に相当する部分を、YAML ではなく JSON オブジェクトで書き直した実装例を解説します。

なぜ JSON オブジェクトで書くのか

前編で触れた通り、YAML 文字列でパイプラインを定義すると次のような課題があります。

• テンプレートリテラル内でのエスケープやインデントミスでハマりやすい
• 型安全性がなく、プロパティ名の typo をコンパイル時に検知できない
• IDE の補完（IntelliSense）が効かず、定義の見通しが悪い

これに対して、TypeScript のオブジェクトとして定義 → JSON.stringify() で文字列化 する方式にすると、以下のメリットがあります。

• TypeScript の型システムと ESLint などの静的解析の恩恵を受けられる
• オブジェクト構造が明確になり、プロパティ名の typo に気づきやすい
• フィールド追加／削除などの差分が Git 上で追いやすい
• 既存の設定を別パイプラインでも簡単に再利用できる

以降では、前編で掲載した公式の DynamoDB → OpenSearch パイプライン定義 を例に、YAML から JSON 定義への落とし込み方を見ていきます。

公式 YAML コードをおさらい

まずは、前編で紹介した公式の YAML ベースのテンプレートを再掲します。

const indexMapping = {
  settings: {
    number_of_shards: 1,
    number_of_replicas: 0,
  },
  mappings: {
    properties: {
      id: {
        type: "keyword",
      },
      isDone: {
        type: "boolean",
      },
      content: {
        type: "text",
      },
      priority: {
        type: "text",
      },
    },
  },
};

const openSearchTemplate = `
version: "2"
dynamodb-pipeline:
  source:
    dynamodb:
      acknowledgments: true
      tables:
        - table_arn: "${tableArn}"
          stream:
            start_position: "LATEST"
          export:
            s3_bucket: "${s3BucketName}"
            s3_region: "${backend.storage.stack.region}"
            s3_prefix: "${tableName}/"
      aws:
        sts_role_arn: "${openSearchIntegrationPipelineRole.roleArn}"
        region: "${backend.data.stack.region}"
  sink:
    - opensearch:
        hosts:
          - "https://${openSearchDomain.domainEndpoint}"
        index: "${indexName}"
        index_type: "custom"
        template_content: |
          ${JSON.stringify(indexMapping)}
        document_id: '\${getMetadata("primary_key")}'
        action: '\${getMetadata("opensearch_action")}'
        document_version: '\${getMetadata("document_version")}'
        document_version_type: "external"
        bulk_size: 4
        aws:
          sts_role_arn: "${openSearchIntegrationPipelineRole.roleArn}"
          region: "${backend.data.stack.region}"
`;

• indexMapping 自体はすでにオブジェクトですが、パイプライン本体は YAML 文字列として openSearchTemplate に埋め込まれています。
• ここから、dynamodb-pipeline.source や sink[0].opensearch の構造を そのまま JSON オブジェクトに起こす のが本記事のゴールです。

パイプライン本体を JSON オブジェクトにする

上記 YAML の構造を、そのまま pipelineConfiguration というオブジェクトに起こすと、概ね次のようになります。

const pipelineConfiguration = {
  version: '2',
  'dynamodb-pipeline': {
    source: {
      dynamodb: {
        acknowledgments: true,
        tables: [
          {
            table_arn: tableArn,
            stream: {
              start_position: 'LATEST',
            },
            export: {
              s3_bucket: s3BucketName,
              s3_region: backend.storage.stack.region,
              s3_prefix: `${tableName}/`,
            },
          },
        ],
        aws: {
          sts_role_arn: openSearchIntegrationPipelineRole.roleArn,
          region: backend.data.stack.region,
        },
      },
    },
    sink: [
      {
        opensearch: {
          hosts: [`https://${openSearchDomain.domainEndpoint}`],
          index: indexName,
          index_type: 'custom',
          // indexMapping はそのまま再利用し、ここで JSON 文字列にする
          template_content: JSON.stringify(indexMapping),
          document_id: '${getMetadata("primary_key")}',
          action: '${getMetadata("opensearch_action")}',
          document_version: '${getMetadata("document_version")}',
          document_version_type: 'external',
          bulk_size: 4,
          aws: {
            sts_role_arn: openSearchIntegrationPipelineRole.roleArn,
            region: backend.data.stack.region,
          },
        },
      },
    ],
  },
}

この定義のポイント

• YAML のトップレベルキー version / dynamodb-pipeline は、そのままオブジェクトのキーにしています。
• tables / stream.start_position / export.s3_bucket なども YAML と 1:1 で対応させています。
• template_content だけは YAML でも JSON.stringify(indexMapping) を埋め込んでいたので、そのまま JSON 側でも JSON.stringify(indexMapping) を呼び出しています。
• document_id / action / document_version は、YAML と同じ文字列（'${getMetadata("primary_key")}' など）で表現します。
• この pipelineConfiguration は、実際に動作確認した構成を簡略化したものであり、そのまま JSON.stringify() して pipelineConfigurationBody に渡す形で問題なく動作します。

このように、「YAML のインデント構造＝JSON オブジェクトのネスト」と見立てて機械的に写すと、差分も追いやすくなります。

CDK から JSON をそのまま渡す

あとは、この pipelineConfiguration を osis.CfnPipeline に渡すだけです。
YAML 版では pipelineConfigurationBody: openSearchTemplate となっていた部分を、JSON 版では次のように書き換えます。

const cfnPipeline = new osis.CfnPipeline(
  backend.data.stack,
  'OpenSearchIntegrationPipeline',
  {
    maxUnits: 4,
    minUnits: 1,
    // ✅ ここで JSON 文字列に変換して渡す
    pipelineConfigurationBody: JSON.stringify(pipelineConfiguration),
    pipelineName: 'dynamodb-integration-2',
    logPublishingOptions: {
      isLoggingEnabled: true,
      cloudWatchLogDestination: {
        logGroup: logGroup.logGroupName,
      },
    },
  },
)

• それ以外のプロパティ（minUnits / maxUnits / pipelineName / logPublishingOptions）は、公式コードそのままです。
• 差分としては、「YAML 文字列を渡していた箇所を、JSON オブジェクトを JSON.stringify() したものに差し替えた」だけになります。

JSON 定義にすることで得られたメリット

実際に YAML 版から JSON 版へ移行してみて、特に効果が大きかったポイントは次の通りです。

• 型安全性の向上
  • TypeScript のオブジェクトなので、プロパティ名の typo や構造の不整合をエディタやビルド時に検知しやすくなりました。
• リファクタリングが容易
  • パイプライン定義が 1 つのオブジェクトにまとまっているため、今後必要に応じて一部を関数に切り出したり、複数パイプライン間で共通化したりしやすくなります。
• 環境変数・パラメータの埋め込みがシンプル
  • 引数に region / domainEndpoint / roleArn / tableArn / indexName を渡すだけでよく、YAML の ${} 展開やエスケープを意識する必要がなくなりました。
• 差分レビューがしやすい
  • JSON オブジェクトの変更は通常の TypeScript コードの diff と同じ感覚でレビューでき、設定変更の意図が伝わりやすくなりました。

OSIS 自体は YAML でも JSON でもパイプラインを受け付けてくれますが、「アプリケーションコードから扱う」 という観点では JSON オブジェクトで管理する方がトータルでは扱いやすいと感じています。

移行時の注意点

YAML から JSON へ書き換える際は、次の点に注意すると安全です。

• 構造を 1:1 で写経する
  • まずは YAML の階層構造をそのままオブジェクトに置き換え、意味が変わらないことを意識します。
• メタデータ参照の式をそのまま移す
  • document_id / action / document_version の式は YAML と同じ文字列で記述します（'${/primary_key}' など）。
• CDK のトークンを意識する
  • domain.domainEndpoint や role.roleArn のような CDK トークンは、JSON.stringify() 時に CloudFormation の組み込み関数へ展開されます。基本的にはそのままで問題ありませんが、テンプレートサイズやネストの制限には一応注意しておくと安心です。

おわりに

• OSIS のパイプライン本体は、YAML 文字列ではなく JSON オブジェクトで定義しても問題なく動作 します。
• 本記事では、前編の公式 YAML テンプレートを 純粋な TypeScript オブジェクト に起こし直し、JSON.stringify() で pipelineConfigurationBody に渡す形を示しました。
• この方式により、型安全性・可読性・再利用性が大きく向上し、複数パイプラインを持つ構成でも設定管理がしやすくなります。

前編の記事で紹介した YAML 版のパイプラインをお使いの場合は、まずは一つのパイプラインだけでも JSON オブジェクトでの定義に置き換えてみると、運用時の変更やレビューがかなり楽になるはずです。