:DataWorks OpenLake ワンストップインテリジェントデータレイクハウス開発

DataWorks の概要

DataWorks は、データレイクハウスの開発とガバナンスのためのインテリジェントなプラットフォームです。Alibaba の 15 年にわたるビッグデータ構築手法に基づいて構築されています。MaxCompute、E-MapReduce、Hologres、Flink、PAI など、数十の Alibaba Cloud ビッグデータおよび AI コンピューティングサービスと深く統合されています。DataWorks は、データウェアハウス、データレイク、および OpenLake データレイクハウスアーキテクチャ向けに、インテリジェントな抽出・変換・書き出し (ETL) 開発、データ分析、およびプロアクティブなデータ資産ガバナンスサービスを提供します。これにより、「Data+AI」ライフサイクル全体を管理できます。2009 年以来、DataWorks は Alibaba のデータアーキテクチャを継続的に製品化してきました。政府、金融、小売、インターネット、自動車、製造など、さまざまな業界にサービスを提供しています。何万もの顧客が DataWorks を使用してデジタルトランスフォーメーションを推進し、価値を創造しています。

DataWorks Copilot の概要

DataWorks Copilot は DataWorks のインテリジェントアシスタントです。DataWorks では、DataWorks のデフォルトモデルである Qwen3-235B-A22B、DeepSeek-R2-0528、または Qwen3-Coder 大規模モデルを選択して Copilot 操作を実行できます。DeepSeek-R2 の深い推論機能により、DataWorks Copilot は自然言語対話を通じて SQL コードの生成、最適化、テストなどの複雑な操作を実行するのに役立ちます。これにより、ETL (抽出·変換·書き出し) 開発とデータ分析の効率が大幅に向上します。

DataWorks Notebook の概要

DataWorks Notebook は、データ開発と分析のためのインテリジェントでインタラクティブなツールです。複数のデータエンジンにまたがる SQL または Python 分析をサポートしています。コードを即座に実行またはデバッグし、視覚化されたデータ結果を表示できます。DataWorks Notebook は、他のタスクノードとオーケストレーションしてワークフローに組み込み、スケジューling システムに送信して実行することもできます。これにより、複雑なビジネスシナリオを柔軟に実装できます。

ステップ 1: データカタログを管理する

データレイクハウスの データカタログ 管理機能を使用すると、DLF、MaxCompute、Hologres などのサービスのデータカタログを管理および作成できます。

1. DataStudio で、左側のメニューの アイコンをクリックして [データカタログ] を開きます。ナビゲーションウィンドウで、管理するメタデータタイプを見つけ、[プロジェクトを追加] をクリックします。このボタンの名前はメタデータタイプによって異なる場合があります。このトピックでは MaxCompute を例として使用します。

   DataWorks ワークスペースからデータソースを追加できます。[MaxCompute-Project] タブで権限を持つ MaxCompute プロジェクトを選択することもできます。

   

2. プロジェクトを追加すると、対応するメタデータタイプの下に表示されます。プロジェクト名をクリックして、データカタログの詳細ページに移動します。

3. データカタログの詳細ページで、スキーマを選択し、任意のテーブル名をクリックしてテーブルの詳細を表示します。

4. データカタログでは、テーブルを視覚的に作成できます。

   データカタログを指定したスキーマの [テーブル] レベルまで展開します。右側の アイコンをクリックして [テーブルを作成] ページを開きます。

   

5. [テーブルを作成] ページでは、次の方法でテーブルを作成できます。

   • エリア ① で、[テーブル名] と [フィールド情報] を入力します。

   • エリア ② では、[DDL] 文を直接入力してテーブルを作成できます。

   

6. ページ上部の [公開] ボタンをクリックしてテーブルを作成します。

ステップ 2：ワークフローを調整する

