データの処理 - - Alibaba Cloud ドキュメントセンター

テーブルの作成

[データ分析] セクションで作成されたワークフローをクリックして [MaxCompute] を右クリックし、[作成] > [テーブル] を選択します。
表示される [テーブルを作成] ダイアログボックスでテーブル名を入力し、 [コミット] をクリックします。
ODS レイヤーに、 ods_log_info_d、DW レイヤーにdw_user_info_all_d、RPT レイヤーに rpt_user_info_d という名前でテーブルを作成します。

DDL 文を実行してテーブルを作成します。

ods_log_info_d テーブルを作成します。

ods_log_info_d テーブルの [編集] タブで [DDL 文] をクリックし、[DDL 文] ダイアログボックスにテーブルを作成する文を入力します。

-- ODS レイヤーでテーブルを作成します
CREATE TABLE IF NOT EXISTS ods_log_info_d (
  ip STRING COMMENT リクエストを送信するクライアントの 'IP アドレス',
  uid STRING COMMENT 'クライアントユーザーの ID',
  time STRING COMMENT 'ユーザーが Web ページにアクセスした時刻 (yyyymmddhh:mm:ss 形式)',
  status STRING COMMENT サーバーから返されたステータスコード',
  bytes STRING COMMENT 'クライアントに返されたバイト数',
  region STRING COMMENT 'IP アドレスに基づいて取得されたユーザーのリージョン',
  method STRING COMMENT 'HTTP リクエストのタイプ',
  url STRING COMMENT 'ユーザーがアクセスした Web ページの URL',
  protocol STRING COMMENT 'HTTP のバージョン番号',
  referer STRING COMMENT 'リクエストされたリソースにリンクした Web ページの URL',
  device STRING COMMENT '端末のタイプ',
  identity STRING COMMENT 'アクセスタイプ (crawler、feed、user、または unknown)'
)
PARTITIONED BY (
  dt STRING
);

dw_user_info_all_d テーブルを作成します。

dw_user_info_all_d テーブルの [編集] タブで [DDL 文] をクリックし、[DDL 文] ダイアログボックスにテーブルを作成する文を入力します。

-- DW レイヤーでテーブルを作成します。
CREATE TABLE IF NOT EXISTS dw_user_info_all_d (
  uid STRING COMMENT 'クライアントユーザーの ID',
  gender STRING COMMENT 'ユーザーの性別',
  age_range STRING COMMENT 'ユーザーの年齢層',
  zodiac STRING COMMENT 'ユーザーの星座',
  region STRING COMMENT 'IP アドレスに基づいて取得されたユーザーのリージョン',
  device STRING COMMENT '端末のタイプ',
  identity STRING COMMENT 'アクセスタイプ (crawler、feed、user、または unknown)'
  method STRING COMMENT 'HTTP リクエストのタイプ',
  url STRING COMMENT 'ユーザーがアクセスした Web ページの URL',
  referer STRING COMMENT 'リクエストされたリソースにリンクした Web ページの URL',
  time STRING COMMENT 'ユーザーが Web ページにアクセスした時刻 (yyyymmddhh:mm:ss 形式)',
)
PARTITIONED BY (
  dt STRING
);

rpt_user_info_d テーブルを作成します。

rpt_user_info_d テーブルの [編集] タブで [DDL 文] をクリックし、[DDL 文] ダイアログボックスにテーブルを作成する文を入力します。

-- RPT レイヤーでテーブルを作成します。
CREATE TABLE IF NOT EXISTS rpt_user_info_d (
  uid STRING COMMENT 'クライアントユーザーの ID',
  region STRING COMMENT 'IP アドレスに基づいて取得されたユーザーのリージョン',
  device STRING COMMENT '端末のタイプ',
  pv BIGINT COMMENT 'ユーザーが Web ページを表示した回数',
  gender STRING COMMENT 'ユーザーの性別',
  age_range STRING COMMENT 'ユーザーの年齢層',
  zodiac STRING COMMENT 'ユーザーの星座',
)
PARTITIONED BY (
  dt STRING
);

[テーブルスキーマの作成] をクリックします。表示される [確認]ダイアログボックスで [OK] をクリックします。
[テーブルの編集] タブの [一般] セクションで、テーブルの表示名を入力します。
[開発環境にコミット] および [本番環境にコミット] を順にクリックします。

ワークフローの設計

ワークフローのノード間の依存関係を設定する方法の詳細については、「データの収集」をご参照ください。

DataStudio コンソールで、左側のナビゲーションペインに作成されたワークフローをダブルクリックします。 [ワークフローの編集] タブで、左側に表示される [ODPS SQL] を選択して右側の [編集] セクションにドラッグします。 [ノードの作成] ダイアログボックスで [ノード名] を指定し、[コミット] をクリックします。

合計で 3 つの ODPS SQL ノードを作成し、それぞれods_log_info_d、dw_user_info_all_d、およびrpt_user_info_dという名前とします。次に、ノード間の依存関係を設定します。

