DataWorks:RestAPI Reader のベストプラクティス

実践 1：期間内のデータを照会する API からのデータ読み取り

シナリオ例：API 定義

事前準備：MaxCompute パーティションテーブルの作成

CREATE TABLE IF NOT EXISTS ods_xiaobo_rest2
(
  `axis`  STRING
  ,`value` BIGINT
  ,`createTime` STRING
)
PARTITIONED BY
(
  ds  STRING
)
LIFECYCLE 3650;

同期タスクの設定

1. RestAPI データソースの追加

   DataWorks ワークスペースに RestAPI データソースを追加します。 詳細については、「RestAPI データソースを追加する」をご参照ください。 [RestAPI データソースの追加] ダイアログで、[データソース名] と [データソースの説明] を指定し、[適用環境] (開発または本番) を選択し、[デフォルトリクエストヘッダー] (デフォルト値: {}) を設定します。 主な設定は以下のとおりです。

   • url：これを RESTful API アドレスに設定します。

   • 認証方式：データソース API でサポートされている認証方式を選択し、必要な認証パラメーターを設定します。

   • Resource group connectivity：RestAPI データソースは、Data Integration 専用リソースグループのみをサポートしています。Data Integration 専用リソースグループを選択し、データソースとリソースグループ間の接続性をテストします。

2. バッチ同期ノードの作成と同期タスクの設定

   DataWorks の DataStudio でバッチ同期ノードを作成します。詳細については、「バッチ同期ノードの作成」をご参照ください。主要な構成は次のとおりです。

   • Data Source の設定：

     • Data Source：前のステップで作成した RestAPI データソースを選択します。

     • Request Method：サンプルの API は GET API です。[GET] を選択します。

     • Return Data Structure: サンプル API の応答は JSON 配列です。Array Data を選択します。

     • The JSON path for data storage：サンプル API によって返されるデータは data フィールドに格納されます。これを data に設定します。

     • Request Param：リクエストパラメーターとスケジューリングパラメーターを組み合わせて使用し、毎日当日のデータを同期します。

       • リクエストパラメーターを startTime=${extract_day} ${start_time}&endTime=${extract_day} ${end_time} に設定します。

       • スケジュール設定で、extract_day=${yyyy-mm-dd}、start_time=00:00:00、end_time=23:59:59 の 3 つのスケジューリングパラメーターを追加します。

       実行日が 2023-01-05 の場合、extract_day の値は 2023-01-04 となり、リクエストパラメーターは startTime=2023-01-04 00:00:00&endTime=2023-01-04 23:59:59 のように連結されます。

   • Data Destination の設定：

     • Data Source と Table：前のステップで作成した MaxCompute パーティションテーブルを選択します。

     • Partition Information：パーティション情報とスケジューリングパラメーターを組み合わせて使用します。

       • パーティション情報を ${bizdate} に設定します。

       • スケジュール設定で、bizdate=$bizdate のスケジューリングパラメーターを 1 つ追加します。

       実行日が 2023-01-05 の場合、パーティション情報の値は 20230104 になります。

   • Field Mapping構成では、API のデータ定義に基づいて、RestAPI 応答から列名を入力します。列名では大文字と小文字が区別されます。列を追加した後、The same name mapping を使用するか、手動で線を引いて列マッピングを作成します。

テスト実行

データ検証

select * from ods_xiaobo_rest2 where ds='20230104' order by createtime;

送信、デプロイ、およびデータのバックフィル

実践 2：ページ分割された RestAPI からのデータ読み取り

シナリオ例：API 定義

事前準備：MaxCompute パーティションテーブルの作成

CREATE TABLE IF NOT EXISTS ods_xiaobo_rest1
(
  `id` BIGINT
  ,`name` STRING
)
PARTITIONED BY
(
  ds  STRING
)
LIFECYCLE 3650;

同期タスクの設定