ワークフローを使用すると、ビジネスロジックに基づいてさまざまなタイプのデータ開発ノードをドラッグアンドドロップしてオーケストレーションできます。各ノードのスケジューリング時間などの共通パラメーターを設定する必要はありません。これにより、複雑なタスクプロジェクトを簡単に管理できます。

1. DataStudio で、左側のプライマリメニューの アイコンをクリックして [データ開発] を開きます。左側のナビゲーションウィンドウで [プロジェクトフォルダ] を見つけ、その横にある アイコンをクリックし、[新しいワークフロー] を選択します。

2. ワークフローの [名前] を入力し、[OK] をクリックしてワークフローエディターを開きます。

3. ワークフローエディターで、キャンバス上の [ノードをドラッグまたはクリックして追加] をクリックします。[ノードを追加] ダイアログボックスで、[ノードタイプ] を [ゼロロードノード] に設定し、カスタムの [ノード名] を入力してから [確認] をクリックします。

4. 左側のノードタイプのリストから、必要なノードタイプを見つけてキャンバスにドラッグします。[ノードを追加] ダイアログボックスで、[ノード名] を入力し、[確認] をクリックします。

   

5. キャンバス上で、依存関係を作成する 2 つのノードを見つけます。一方のノードの下端の中央にマウスポインターを合わせます。ポインターが + に変わったら、矢印をもう一方のノードにドラッグしてマウスを離します。図のように依存関係を設定したら、上部のツールバーで [保存] をクリックします。

   

6. [保存] をクリックした後、必要に応じてキャンバスのレイアウトを調整できます。

7. ワークフローキャンバスの右側で、[スケジューリング構成] をクリックします。[スケジューリング構成] パネルを使用して、ワークフローのスケジューリングパラメーターとノードの依存関係を構成します。[スケジューリングパラメーター] セクションで、[パラメーターを追加] をクリックします。パラメーター名フィールドに bizdate と入力します。パラメーター値のドロップダウンリストから、$[yyyymmdd-1] を選択します。

   

8. [ワークスペースルートノードを使用] をクリックして、このノードをワークフローの上流依存関係として設定します。

   

9. ワークフローキャンバスの上にある [公開] をクリックします。右下隅に [公開コンテンツ] パネルが表示されます。パネルで、[本番環境への公開を開始] をクリックし、プロンプトに従って確認します。

   

ステップ 3: マルチエンジン共同開発を使用する

DataStudio は、Data Integration、MaxCompute、Hologres、EMR、Flink、Python、Notebook、ADB など、さまざまなエンジンにまたがる数十種類のノードタイプのデータウェアハウス開発をサポートしています。複雑なスケジューリング依存関係をサポートし、開発環境を本番環境から分離する開発モデルを提供します。この実験では、Flink SQL Streaming ノードの作成を例として使用します。

1. DataStudio で、左側のナビゲーションウィンドウの アイコンをクリックして [データ開発] ページを開きます。ナビゲーションウィンドウで [プロジェクトフォルダ] を見つけ、その横にある アイコンをクリックします。カスケードメニューで [Flink SQL Streaming] をクリックしてノードエディターを開きます。エディターが開く前に、[ノード名] を入力して Enter キーを押します。

   プリセットノード名: ads_ec_page_visit_log。

   