UDF の作成

リソースを作成します。
1. ip2region.jarファイルをダウンロードし、ローカルのディレクトリに保存します。
2. [データ分析] セクションで、作成したワークフローをクリックし、[MaxCompute] を右クリックして [作成] > [リソース] > [JAR] を選択します。
3. [リソースの作成] ダイアログボックスで [アップロード] をクリックし、ダウンロードした ip2region.jar ファイルを選択します。
4. [OK] をクリックします。

UDF (ユーザー定義関数) を登録します。

[データ分析] セクションで、作成したワークフローをクリックし、[MaxCompute] を右クリックして [作成] > [関数] を選択します。
[関数の作成]ダイアログボックスで、UDF の名前 (getregionなど) を入力し、 [コミット] をクリックします。

[設定] タブの [関数の登録]セクションでパラメーターを設定し、[保存] および [送信] アイコンを順番にクリックします。


パラメーター	説明
Function Type	UDF の型を設定します。
Engine Instance MaxCompute	現在のワークスペースにバインドされた MaxCompute のエンジンインスタンスを指定します。エンジンインスタンスは変更できません。
Function Name	UDF の作成時に入力した、UDFの名前を設定します。
Owner	ノードの所有者を設定します。
Class Name	UDF を実装するためのクラス名を設定します。 `org.alidata.odps.udf.Ip2Region` に設定します。
Resources	作成したリソースを設定します。値を `ip2region.jar` に設定します。
Description	DAG の説明を記述します。 "IP アドレスに基づくリージョンの変換" に設定します。
Expression Syntax	UDF の式の構文を設定します。 `getregion('ip')` に設定します。
Parameter Description	式のパラメーターの説明を記述します。 "IP アドレス" に設定します。

ODPS SQL ノードの設定

ods_log_info_dノードを設定します。

ods_log_info_d ノードをダブルクリックします。

[ノードの編集] タブで、以下のSQL 文を入力します。

INSERT OVERWRITE TABLE ods_log_info_d PARTITION (dt=${bdp.system.bizdate})
SELECT ip
  , uid
  , time
  , status
  , bytes 
  , getregion(ip) AS region -- UDF を使用して、IP アドレスに基づくリージョンを取得。
  , regexp_substr(request, '(^[^ ]+ )') AS method -- 正規表現を使用して、リクエストから 3 つのフィールドを抽出。
  , regexp_extract(request, '^[^ ]+ (. *) [^ ]+$') AS url
  , regexp_substr(request, '([^ ]+$)') AS protocol 
  , regexp_extract(referer, '^[^/]+://([^/]+){1}') AS referer -- 正規表現を使用して、HTTP リファラを消去し、正確な URL を取得。
  , CASE
    WHEN TOLOWER(agent) RLIKE 'android' THEN 'android' -- エージェントのパラメータ値から、端末およびアクセスタイプを取得。
    WHEN TOLOWER(agent) RLIKE 'iphone' THEN 'iphone'
    WHEN TOLOWER(agent) RLIKE 'ipad' THEN 'ipad'
    WHEN TOLOWER(agent) RLIKE 'macintosh' THEN 'macintosh'
    WHEN TOLOWER(agent) RLIKE 'windows phone' THEN 'windows_phone'
    WHEN TOLOWER(agent) RLIKE 'windows' THEN 'windows_pc'
    ELSE 'unknown'
  END AS device
  , CASE
    WHEN TOLOWER(agent) RLIKE '(bot|spider|crawler|slurp)' THEN 'crawler'
    WHEN TOLOWER(agent) RLIKE 'feed'
    OR regexp_extract(request, '^[^ ]+ (. *) [^ ]+$') RLIKE 'feed' THEN 'feed'
    WHEN TOLOWER(agent) NOT RLIKE '(bot|spider|crawler|feed|slurp)'
    AND agent RLIKE '^[Mozilla|Opera]'
    AND regexp_extract(request, '^[^ ]+ (. *) [^ ]+$') NOT RLIKE 'feed' THEN 'user'
    ELSE 'unknown'
  END AS identity
  FROM (
    SELECT SPLIT(col, '##@@')[0] AS ip
    , SPLIT(col, '##@@')[1] AS uid
    , SPLIT(col, '##@@')[2] AS time
    , SPLIT(col, '##@@')[3] AS request
    , SPLIT(col, '##@@')[4] AS status
    , SPLIT(col, '##@@')[5] AS bytes
    , SPLIT(col, '##@@')[6] AS referer
    , SPLIT(col, '##@@')[7] AS agent
  FROM ods_raw_log_d
  WHERE dt = ${bdp.system.bizdate}
) a;

左上に表示される [保存] アイコンをクリックします。

dw_user_info_all_dノードを設定します。

dw_user_info_all_dノードをダブルクリックします。

[ノードの編集] タブで、以下のSQL 文を入力します。