1. RestAPI データソースの追加

   DataWorks ワークスペースに RestAPI データソースを追加します。 詳細については、「RestAPI データソースの追加」をご参照ください。 [Add RestAPI Data Source] ダイアログで、[Data Source Name] と [Data Source Description] を指定し、[Applicable Environment] (開発または本番) を選択し、[Default Request Header] (デフォルト値: {}) を設定します。 主な設定は次のとおりです。

   • url：これを RESTful API アドレスに設定します。

   • 認証方式：データソース API でサポートされている認証方式を選択し、必要な認証パラメーターを設定します。

   • Resource group connectivity：RestAPI データソースは、Data Integration 専用リソースグループのみをサポートしています。Data Integration 専用リソースグループを選択し、データソースとリソースグループ間の接続性をテストします。

2. バッチ同期ノードの作成と同期タスクの設定

   DataWorks の DataStudio でバッチ同期ノードを作成します。詳細については、「バッチ同期ノードの作成」をご参照ください。主要な構成は次のとおりです。

   • Data Source の設定：

     • Data Source：前のステップで作成した RestAPI データソースを選択します。

     • Request Method：サンプルの API は GET API です。[GET] を選択します。

     • Return Data Structure: サンプル API の応答は JSON 配列です。Array Data を選択します。

     • The JSON path for data storage：サンプル API によって返されるデータは data フィールドに格納されます。これを data に設定します。

     • Request Param：ページサイズは固定です。これを pageSize=50 に設定します。ページサイズを大きく設定しすぎると、RestAPI サーバーと同期タスクに過度の負荷がかかる可能性があるため、推奨されません。

     • The number of requests：この例では Multiple Requests を選択します。

       この API のページネーションパラメーターは pageNum です。Multiple Requests を選択した後、関連パラメーターを次のように設定します。

       • Parameter used for multiple requests：これを pageNum に設定します。

       • 開始インデックス：これを 1 に設定します。

       • ステップ：これを 1 に設定します。

       • 終了インデックス：これを 100 に設定します。

   • Data Destination の設定：

     • Data Source と Table：前のステップで作成した MaxCompute パーティションテーブルを選択します。

     • Partition Information：パーティション情報とスケジューリングパラメーターを組み合わせて使用します。

       • パーティション情報を ${bizdate} に設定します。

       • スケジュール設定で、bizdate=$bizdate のスケジューリングパラメーターを 1 つ追加します。

       実行日が 2023-01-05 の場合、パーティション情報の値は 20230104 になります。

   • Field Mapping 構成では、API のデータ定義に基づいて、RestAPI 応答から列名を入力します。列名では大文字と小文字が区別されます。列を追加した後、The same name mapping を使用するか、手動で線を引いて列マッピングを作成します。

テスト実行

データ検証

select * from ods_xiaobo_rest1 where ds='20230104' order by id;

実践 3：POST タイプの RestAPI からのデータ読み取り

シナリオ例：API 定義

事前準備：MaxCompute パーティションテーブルの作成

CREATE TABLE IF NOT EXISTS ods_xiaobo_rest3
(
  `user_id` BIGINT
  ,`name` STRING
  ,`axis`  STRING
  ,`value` BIGINT
  ,`create_time` STRING
)
PARTITIONED BY
(
  ds  STRING
)
LIFECYCLE 3650;

同期タスクの設定

1. RestAPI データソースの追加

   DataWorks ワークスペースに RestAPI データソースを追加します。 詳細については、「RestAPI データソースを追加する」をご参照ください。 [RestAPI データソースの追加] ダイアログで、[データソース名] と [データソースの説明] を指定し、[適用環境] (開発または本番) を選択して、[デフォルトリクエストヘッダー] (デフォルト値: {}) を設定します。 主な設定は次のとおりです。

   • url：これを RESTful API アドレスに設定します。

   • 認証方式：データソース API でサポートされている認証方式を選択し、必要な認証パラメーターを設定します。

   • Resource group connectivity：RestAPI データソースは、Data Integration 専用リソースグループのみをサポートしています。Data Integration 専用リソースグループを選択し、データソースとリソースグループ間の接続性をテストします。