2. ノードエディターで、プリセットの Flink SQL Stream コードをコードエディターに貼り付けます。

   

   プリセット Flink SQL Stream コード

   CREATE TEMPORARY VIEW log_ri_base
   AS 
   SELECT 
     visit_time
     ,substring(visit_time,1,8) as stat_date
     ,substring(visit_time,9,2) as stat_hour
     ,visitor_id
     ,item_id
     ,cate_id
     ,ext_pv
   FROM vvp_ec_ads.dws_log_all_itm_ipv_ri
   WHERE
     bc_type IN ('b', 'z')
     AND coalesce(cate_id, '') <> ''
     AND visitor_type = 'uid'
     and coalesce(item_id, '') <> ''
     AND substring(visit_time,1,8) >= '${bizdate}'
   ;
   
   
   CREATE TEMPORARY VIEW itm_log_day
   AS
   SELECT
     sum(ext_pv) as pv
     ,count(DISTINCT visitor_id) FILTER (WHERE ext_pv>0) as uv
     ,stat_date
     ,cate_id
     ,item_id
   FROM log_ri_base
   GROUP BY stat_date
     ,cate_id
     ,item_id
   ;
   
   
   CREATE TEMPORARY VIEW itm_log_hh_00
   AS
   SELECT
     sum(ext_pv) as pv
     ,count(DISTINCT visitor_id) FILTER (WHERE ext_pv>0) as uv
     ,stat_date
     ,stat_hour
     ,item_id
     ,cate_id
   FROM log_ri_base
   GROUP BY stat_date
     ,stat_hour
     ,cate_id
     ,item_id
   ;
   
   BEGIN STATEMENT SET;
   
   INSERT INTO vvp_ec_ads.ads_ec_log
   SELECT
     a.stat_date
     ,cast(a.item_id as varchar) as item_id
     ,a.cate_id
     ,b.cate_name
     ,cast(b.industry_id as varchar) as industry_id
     ,cast(b.xcat1_id as varchar) as xcat1_id
     ,cast(b.xcat2_id as varchar) as xcat2_id
     ,cast(b.xcat3_id as varchar) as xcat3_id
     ,cast(b.cate_level1_id as varchar) as cate_level1_id
     ,cast(b.cate_level2_id as varchar) as cate_level2_id
     ,cast(b.is_sw as varchar) as is_sw
     ,a.pv as mbr_ipv_1d
     ,a.uv as mbr_ipv_uv_1d
     ,DATE_FORMAT(CURRENT_TIMESTAMP,'yyyy-MM-dd HH:mm:ss') as log_gmt_modify
     ,DATE_FORMAT(CURRENT_TIMESTAMP,'yyyy-MM-dd HH:mm:ss') as gmt_modify
   FROM itm_log_day a
   JOIN ec.dim_tm_cate_360_2pt_ri FOR SYSTEM_TIME AS OF PROCTIME() AS b
   ON vvp_dt_rtcdm.DateAddOrSub(a.stat_date, -2) = b.stat_date 
       AND a.cate_id = b.cate_id
   ;
   
   --書き込み
   INSERT INTO vvp_ec_ads.ads_ec_log_hh
   
   SELECT
     a.stat_date
     ,a.stat_hour
     ,cast(a.item_id as varchar) as item_id
     ,a.cate_id
     ,b.cate_name
     ,cast(b.industry_id as varchar) as industry_id
     ,cast(b.xcat1_id as varchar) as xcat1_id
     ,cast(b.xcat2_id as varchar) as xcat2_id
     ,cast(b.xcat3_id as varchar) as xcat3_id
     ,cast(b.cate_level1_id as varchar) as cate_level1_id
     ,cast(b.cate_level2_id as varchar) as cate_level2_id
     ,cast(b.is_sw as varchar) as is_sw
     ,a.pv as mbr_ipv_1h
     ,a.uv as mbr_ipv_uv_1h
     ,DATE_FORMAT(CURRENT_TIMESTAMP,'yyyy-MM-dd HH:mm:ss') as log_gmt_modify
     ,DATE_FORMAT(CURRENT_TIMESTAMP,'yyyy-MM-dd HH:mm:ss') as gmt_modify
   FROM itm_log_hh_00 a
   JOIN ec.dim_tm_cate_360_2pt_ri FOR SYSTEM_TIME AS OF PROCTIME() AS b
   ON vvp_ec_ads.DateAddOrSub(a.stat_date, -2) = b.stat_date 
       AND a.cate_id = b.cate_id
   ;
   
   END;

