---
sourceDocument: Yokohama データと自動化
sourceDocumentLink: https://servicenow-prod.fluidtopics.net/r/ja-JP/yokohama/integrate-applications

 Release :

    - yokohama

ft:locale :

    - ja-JP

ft:publication_title :

    - Yokohama データと自動化

ft:clusterId :

    - crint

bundleId :

    - crint

workflow :

    - Creator


---

# ETL および変換マップコンシューマーを使用したデータのインポート

# ETL および変換マップコンシューマーを使用したデータのインポート {#ariaid-title1}

* リリースバージョン: Yokohama
* 
* 更新日 2025年01月30日
* 
* ![](https://www.servicenow.com/docs/portal-asset/ico-clock) 所要時間：7分

ETL 定義と変換マップを使用して、Kafka イベントを処理します。ETL 定義と変換マップは、スケジュール済みインポートを通じてデータを取得するときに使用する変換ロジックを指定します。同じ ETL 定義と変換マップを使用して、Kafka を介して受信したイベントを変換できます。

## 始める前に

* 必要なロール： integration_hub_admin
* このページは [ストリームコネクト クイックスタートガイド](https://servicenow-prod.fluidtopics.net/zZ~~4QfzQrdZE1mHsEGHQA "このクイックスタートガイドでは、ServiceNow インスタンスと Kafka 環境の間でデータを交換するために、6 つのストリームコネクトプロデューサーとコンシューマをそれぞれ設定して使用する方法の概要について説明します。")の一部です。
* このドキュメントでは、Hermes への接続を介してメッセージを消費する方法について説明します。接続の設定とテストについては、「[Hermes メッセージングサービス への安全な接続の設定](https://servicenow-prod.fluidtopics.net/Wtz6jsvqc~7S8pILnM8hPA "ServiceNow インスタンス署名証明書を生成して Kafka トピックを保護します。")」および「[ADCv2 ゲートウェイを介した Hermes への接続のテスト](https://servicenow-prod.fluidtopics.net/XLA5cKYynip6vJjNsbuQKQ "ADCv2 ゲートウェイを介して Hermes への接続をテストすることで、キーストアとトラストストアが正しく生成されていることを確認します。")」を参照してください。
* このページの前半では、ETL コンシューマーを使用してメッセージを消費する方法を説明します。ETL コンシューマーを使用するには、[強力なインポートセットトランスフォーマー](https://servicenow-prod.fluidtopics.net/S5xYYwkowPeE73VrERe5~w "データを抽出、変換して 1 つ以上のターゲットテーブルにロードする場合は、変換マップの代わりに強力なインポートセットトランスフォーマーを使用します。")が必要です。
* このページの後半では、変換マップコンシューマーを使用してメッセージを消費する方法を説明します。変換マップコンシューマーを使用するには、[変換マップ](https://servicenow-prod.fluidtopics.net/RPXxiBEruFYNmW1p2od7Yw "変換マップは、インポートセットのフィールドと、インシデント [incident] やユーザー [sys_user] などの既存の ServiceNow テーブルのフィールドとの関係を決定するフィールドマップのセットです。")が必要です。
{#use-etl-transform-map-consumers__ul_gm5_njx_cyb}

## 手順

1. ETL および変換マップのコンシューマーメッセージを消費するトピックを作成します。
   1. 次のコマンドを使用して、Hermes にトピックを作成します。  
      この例では、トピック名は topic2 です。

          bin/kafka-topics.sh --create --command-config config/bootcamp.properties  --bootstrap-server <instance name>.service-now.com:4000,<instance name>.service-now.com:4001,<instance name>.service-now.com:4002, <instance name>.service-now.com:4003 --topic snc.<instance name>.topic2

      必ず `<instance name>` をインスタンス名で置き換えてください。
   2. トピックを表示するには、インスタンスにログインし、 すべて統合ハブストリームコネクトトピック.  
      注:  
      トピックの作成には約 10 分かかります。
   {#use-etl-transform-map-consumers__substeps_lv3_ylx_cyb}
2. トピックからデータを消費する ETL コンシューマーを作成します。
   1. 移動先 すべて統合ハブコンシューマーETL コンシューマー.
   2. \[新規\] を選択します。
   3. Kafka コンシューマーフォームで、\[名前\] フィールドに名前を入力します。  
      この例では、Member Import という名前を使用します。
   4. \[強力なインポートセットトランスフォーマー\] フィールドで、既存の強力なインポートセットトランスフォーマーを選択します。  
      強力なインポートセットトランスフォーマーの場合は、データを単一列に格納するオプションを使用していないものを選択します。メッセージを表示できるようにするには、強力なインポートセットトランスフォーマーで \[詳細 \] オプションが有効になっていることを確認してください。
   5. \[単一列のデータ\] オプションの選択を解除します。  
      このオプションは、単一列のデータを含むインポートセットテーブルに対してのみ選択する必要があります。
   6. \[列マッピング\] を \[ラベル\] に設定します。  
      これは、入力メッセージの JSON キーがインポートセットテーブルの列ラベルにマッピングされることを意味します。  
   7. フォームを保存します。
   8. コンシューマー用に [Kafka ストリームの作成](https://servicenow-prod.fluidtopics.net/bfgmjteqZTCKAC4n1qfTRA "抽出変換ロード (ETL)、変換マップ、スクリプトコンシューマーのデータストリームを定義します。Kafka ストリームは、コンシューマーへのデータのストリームを定義します。") を行います。
   9. \[Kafka ストリーム\] フォームで、次のフィールド値を設定します。  
      * \[名前\] を「<kbd class="ph userinput">Topic2 Stream</kbd>」を設定します。
      * \[トピック\] を前に作成したトピック topic2 に設定します。
      * \[最大同時実行\] を「<kbd class="ph userinput">1</kbd>」に設定します。
      * \[消費を開始\] を \[最初の有効化後に受信したメッセージ (Messages received after the first activation)\] に設定します。
      * \[メッセージ処理 (Message handling)\] を \[自動的に最適化 (Automatically optimize)\] に設定します。
      {#use-etl-transform-map-consumers__ul_pbp_rqm_2yb}
   10. \[関連リンク\] の下で \[アクティブ化\] リンクを選択します。  
       アクティブ化されたサブスクリプションを確認できるはずです。  
   {#use-etl-transform-map-consumers__substeps_ajg_hkx_cyb}
3. topic2 にメッセージを公開します。
   1. 次のコマンドを実行して、topic2 にメッセージを公開します。  

          bin/kafka-console-producer.sh --topic snc.<instance name>.topic2  --producer.config  config/bootcamp.properties  --bootstrap-server <instance name>.service-now.com:4000,<instance name>.service-now.com:4001,<instance name>.service-now.com:4002,<instance name>.service-now.com:4003

      必ず `<instance name>` をインスタンス名で置き換えてください。

      このコマンドはメッセージの入力を促すものです。
   2. 次の JSON メッセージを送信します。  

          {"city":"San Diego","name":"Jhon","id":"SN001","state":"California"}

      このメッセージは任意の値で送信できますが、JSON メッセージキーはインポートセットテーブルのラベルと一致する必要があります。
   3. 作成した ETL コンシューマーに戻ります。  
      約 1 分後、インポートセットを介してデータがインポートされているはずです。  
   4. インポートセットに関する詳細情報を表示するには、インポートセット番号を選択します。  
   5. インポートセット行に関する詳細情報を表示するには、インポートセット行番号を選択します。  
   {#use-etl-transform-map-consumers__substeps_uzy_4kx_cyb}
4. データを消費する変換マップコンシューマーを作成します。
   1. 移動先 すべて統合ハブコンシューマー変換マップコンシューマー.
   2. \[新規\] を選択します。
   3. \[Kafka 変換マップコンシューマー\] フォームで、\[名前\] フィールドに名前を入力します。
   4. \[変換マップ\] フィールドで、既存の変換マップを選択します。  
   5. フォームを保存します。
   6. コンシューマー用に [Kafka ストリームの作成](https://servicenow-prod.fluidtopics.net/bfgmjteqZTCKAC4n1qfTRA "抽出変換ロード (ETL)、変換マップ、スクリプトコンシューマーのデータストリームを定義します。Kafka ストリームは、コンシューマーへのデータのストリームを定義します。") を行います。
   7. \[Kafka ストリーム\] フォームで、次のフィールド値を設定します。  
      * \[名前\] を「<kbd class="ph userinput">変換マップストリーム</kbd>」に設定します。
      * \[トピック\] を前に作成したトピック topic2 に設定します。
      * \[最大同時実行\] を「<kbd class="ph userinput">1</kbd>」に設定します。
      * \[消費を開始\] を \[最初の有効化後に受信したメッセージ (Messages received after the first activation)\] に設定します。
      * \[メッセージ処理 (Message handling)\] を \[自動的に最適化 (Automatically optimize)\] に設定します。
      {#use-etl-transform-map-consumers__ul_u4q_1wm_2yb}
   8. \[関連リンク\] の下で \[アクティブ化\] リンクを選択します。  
      アクティブ化されたサブスクリプションを確認できるはずです。  
   {#use-etl-transform-map-consumers__substeps_yt5_vkx_cyb}
5. topic2 にメッセージを公開します。
   1. 次のコマンドを実行して、topic2 にメッセージを公開します。  

          bin/kafka-console-producer.sh --topic snc.<instance name>.topic2  --producer.config  config/bootcamp.properties  --bootstrap-server <instance name>.service-now.com:4000,<instance name>.service-now.com:4001,<instance name>.service-now.com:4002,<instance name>.service-now.com:4003

      必ず `<instance name>` をインスタンス名で置き換えてください。

      このコマンドはメッセージの入力を促すものです。
   2. 次の JSON メッセージを送信します。  

          {"city":"San Diego","name":"Jhon","id":"SN001","state":"California"}

      このメッセージは任意の値で送信できますが、JSON メッセージキーはインポートセットテーブルのラベルと一致する必要があります。
   3. 作成した変換マップコンシューマーに移動します。  
      約 1 分後、インポートセットを介してデータがインポートされているはずです。
   4. インポートセットに関する詳細情報を表示するには、上記ステップ 3d と同様にインポートセット番号を選択します。
   {#use-etl-transform-map-consumers__substeps_t4b_clx_cyb}

## 次のタスク

[Kafka プロデューサーステップと ProducerV2 API を使用したメッセージの公開](https://servicenow-prod.fluidtopics.net/RE4B67ciMC_MnKJuIPpwSg "ServiceNow から Hermes にデータをプッシュし、Kafka コンシューマーが読み取れるようにします。")に展開したカスタムモバイルアプリケーションを使用して管理しました。