2. バッチ同期ノードの作成と同期タスクの設定

   DataWorks の DataStudio でバッチ同期ノードを作成します。詳細については、「バッチ同期ノードの作成」をご参照ください。主要な構成は次のとおりです。

   • Data Source の設定：

     • Data Source：前のステップで作成した RestAPI データソースを選択します。

     • Request Method：サンプルの API は POST API です。[POST] を選択します。

     • Return Data Structure: サンプル API の応答は JSON 配列です。Array Data を選択します。

     • The JSON path for data storage：サンプル API によって返されるデータは data フィールドに格納されます。これを data に設定します。

     • ヘッダー：この例の POST API は JSON リクエストボディを受け入れます。これを {"Content-Type":"application/json"} に設定します。

     • Request Param：リクエストパラメーターとスケジューリングパラメーターを組み合わせて使用し、毎日当日のデータを同期します。

       • リクエストパラメーターを次のように設定します：

         {
         
             "userId":16,
         
             "startTime":"${extract_day} 00:00:00",
         
             "endTime":"${extract_day} 23:59:59"
         
         }

       • スケジュール設定で、extract_day=${yyyy-mm-dd} のスケジューリングパラメーターを 1 つ追加します。

   • Data Destination の設定：

     • Data Source と Table：前のステップで作成した MaxCompute パーティションテーブルを選択します。

     • Partition Information：パーティション情報とスケジューリングパラメーターを組み合わせて使用します。

       • パーティション情報を ${bizdate} に設定します。

       • スケジュール設定で、bizdate=$bizdate のスケジューリングパラメーターを 1 つ追加します。

       実行日が 2023-01-05 の場合、パーティション情報の値は 20230104 になります。

   • Field Mapping構成: API のデータ定義に基づき、RestAPI 応答の列名を入力します。ドット (.) を使用して、ネストされた列を区切ることができます。列名では大文字と小文字が区別されます。列を追加した後、The same name mappingを使用するか、手動で線を描画して列マッピングを作成します。

テスト実行

データ検証

select * from ods_xiaobo_rest3 where ds='20230105' order by create_time;

実践 4：リクエストパラメーターをループ処理して RestAPI から読み取る

シナリオ例：API 定義

この実践では、RESTful API からデータを反復的に読み取り、MaxCompute のパーティションテーブルに書き込みます。サンプルの RESTful API は、自己構築したテスト用の GET API で、日付、省、市の入力パラメーターに基づいて温度データを返します。

実際の操作を行う際は、使用する API に基づいて構成を調整してください。この例の API は、ワークフローを理解していただくためのデモとしてのみ使用されます。

• リクエストの例：

  http://TestAPIAddress:Port/rest/test5?date=2023-01-04&province=zhejiang&city=hangzhou

• 応答の例：

  {
    "province": "P1",
    "city": "hz",
    "date": "2023-01-04",
    "minTemperature": "-14",
    "maxTemperature": "-7",
    "unit": "℃",
    "weather": "cool"
  }

• テストツールでの API 呼び出しの例：Postman などの API テストツールで GET リクエストを送信します。date=2023-01-04、province=p1、city=hz のパラメーターを渡します。API は、対応する minTemperature (-14)、maxTemperature (-7)、unit (℃)、weather (COOL) の情報を含む JSON 応答を返します。

事前準備：パラメーターテーブルと MaxCompute パーティションテーブルの作成

この実践では、API から読み取ったデータが MaxCompute のパーティションテーブルに同期されます。まず、反復処理のための省と市の値を格納するパラメーターテーブルを作成し、次に同期されたデータを格納するためのパーティションテーブルを作成する必要があります。

パーティションテーブルを上書きコマンドと共に使用すると、パーティションの上書き効果が得られます。これにより、同期タスクはデータの重複なしに再実行可能になり、パーティションテーブルはデータ分析にも使いやすくなります。

CREATE TABLE 文は次のとおりです：

CREATE TABLE IF NOT EXISTS `citys`
(
  `province` STRING ,
  `city` STRING
);

insert into citys
select 'shanghai','shanghai'
union all select 'zhejiang','hangzhou'
union all select 'sichuan','chengdu';

CREATE TABLE IF NOT EXISTS ods_xiaobo_rest5
(
    `minTemperature` STRING ,
    `maxTemperature` STRING ,
    `unit` STRING ,
    `weather` STRING 
)
PARTITIONED BY 
(
    `province` STRING ,
    `city` STRING ,
    `ds`  STRING
)
LIFECYCLE 3650;

DataWorks Standard Edition を使用し、作成したテーブルを本番環境に送信すると、データマップでテーブルを表示できます。

同期タスクの設定