3. ノードエディターで、コードエディターの右側にある [リアルタイム構成] をクリックして、[Flink リソース情報]、[スクリプトパラメーター]、および [Flink 実行時パラメーター] を構成します。

   

   プリセット Flink SQL Stream リアルタイム構成 - エキスパートモードコード

   {
     "nodes": [
       {
         "profile": {
           "parallelism": 256,  -- 並列度
           "maxParallelism": 32768,  -- 最大並列度
           "minParallelism": 1,  -- 最小並列度
           "group": "0"  -- グループ
         },
         "id": 1,  -- ID
         "type": "StreamExecTableSourceScan",  -- タイプ
         "desc": "Source: vvp_dt_rtcdm_dwd_tb_trd_ord_pay_nrt_ri[71980]"  -- 説明
       },
       {
         "profile": {
           "parallelism": 256,  -- 並列度
           "maxParallelism": 32768,  -- 最大並列度
           "minParallelism": 1,  -- 最小並列度
           "group": "0"  -- グループ
         },
         "id": 2,  -- ID
         "type": "StreamExecCalc",  -- タイプ
         "desc": "Calc[71981]"  -- 説明
       },
       {
         "profile": {
           "parallelism": 256,  -- 並列度
           "maxParallelism": 32768,  -- 最大並列度
           "minParallelism": 1,  -- 最小並列度
           "group": "0"  -- グループ
         },
         "id": 3,  -- ID
         "type": "StreamExecLookupJoin",  -- タイプ
         "desc": "LookupJoin[71982]"  -- 説明
       },
       {
         "profile": {
           "parallelism": 256,  -- 並列度
           "maxParallelism": 32768,  -- 最大並列度
           "minParallelism": 1,  -- 最小並列度
           "group": "0"  -- グループ
         },
         "id": 4,  -- ID
         "type": "StreamExecCalc",  -- タイプ
         "desc": "Calc[71983]"  -- 説明
       },
       {
         "profile": {
           "parallelism": 256,  -- 並列度
           "maxParallelism": 32768,  -- 最大並列度
           "minParallelism": 1,  -- 最小並列度
           "group": "1"  -- グループ
         },
         "id": 6,  -- ID
         "state": [
           {
             "userDefined": false,  -- ユーザー定義かどうか
             "name": "groupAggregateState",  -- 名前
             "index": 0,  -- インデックス
             "ttl": "36 h"  -- TTL
           }
         ],
         "type": "StreamExecGroupAggregate",  -- タイプ
         "desc": "GroupAggregate[71985]"  -- 説明
       },
       {
         "profile": {
           "parallelism": 256,  -- 並列度
           "maxParallelism": 32768,  -- 最大並列度
           "minParallelism": 1,  -- 最小並列度
           "group": "1"  -- グループ
         },
         "id": 7,  -- ID
         "type": "StreamExecCalc",  -- タイプ
         "desc": "Calc[71986]"  -- 説明
       },
       {
         "profile": {
           "parallelism": 256,  -- 並列度
           "maxParallelism": 32768,  -- 最大並列度
           "minParallelism": 1,  -- 最小並列度
           "group": "1"  -- グループ
         },
         "id": 8,  -- ID
         "type": "StreamExecSink",  -- タイプ
         "desc": "ConstraintEnforcer[71987]"  -- 説明
       },
       {
         "profile": {
           "parallelism": 256,  -- 並列度
           "maxParallelism": 32768,  -- 最大並列度
           "minParallelism": 1,  -- 最小並列度
           "group": "2"  -- グループ
         },
         "id": 10,  -- ID
         "state": [
           {
             "userDefined": false,  -- ユーザー定義かどうか
             "name": "sinkMaterializeState",  -- 名前
             "index": 0,  -- インデックス
             "ttl": "36 h"  -- TTL
           }
         ],
         "type": "StreamExecSink",  -- タイプ
         "desc": "SinkMaterializer[71987]"  -- 説明
       },
       {
         "profile": {
           "parallelism": 256,  -- 並列度
           "maxParallelism": 32768,  -- 最大並列度
           "minParallelism": 1,  -- 最小並列度
           "group": "2"  -- グループ
         },
         "id": 11,  -- ID
         "type": "StreamExecSink",  -- タイプ
         "desc": "Sink: vvp_dt_ads_tb_dev_ads_tb_idec_seckill_cate_bc_trd_flow_htr_000[71987]"  -- 説明
       }
     ],
     "vertices": {  -- 頂点
       "2d95a2974e3b3137fd533ecfd3490bc5": [  -- ハッシュ値
         10,  -- ID
         11  -- ID
       ],
       "717c7b8afebbfb7137f6f0f99beb2a94": [  -- ハッシュ値
         1,  -- ID
         2,  -- ID
         3,  -- ID
         4  -- ID
       ],
       "44b79c13fdb45883c7f21ee510155f4d": [  -- ハッシュ値
         6,  -- ID
         7,  -- ID
         8  -- ID
       ]
     },
     "edges": [  -- エッジ
       {
         "mode": "PIPELINED",  -- モード
         "source": 1,  -- ソース
         "strategy": "FORWARD",  -- 戦略
         "target": 2  -- ターゲット
       },
       {
         "mode": "PIPELINED",  -- モード
         "source": 2,  -- ソース
         "strategy": "FORWARD",  -- 戦略
         "target": 3  -- ターゲット
       },
       {
         "mode": "PIPELINED",  -- モード
         "source": 3,  -- ソース
         "strategy": "FORWARD",  -- 戦略
         "target": 4  -- ターゲット
       },
       {
         "mode": "PIPELINED",  -- モード
         "source": 4,  -- ソース
         "strategy": "HASH",  -- 戦略
         "target": 6  -- ターゲット
       },
       {
         "mode": "PIPELINED",  -- モード
         "source": 6,  -- ソース
         "strategy": "FORWARD",  -- 戦略
         "target": 7  -- ターゲット
       },
       {
         "mode": "PIPELINED",  -- モード
         "source": 7,  -- ソース
         "strategy": "FORWARD",  -- 戦略
         "target": 8  -- ターゲット
       },
       {
         "mode": "PIPELINED",  -- モード
         "source": 8,  -- ソース
         "strategy": "HASH",  -- 戦略
         "target": 10  -- ターゲット
       },
       {
         "mode": "PIPELINED",  -- モード
         "source": 10,  -- ソース
         "strategy": "FORWARD",  -- 戦略
         "target": 11  -- ターゲット
       }
     ],
     "ssgProfiles": [  -- SSG プロファイル
       {
         "managed": {},  -- 管理対象
         "name": "0",  -- 名前
         "cpu": 0.25,  -- CPU
         "offHeap": "32 mb",  -- オフヒープ
         "heap": "992 mb",  -- ヒープ
         "extended": {}  -- 拡張
       },
       {
         "managed": {  -- 管理対象
           "STATE_BACKEND": "512 mb"  -- STATE_BACKEND
         },
         "name": "1",  -- 名前
         "cpu": 0.25,  -- CPU
         "offHeap": "32 mb",  -- オフヒープ
         "heap": "480 mb",  -- ヒープ
         "extended": {}  -- 拡張
       },
       {
         "managed": {  -- 管理対象
           "STATE_BACKEND": "512 mb"  -- STATE_BACKEND
         },
         "name": "2",  -- 名前
         "cpu": 0.25,  -- CPU
         "offHeap": "32 mb",  -- オフヒープ
         "heap": "480 mb",  -- ヒープ
         "extended": {}  -- 拡張
       }
     ]
   }

   プリセット Flink SQL Stream リアルタイム構成 - Flink 実行時パラメーター - その他の構成

   blob.fetch.backlog: 1000
   taskmanager.debug.memory.log-interval: 5000

