JSON形式のデータをOSSからMaxComputeに移行する

このトピックでは、DataWorks Data Integrationを使用してJSON形式のデータをObject Storage Service (OSS) からMaxComputeに移行し、GET_JSON_OBJECT関数を使用してJSONオブジェクトを抽出する方法について説明します。

前提条件

MaxComputeおよびDataWorksサービスが有効化されます。詳細については、「MaxComputeとDataWorksの有効化」をご参照ください。
MaxComputeデータソースが追加されました。詳細については、「MaxComputeデータソースの追加」をご参照ください。
ワークフローは、DataWorksコンソールのワークスペースに作成されます。この例では、基本モードのワークスペースにワークフローが作成されます。詳細については、「ワークフローの作成」をご参照ください。

JSON形式のデータを含むTXTファイルがOSSバケットにアップロードされます。この例では、中国 (上海) リージョンにあるOSSバケットが使用されています。 TXTファイルには、次のJSON形式のデータが含まれています。

{
    "store": {
        "book": [
             {
                "category": "reference",
                "author": "Nigel Rees",
                "title": "Sayings of the Century",
                "price": 8.95
             },
             {
                "category": "fiction",
                "author": "Evelyn Waugh",
                "title": "Sword of Honour",
                "price": 12.99
             },
             {
                 "category": "fiction",
                 "author": "J. R. R. Tolkien",
                 "title": "The Lord of the Rings",
                 "isbn": "0-395-19395-8",
                 "price": 22.99
             }
          ],
          "bicycle": {
              "color": "red",
              "price": 19.95
          }
    },
    "expensive": 10
}

OSSデータソースを追加します。詳細については、「OSSデータソースの追加」をご参照ください。
DataWorksにテーブルを作成し、OSSから移行するJSON形式のデータを保存します。
1. DataWorks コンソールにログインします。上部のナビゲーションバーで、目的のリージョンを選択します。左側のナビゲーションウィンドウで、[データ開発と操作とメンテナンス] > [データ開発] を選択します。表示されるページで、ドロップダウンリストから目的のワークスペースを選択し、[入力データ開発] をクリックします。
2. 左側のナビゲーションウィンドウで、[データソース] をクリックし、[データソース] ページで、[データソースの作成] をクリックしてMaxComputeプロジェクトを追加します。
3. [データソースの作成] ダイアログボックスでパラメーターを入力し、[データソースの追加とDataStudioへの関連付け] をクリックします。
4. DataStudioページで、ポインタをアイコンの上に移動し、[テーブルの作成] > [MaxCompute] > [テーブル] を選択します。
5. [テーブルの作成] ダイアログボックスで、[パス] と [名前] パラメーターを設定し、[作成] をクリックします。
  説明
  複数のMaxComputeデータソースがDataStudioに関連付けられている場合は、[テーブルの作成] ダイアログボックスでMaxComputeデータソースを選択する必要があります。
6. テーブル編集ページで、DDLステートメント.をクリックします。
7. [DDL] ダイアログボックスで、次のテーブル作成ステートメントを入力し、[テーブルスキーマの生成] をクリックします。
```
CREATE TABLE mqdata (mq_data string);
```
8. [確認] メッセージの [OK] をクリックします。
9. テーブルスキーマが生成されたら、[全般] セクションで [表示名] パラメーターを設定し、[開発環境にコミット] および [本番環境にコミット] をクリックします。
  説明
  基本モードでワークスペースを使用する場合は、[本番環境にコミット] をクリックするだけです。

バッチ同期タスクを作成します。

データ分析ページに移動します。指定したワークフローを右クリックし、[新規] > [データ統合] > [オフライン同期] を選択します。
でノードを作成するダイアログボックスは、ノード名をクリックし、送信.をクリックします。
上部のナビゲーションバーで、アイコンが表示されます。
スクリプトモードで、アイコンが表示されます。
でテンプレートのインポートダイアログボックスソースタイプ,データソース,ターゲットタイプとデータソースをクリックし、確認.をクリックします。

JSONコードを変更し、アイコンが表示されます。

サンプルコード：

{
    "type": "job",
    "steps": [
        {
            "stepType": "oss",
            "parameter": {
                "fieldDelimiterOrigin": "^",
                "nullFormat": "",
                "compress": "",
                "datasource": "OSS_userlog",
                "column": [
                    {
                        "name": 0,
                        "type": "string",
                        "index": 0
                    }
                ],
                "skipHeader": "false",
                "encoding": "UTF-8",
                "fieldDelimiter": "^",
                "fileFormat": "binary",
                "object": [
                    "applog.txt"
                ]
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "odps",
            "parameter": {
                "partition": "",
                "isCompress": false,
                "truncate": true,
                "datasource": "odps_source",// The name of the MaxCompute data source.
                "column": [
                    "mqdata"
                ],
                "emptyAsNull": false,
                "table": "mqdata"
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "version": "2.0",
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    },
    "setting": {
        "errorLimit": {
            "record": ""
        },
        "speed": {
            "concurrent": 2,
            "throttle": false
        }
    }
}

結果を確認する

ODPS SQLノードを作成します。

ワークフローを右クリックし、新しい > MaxCompute > ODPS SQL.を選択します。
で関数を作成するダイアログボックスは、関数名をクリックします。送信.をクリックします。

ODPS SQLノードの [設定] タブで、次のステートメントを入力します。

-- Query data in the mqdata table. 
SELECT * from mqdata;
-- Obtain the value of the expensive field. 
SELECT GET_JSON_OBJECT(mqdata.MQdata,'$.expensive') FROM mqdata;

アイコンをクリックしてコードを実行します。
操作ログで結果を表示します。