1. RestAPI データソースの追加

   DataWorks ワークスペースに RestAPI データソースを追加します。 詳細については、「RestAPI データソースを追加する」をご参照ください。 [RestAPI データソースの追加] ダイアログで、[データソース名]と[データソースの説明]を指定し、[適用環境] (開発または本番) を選択して、[デフォルトリクエストヘッダー] (デフォルト値: {}) を構成します。 主な設定は次のとおりです：

   • url：これを RESTful API アドレスに設定します。

   • 認証方式：データソース API でサポートされている認証方式を選択し、必要な認証パラメーターを設定します。

   • Resource group connectivity：指定されたリソースグループを選択し、接続性をテストします。

2. DataStudio で setval_citys という名前の代入ノードを作成します。 詳細については、「代入ノードを作成する」をご参照ください。

   主要な構成は次のとおりです：

   番号	説明
   ①	代入言語：ODPS SQL 代入コード： SELECT province ,city FROM citys;
   ②	再実行プロパティ: You can run again after successful or failed operation. に設定します。

   設定が完了したら、代入ノードを送信してデプロイします。

3. DataStudio で for-each ノードを作成します。詳細については、「for-each ノードを作成する」をご参照ください。キー構成は次のとおりです:

   番号	説明
   ①	再実行プロパティ: You can run again after successful or failed operation. に設定します。
   ②	上流ノードの依存関係：前のステップのノード、つまり setval_citys ノードを選択します。
   ③	ノードコンテキストパラメーター：入力パラメーターのソースを選択します。
   ④	バッチ同期ノード：for-each ノード内でバッチ同期ノードを設定します。詳細については、次のステップをご参照ください。

4. バッチ同期ノードを作成し、同期タスクを設定します。詳細については、「バッチ同期ノードの作成」をご参照ください。

   主要な構成は次のとおりです：

   番号	説明
   ①	スケジューリングパラメーターを次のように設定します： bizdate=$[yyyymmdd-1] bizdate_year=$[yyyy-1] bizdate_month=$[mm-1] bizdate_day=$[dd-1]
   ②	RestAPI リクエストパラメーターを設定します。province と city パラメーターは for-each ノードから取得されます。 date=${bizdate_year}-${bizdate_month}-${bizdate_day}&province=${dag.foreach.current[0]}&city=${dag.foreach.current[1]}
   ③	MaxCompute パーティションパラメーターを設定します。province パラメーターは for-each ノードから取得されます。 province=${dag.foreach.current[0]}
   ④	MaxCompute パーティションパラメーターを設定します。city パラメーターは for-each ノードから取得されます。 city=${dag.foreach.current[1]}
   ⑤	MaxCompute パーティションパラメーターを設定します。ds パラメーターはスケジューリングパラメーターから取得されます。 ds=${bizdate}
   ⑥	API のデータ定義に基づいて、RestAPI 応答から列名を入力します。列名では大文字と小文字が区別されます。列を追加した後、The same name mapping を使用するか、手動で線を引いて列マッピングを作成します。

   設定が完了したら、for-each ノードを送信してデプロイします。

テスト実行

1. 代入ノードと for-each ノードを正常に送信して デプロイ した後、オペレーションセンターの定期タスクに移動し、代入ノードでデータのバックフィル操作を実行します。詳細については、「データのバックフィル」をご参照ください。

2. 実際のシナリオに基づいて、データのバックフィルタスクのビジネス日付と下流ノードを選択します。

3. タスクが実行された後、データのバックフィルインスタンスの詳細で実行パラメーターを確認し、期待どおりであるかを確認し、ランタイムログの正当性を確認します。

   この例では、データは MaxCompute テーブルの province=shanghai,city=shanghai,ds=20231215 パーティションに書き込まれます。

データ検証

DataStudio でアドホッククエリを実行して、データが MaxCompute に正しく同期されているかを確認できます。以下はクエリの例です：

この例では、ods_xiaobo_rest5 は 事前準備 ステップで作成した MaxCompute パーティションテーブルです。

SELECT  weather
        ,mintemperature
        ,maxtemperature
        ,unit
        ,province
        ,city
        ,ds
FROM    ods_xiaobo_rest5
WHERE   ds != 1
ORDER BY ds,province,city;

クエリが完了したら、データが MaxCompute に正しく同期されているかを確認します。