4. リアルタイム設定を構成した後、コードエディターの上にある [保存] をクリックします。次に [公開] をクリックします。右下隅に表示される [公開コンテンツ] パネルで、[本番環境への公開を開始] をクリックし、プロンプトに従って確認します。

   

ステップ 4：個人開発環境に入る

個人開発環境は、カスタムコンテナイメージ、ユーザー NAS および Git への接続、Notebook を使用した Python でのプログラミングをサポートしています。

DataStudio で、ページ上部の アイコンをクリックします。ドロップダウンメニューで、入りたい個人開発環境を選択し、ページが読み込まれるのを待ちます。

ステップ 5: Python でプログラミングとデバッグを行う

DataWorks は DSW と深く統合されています。個人開発環境に入った後、DataStudio は Python コードの記述、デバッグ、実行、スケジューリングをサポートします。

1. DataStudio ページで、個人の開発者環境の workspace フォルダをクリックします。[個人用フォルダ] の右側にある アイコンをクリックします。左側のリストに無題のファイルが表示されます。ファイルに名前を付け、Enter キーを押し、生成されるのを待ちます。

   プリセットファイル名: ec_item_rec.py。

   

2. Python ファイルページのコードエディターに、プリセットの Python コードを入力します。次に、コードエディターの上にある [Python ファイルを実行] をクリックし、ページ下部の [ターミナル] で結果を確認します。

   

   

   プリセット Python コード

   import pandas as pd
   from surprise import Dataset, Reader, SVD
   from surprise.model_selection import train_test_split
   from surprise import accuracy
   
   # サンプルデータを作成
   data_dict = {
       'user_id': [1, 1, 1, 2, 2, 2, 3, 3, 4],
       'item_id': [101, 102, 103, 101, 104, 105, 102, 105, 101],
       'rating': [5, 3, 4, 2, 4, 5, 4, 5, 3]
   }
   
   # データを DataFrame に変換
   df = pd.DataFrame(data_dict)
   
   # Surprise ライブラリを使用してデータセットを準備
   reader = Reader(rating_scale=(1, 5))
   data = Dataset.load_from_df(df[['user_id', 'item_id', 'rating']], reader)
   
   # データセットをトレーニングセットとテストセットに分割
   trainset, testset = train_test_split(data, test_size=0.2)
   
   # SVD アルゴリズムをレコメンデーションに使用
   model = SVD()
   model.fit(trainset)
   
   # 予測を行う
   predictions = model.test(testset)
   
   # RMSE を計算
   rmse = accuracy.rmse(predictions)
   print(f'RMSE: {rmse:.2f}')
   
   # ユーザーにおすすめの商品を取得
   def get_recommendations(user_id, model, all_items, n=3):
       item_ids = all_items['item_id'].unique()
       user_item_col = all_items[(all_items['user_id'] == user_id)]['item_id']
       unseen_items = [item for item in item_ids if item not in user_item_col.values]
   
       # 未評価のアイテムの評価を予測
       predictions = []
       for item in unseen_items:
           pred = model.predict(user_id, item)
           predictions.append((item, pred.est))
   
       # 予測評価でソート
       predictions.sort(key=lambda x: x[1], reverse=True)
       return predictions[:n]
   
   # 商品のレコメンデーションを取得
   all_items = df
   user_id = 1  # レコメンデーションを取得するユーザーの ID
   recommendations = get_recommendations(user_id, model, all_items)
   
   print(f'Recommended products for user {user_id}:')
   for item_id, score in recommendations:
       print(f'Product ID: {item_id}, Predicted score: {score:.2f}')

   Python 環境のインストール

   pip install pandas scikit-surprise  # pandas と scikit-surprise をインストール

3. コードをデバッグするには、コードエディターの上にある [Python ファイルをデバッグ] をクリックするか、コードエディターの左側にあるパネルの アイコンをクリックします。行番号の左側をクリックしてブレークポイントを設定できます。

   

ステップ 6: Notebook でデータを探索する

Notebook のデータ探索操作は、個人開発環境で実行されます。開始する前に ステップ 4: 個人開発環境に入る を完了する必要があります。