INSERT OVERWRITE TABLE dw_user_info_all_d PARTITION (dt='${bdp.system.bizdate}')
SELECT COALESCE(a.uid, b.uid) AS uid
  , b.gender
  , b.age_range
  , b.zodiac
  , a.region
  , a.device
  , a.identity
  , a.method
  , a.url
  , a.referer
  , a.time
FROM (
  SELECT *
  FROM ods_log_info_d
  WHERE dt = ${bdp.system.bizdate}
) a
LEFT OUTER JOIN (
  SELECT *
  FROM ods_user_info_d
  WHERE dt = ${bdp.system.bizdate}
) b
ON a.uid = b.uid;

左上に表示される [保存] アイコンをクリックします。

rpt_user_info_dノードを設定します。
1. rpt_user_info_d ノードをダブルクリックします。
2. 表示される [ノードの編集] タブで、以下の SQL 文を入力します。
```
INSERT OVERWRITE TABLE rpt_user_info_d PARTITION (dt='${bdp.system.bizdate}')
SELECT uid
  , MAX(region)
  , MAX(device)
  , COUNT(0) AS pv
  , MAX(gender)
  , MAX(age_range)
  , MAX(zodiac)
FROM dw_user_info_all_d
WHERE dt = ${bdp.system.bizdate}
GROUP BY uid;
```
3. 左上に表示される [保存] アイコンをクリックします。

ワークフローのコミット

[ワークフローの編集] タブで [送信] アイコンをクリックしてワークフローに設定したノードをコミットします。
表示される [コミット] ダイアログボックスでコミットするノードを選択して説明を入力し、[I/O 不整合アラートを無視]をクリックします。
[コミット] をクリックします。

ワークフローの実行

[ワークフローの編集] タブで [実行] アイコンをクリックし、ノードコードのロジックを確認します。
すべてのノードが実行され、各ノードの横に緑色のチェックマーク (✓) が表示されたら、ナビゲーションサブメニューの [アドホッククエリ] をクリックします。 [アドホッククエリ] セクションが表示されます。
[アドホッククエリ] を右クリックして、[ノードの作成] > [ODPS SQL] をクリックします。
[ノードの作成] ダイアログボックスでノード名を入力し、[コミット] をクリックします。表示されるタブで、SQL 文を実行してノードの実行結果を照会し、必要なデータが生成されているかどうかを確認します。
以下のSQL文を使用します。デフォルトでは、ノードのデータタイムスタンプはノード実行の 1 日前です。
```
--- View data in the rpt_user_info_d table.
select * from rpt_user_info_d where dt=Data timestamp limit 10;
```

ワークフローのデプロイ

ワークフローをコミットすると、ワークフローのノードは開発環境にデプロイされます。開発環境のノードは自動的にスケジュールできないため、設定済みのノードを本番環境にデプロイする必要があります。

注

ノードコードをテストし、コードが正しいことを確認してからノードを運用環境にデプロイします。
基本モードのワークスペースでは、ワークフローの [デプロイ] アイコンが使用できません。 [オペレーションセンターに移動] をクリックして、オペレーションセンターでノードをスケジュールします。

[ワークフローの編集] タブで、 [デプロイ] アイコンをクリックします。
表示される[デプロイ] ページでデプロイするノードを選択し、[リストに追加] をクリックします。
右上に表示される [デプロイ予定ノードのリスト] をクリックします。 [デプロイするノード] ダイアログボックスで、[すべてデプロイ] をクリックします。
表示される [デプロイタスクの作成] ダイアログボックスで、 [デプロイ] をクリックします。
左側のナビゲーションペインで [パッケージのリリース] をクリックし、デプロイのステータスを表示します。

本番環境でのノード実行

ワークフローのデプロイ後、右上に表示される [オペレーションセンター] をクリックします。
[ワークフローの編集] タブのツールバーで、 [オペレーションセンターに移動] をクリックして[オペレーションセンター] ページに移動することもできます。
[オペレーションセンター] ページの左側に表示されるナビゲーションペインで、[定期タスクのメンテナンス] > [定期タスク] in the left-side navigation pane. [定期タスク] ページで、ワークショップのワークフローをクリックします。
右側に表示される DAG (有向非循環グラフ) で workshop_start ノードを右クリックし、 [実行] > [現在および子孫ノードを遡及的に] を選択します。
表示される[補完データ] ダイアログボックスで、遡及データを生成するノードを選択し、データのタイムスタンプを指定して [OK] をクリックします。 [補完データ] ページが表示されます。
すべての遡及データ生成インスタンスのステータスが [成功] になるまで [更新] をクリックします。

次のステップ

このトピックでは、SQLノードを作成して生ログを処理する方法を学びました。次のチュートリアルに進み、開発したノードによって生成されたデータの品質をモニタリングおよび保証する方法を学ぶことができます。詳細については、「データ品質モニタリング」をご参照ください。

:データの処理

始める前に