DataWorks:DataWorks OpenLake ソリューションに基づくエンドツーエンドのインテリジェント レイクハウス データ開発

DataWorks の概要

DataWorks は、15 年にわたるビッグデータの経験に基づく Alibaba グループのビッグデータ開発手法を活用した、インテリジェントなレイクハウスベースのデータ開発およびガバナンスプラットフォームです。 DataWorks は、MaxCompute、E-MapReduce（EMR）、Hologres、Realtime Compute for Apache Flink、Platform for AI（PAI）など、Alibaba Cloud が提供する数十のビッグデータおよび AI コンピューティングサービスと互換性があります。 DataWorks は、データウェアハウス、データレイク、OpenLake レイクハウスアーキテクチャ向けのインテリジェントな抽出、変換、ロード（ETL）開発、データ分析、プロアクティブなデータ資産ガバナンスをサポートし、Data+AI ライフサイクル全体でデータ管理を促進します。 2009 年以来、DataWorks は Alibaba データシステムを継続的に製品化および改良し、公共サービス部門、金融、小売、インターネット、自動車、製造などのさまざまな業界にサービスを提供してきました。 DataWorks は、デジタルトランスフォーメーションと価値創造のために DataWorks を選択した数万の顧客の信頼を獲得しています。

DataWorks Copilot の概要

DataWorks Copilot は、DataWorks のインテリジェントアシスタントです。 DataWorks では、デフォルトの大規模言語モデル（LLM）、DeepSeek-R1-671B（フルパワーエディション）、または DeepSeek-R1-Distill-Qwen-32B を使用して、DataWorks での関連操作を完了できます。 DeepSeek-R1 の高度な推論能力により、DataWorks Copilot では、自然言語インタラクションに基づいて、SQL コードの生成、最適化、テストなどの複雑なタスクを実行できます。 これにより、抽出、変換、ロード（ETL）開発とデータ分析の効率が大幅に向上します。

DataWorks ノートブック機能の概要

DataWorks のノートブック機能は、エンジン固有の SQL または Python コード分析を実行し、コードをリアルタイムで実行またはデバッグするために使用できる、インテリジェントなインタラクティブデータ開発および分析ツールを提供します。 このようにして、視覚化されたデータ処理結果を取得できます。 さらに、ノートブックと他のタイプのノードを組み合わせてワークフローを形成し、ワークフローをスケジューリングシステムにコミットして実行できます。 これにより、複雑なビジネスシナリオを柔軟に実装できます。

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

データレイクハウスソリューションのデータカタログ管理機能を使用すると、Data Lake Formation（DLF）、MaxCompute、Hologres などのコンピュートエンジンのタイプのデータカタログを作成および管理できます。

1. Data Studio ページの左側のナビゲーションウィンドウで、 アイコンをクリックします。[DATA CATALOG] ウィンドウが表示されます。[DATA CATALOG] ウィンドウで、管理するメタデータタイプを見つけ、目的のデータカタログの名前にポインターを移動し、データカタログの右側にある アイコンをクリックして、[開く] をクリックします。データカタログの構成タブが表示されます。

   

   

2. データカタログの構成タブで、スキーマの名前をクリックします。 表示されるタブで、テーブルの名前をクリックします。 テーブルの詳細ページが表示されます。

   

3. Data Studioデータカタログテーブルの作成 ページの左側のナビゲーションウィンドウで、 アイコンをクリックします。 ウィンドウが表示されます。 [データカタログ] ウィンドウで、管理するメタデータタイプを見つけ、目的のデータカタログの名前にポインタを移動し、データカタログの右側にある アイコンをクリックして、 をクリックします。 [テーブルの作成] タブが表示されます。

   

4. [テーブルの作成] タブの左側のエリアで、[テーブル名] と [フィールド名] を指定します。または、[テーブルの作成] タブの右側のエリアに [DDL] 文を入力してテーブルを作成します。次に、上部のツールバーの [デプロイ] をクリックします。

   

   サンプルプリセットコード

   CREATE TABLE dwd_mbr_user_info
   (
     id                BIGINT COMMENT 'プライマリキー', -- The primary key
     gmt_create       DATETIME COMMENT '作成時間', -- The creation time
     gmt_modified     DATETIME COMMENT '変更時間', -- The modification time
     user_id          BIGINT COMMENT 'メンバーのユーザー ID', -- The user ID of a member
     nick             STRING COMMENT 'メンバーのニックネーム', --  The nickname of a member
     reg_fullname     STRING COMMENT '個人認証が実行された場合は個人の実名を使用し、企業認証が実行された場合は企業名を使用します。', -- The real name of a person is used if individual verification is performed, and an enterprise name is used if enterprise verification is performed.
     reg_mobile_phone STRING COMMENT '登録時に入力された携帯電話番号', -- The mobile phone number entered during registration
     reg_email        STRING COMMENT '登録時に入力されたメールアドレス（ユーザーが変更可能）', -- The email address entered during registration (modifiable by the user)
     reg_gender       STRING COMMENT '登録時に入力された性別（F は女性、M は男性を表します。どちらも入力されていない場合、性別は不明であり、これは性別を非公開にすることを示します。）', -- The gender entered during registration (F represents female and M represents male. If neither one is entered, the gender is unknown, which indicates that the person wants to keep the gender confidential.)
     reg_birthdate    DATETIME COMMENT '登録時に入力された誕生日（ユーザーが変更可能）', -- The birthday entered during registration (modifiable by the user)
     reg_address      STRING COMMENT '登録時に入力された住所（ユーザーが変更可能）', -- The address entered during registration (modifiable by the user)
     reg_nation_id    STRING COMMENT '登録時に入力された国 ID（空のまま）', -- The country ID entered during registration (left empty)
     reg_prov_id      STRING COMMENT '登録時に入力された都道府県 ID', -- The province ID entered during registration
     reg_city_id      STRING COMMENT '登録時に入力された市区町村 ID', -- The city ID entered during registration
     user_regip       STRING COMMENT '登録時に入力された IP アドレス', -- The IP address entered during registration
     id_card_type     BIGINT COMMENT 'メンバーシップ認証に使用される証明書のタイプ。 0：不明。 1：ID カード番号。 2：営業許可証番号', -- The certificate type used for membership authentication. 0: unknown; 1: ID card number; 2: business license number
     id_card_number   STRING COMMENT '個人認証が実行された場合は、ID カード番号を示します。 企業認証が実行された場合は、営業許可証番号を示します。 認証が実行されていない場合、精度は保証されません。', -- If individual verification is performed, it indicates the ID card number. If enterprise verification is performed, it indicates the business license number. If no verification is performed, accuracy cannot be guaranteed.
     user_regdate     DATETIME COMMENT '登録時間', -- The registration time
     user_active_type STRING COMMENT 'ユーザーのアクティベーション方法。 1：メール。 2：携帯電話', -- The user activation method. 1: email; 2: mobile phone
     user_active_time DATETIME COMMENT 'アクティベーション時間', -- The activation time
     vip_level        STRING COMMENT 'VIP ランク', -- The VIP rank
     is_delete        STRING COMMENT '削除するかどうか' -- Delete or not
   )
   COMMENT 'メンバーシップ登録フォーム' -- The membership registration form
   PARTITIONED BY 
   (
     ds                STRING COMMENT 'YYYYMMDD'
   )
   LIFECYCLE 7;

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

DataWorks では、視覚的にドラッグ操作を実行して、ビジネスの観点からワークフロー内のさまざまなタイプのデータ開発ノードを調整できます。 スケジュール時間などの共通パラメータを個別に構成する必要はありません。 これにより、複雑なタスクプロジェクトを簡単に管理できます。

1. Data StudioDATASTUDIOワークスペースディレクトリ[ワークフローの作成] ページの左側のナビゲーションウィンドウで、 アイコンをクリックします。 ウィンドウが表示されます。 [DATASTUDIO] ウィンドウで、 をクリックし、右側にある アイコンをクリックして、 をクリックします。

   

2. 表示されるポップオーバーに、ワークフロー名を入力し、Enter キーを押します。

   プリセットワークフロー名: 小売 E コマースビジネスの概要

3. ワークフローの構成タブで、キャンバスの中央にある [ノードを追加するには、ドラッグまたはクリックします] をクリックします。[ノードの作成] ダイアログボックスで、[ノードタイプ] パラメーターと [ノード名] パラメーターを構成し、[確認] をクリックします。

   • プリセットノード名: 小売 E コマースの概要

   • プリセットノードタイプ: ゼロロードノード

   

4. ワークフローの構成タブの [ノードの作成] セクションで、使用するノードタイプを見つけ、ノードタイプをキャンバスにドラッグして、リリースします。 [ノードの作成] ダイアログボックスで、[ノード名] を指定し、[確認] をクリックします。

   次の表は、プリセットノード名とタイプを示しています。

   ノードタイプ	ノード名
   Data Integration - バッチ同期	ods_mbr_user_info
   MaxCompute-MaxCompute SQL	dim_ec_mbr_user_info
   MaxCompute-MaxCompute SQL	dws_ec_mbr_cnt_nd
   ノートブック	ads_ec_kpi_report

   

   

5. キャンバスで、依存関係を構成する 2 つのノードを見つけ、一方のノードの長方形ボックスの下端の中央にポインタを移動します。 + アイコンが表示されたら、接続線をドラッグして他のノードに接続し、リリースします。

   

   

6. 必要なノードを順番に作成し、ノードの依存関係を構成して、保存 をクリックします。

   

7. 構成を 保存 した後、ビジネス要件に基づいて、トップ ツールバーでキャンバス上のノードの レイアウト モード を変更できます。

   

8. ワークフローの構成タブの右側のナビゲーションウィンドウで、[プロパティ] をクリックします。[プロパティ] タブで、ワークフローのスケジューリングパラメーターとノードの依存関係を構成します。[スケジューリング パラメーター] セクションで、[パラメーターの追加] をクリックします。表示される入力ボックスで、[パラメーター名] フィールドに bizdate と入力し、[パラメーター値] ドロップダウンリストから $[yyyymmdd-1] を選択します。

   

9. タブの [スケジューリングの依存関係] セクションで、[依存関係の追加] をクリックします。表示されるセクションの [祖先オブジェクト] フィールドに、ワークフローに属していないノードの名前を入力し、Enter キーを押します。結果が返されるまで待ちます。結果リストから必要な祖先オブジェクトを選択し、[追加] をクリックします。

   

10. 上部のツールバーで、[デプロイ] をクリックします。 右下に [デプロイメント] タブが表示されます。 [デプロイメントの説明] フィールドの右側にある [本番環境へのデプロイを開始] をクリックし、画面上の指示に基づいてチェックと確認操作を順番に実行します。

    

ステップ 3：マルチエンジン共同開発を実行する

Data Studio は、データ同期ノード、MaxCompute、Hologres、EMR、Flink、ADB などの数十種類のコンピュートエンジンのノード、およびノートブックと Python を使用して開発されたノードのウェアハウス開発をサポートしています。 Data Studio は、これらのノードの複雑なスケジューリング構成もサポートしています。 Data Studio は、開発 - 本番環境分離の R&D モードを提供します。 この実験では、Flink SQL Streaming ノードが作成されます。

1. Data StudioDATASTUDIOワークスペースディレクトリFlink SQL ストリーミングノード名 ページの左側のナビゲーションウィンドウで、 アイコンをクリックします。 ウィンドウが表示されます。 [DATASTUDIO] ウィンドウで、 をクリックし、右側にある アイコンをクリックして、[ノードの作成] > [Flink] > を選択します。 表示されるポップオーバーに、 を入力し、Enter キーを押します。

   プリセットノード名: ads_ec_page_visit_log

   

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

   

   Flink SQL Streaming ノードのプリセットコード

   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,  -- 訪問者 ID
     item_id,  -- 商品 ID
     cate_id,  -- カテゴリ ID
     ext_pv  -- 拡張 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,  -- PV
     count(DISTINCT visitor_id) FILTER (WHERE ext_pv>0) as uv,  -- UV
     stat_date,  -- 統計日付
     cate_id,  -- カテゴリ ID
     item_id  -- 商品 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,  -- PV
     count(DISTINCT visitor_id) FILTER (WHERE ext_pv>0) as uv,  -- UV
     stat_date,  -- 統計日付
     stat_hour,  -- 統計時間
     item_id,  -- 商品 ID
     cate_id  -- カテゴリ 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,  -- 商品 ID
     a.cate_id,  -- カテゴリ ID
     b.cate_name,  -- カテゴリ名
     cast(b.industry_id as varchar) as industry_id,  -- 業種 ID
     cast(b.xcat1_id as varchar) as xcat1_id,  -- xcat1 ID
     cast(b.xcat2_id as varchar) as xcat2_id,  -- xcat2 ID
     cast(b.xcat3_id as varchar) as xcat3_id,  -- xcat3 ID
     cast(b.cate_level1_id as varchar) as cate_level1_id,  -- カテゴリレベル 1 ID
     cast(b.cate_level2_id as varchar) as cate_level2_id,  -- カテゴリレベル 2 ID
     cast(b.is_sw as varchar) as is_sw,  -- ソフトウェアかどうか
     a.pv as mbr_ipv_1d,  -- 1 日あたりのメンバー IPV
     a.uv as mbr_ipv_uv_1d,  -- 1 日あたりのメンバー IPV UV
     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
   ;
   
   -- データの書き込み -- Write data.
   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,  -- 商品 ID
     a.cate_id,  -- カテゴリ ID
     b.cate_name,  -- カテゴリ名
     cast(b.industry_id as varchar) as industry_id,  -- 業種 ID
     cast(b.xcat1_id as varchar) as xcat1_id,  -- xcat1 ID
     cast(b.xcat2_id as varchar) as xcat2_id,  -- xcat2 ID
     cast(b.xcat3_id as varchar) as xcat3_id,  -- xcat3 ID
     cast(b.cate_level1_id as varchar) as cate_level1_id,  -- カテゴリレベル 1 ID
     cast(b.cate_level2_id as varchar) as cate_level2_id,  -- カテゴリレベル 2 ID
     cast(b.is_sw as varchar) as is_sw,  -- ソフトウェアかどうか
     a.pv as mbr_ipv_1h,  -- 1 時間あたりのメンバー IPV
     a.uv as mbr_ipv_uv_1h,  -- 1 時間あたりのメンバー IPV UV
     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 ランタイム パラメーター をクリックします。 表示されるタブで、、、 を構成します。 次の図は、パラメータ値を示しています。

   

   エキスパートモードのプリセットコード

   {
     "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": {}  -- 拡張
       }
     ]
   }

   その他のプリセット構成

   blob.fetch.backlog: 1000  -- バックログをフェッチする Blob
   taskmanager.debug.memory.log-interval: 5000  -- TaskManager デバッグメモリログ間隔

4. [リアルタイム構成] タブでパラメータを構成した後、上部のツールバーで [保存] をクリックし、次に [デプロイ] をクリックします。 右下に [デプロイメント] タブが表示されます。 [デプロイメントの説明] フィールドの右側にある [本番環境へのデプロイを開始] をクリックし、画面上の指示に基づいてチェックと確認操作を順番に実行します。

   

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

個人開発環境では、カスタムコンテナイメージ、NAS ファイルシステム（NAS）および Git への接続、Python ベースのプログラミングとノートブックがサポートされています。

Data Studio[個人用開発環境を選択] ページの上部のナビゲーションバーで、 をクリックし、入力する個人開発環境を選択します。

ステップ 5：Python コードを記述してデバッグする

DataWorks は DSW と緊密に統合されています。 Data Studio は、個人開発環境での Python コードの記述、デバッグ、スケジューリング、および実行をサポートしています。

1. Data Studioワークスペース個人用ディレクトリ ページの選択した個人開発環境で、 の下の をクリックし、 アイコンをクリックします。 名前のないファイルがリストに追加されます。 プリセットファイル名を入力し、Enter キーを押して、ファイルが生成されるまで待ちます。

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

   

2. Python ファイルの構成タブにあるコードエディタに、プリセット Python コードを入力します。 上部のツールバーで、[python ファイルの実行] を選択します。 構成タブの下部にある [ターミナル] タブで、実行結果をクエリします。

   

   

   プリセット Python コード

   import pandas as pd  # pandas をインポート
   from surprise import Dataset, Reader, SVD  # surprise から Dataset、Reader、SVD をインポート
   from surprise.model_selection import train_test_split  # surprise.model_selection から train_test_split をインポート
   from surprise import accuracy  # surprise から accuracy をインポート
   
   # サンプルデータを作成します。 -- Create sample data.
   data_dict = {
       'user_id': [1, 1, 1, 2, 2, 2, 3, 3, 4],  # ユーザー ID
       'item_id': [101, 102, 103, 101, 104, 105, 102, 105, 101],  # 商品 ID
       'rating': [5, 3, 4, 2, 4, 5, 4, 5, 3]  # 評価
   }
   
   # データエントリを DataFrame に変換します。 -- Convert data entries into a DataFrame.
   df = pd.DataFrame(data_dict)
   
   # Surprise ライブラリを使用してデータセットを準備します。 -- Prepare a dataset by using the Surprise library.
   reader = Reader(rating_scale=(1, 5))  # 評価スケールを設定
   data = Dataset.load_from_df(df[['user_id', 'item_id', 'rating']], reader)  # DataFrame からデータセットを読み込み
   
   # データセットをトレーニングセットとテストセットに分割します。 -- Split the dataset into a training set and a test set.
   trainset, testset = train_test_split(data, test_size=0.2)  # テストサイズを 0.2 に設定
   
   # 特異値分解（SVD）アルゴリズムに基づいて商品を推奨します。 -- Recommend products based on the Singular Value Decomposition (SVD) algorithm.
   model = SVD()  # SVD モデルを作成
   model.fit(trainset)  # トレーニングセットでモデルをトレーニング
   
   # 予測を行います。 -- Make predictions.
   predictions = model.test(testset)  # テストセットで予測を行う
   
   # 平均二乗誤差（RMSE）を計算します。 -- Calculate the root mean square error (RMSE).
   rmse = accuracy.rmse(predictions)  # RMSE を計算
   print(f'RMSE: {rmse:.2f}')  # RMSE を出力
   
   # ユーザーに推奨される商品を取得します。 -- Obtain the products that are recommended to a user.
   def get_recommendations(user_id, model, all_items, n=3):  # 推奨商品を取得する関数
       item_ids = all_items['item_id'].unique()  # すべての商品 ID を取得
       user_item_col = all_items[(all_items['user_id'] == user_id)]['item_id']  # ユーザーが購入した商品 ID を取得
       unseen_items = [item for item in item_ids if item not in user_item_col.values]  # ユーザーがまだ購入していない商品 ID を取得
   
       # 見えない商品のスコアを予測します。 -- Predict the scores of unseen products.
       predictions = []  # 予測結果を格納するリスト
       for item in unseen_items:  # ユーザーがまだ購入していない商品ごとに
           pred = model.predict(user_id, item)  # 予測スコアを計算
           predictions.append((item, pred.est))  # 予測結果をリストに追加
   
       # 予測スコアで商品をソートします。 -- Sort products by the predicted scores.
       predictions.sort(key=lambda x: x[1], reverse=True)  # 予測スコアでソート
       return predictions[:n]  # 上位 n 件の商品を返す
   
   # 商品の推奨を取得します。 -- Obtain the product recommendations.
   all_items = df  # すべての商品
   user_id = 1  # 推奨対象ユーザーの ID -- The ID of the user to whom the products are recommended.
   recommendations = get_recommendations(user_id, model, all_items)  # 推奨商品を取得
   
   print(f'The products that are recommended to {user_id}:')  # 推奨商品を出力
   for item_id, score in recommendations:  # 推奨商品ごとに
       print(f'Product ID: {item_id}, Predicted score: {score:.2f}')  # 商品 ID と予測スコアを出力
   

   Python 環境のインストール

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

3. Python ファイルの構成タブの上部のツールバーで、[python ファイルのデバッグ] を選択します。 コードエディタでコード行番号にポインタを移動します。 赤い点が表示されます。 赤い点をクリックして、ブレークポイントを追加します。 [DATASTUDIO] ウィンドウの上にある左上で、 をクリックしてコードをデバッグします。

   

ステップ 6：ノートブックに基づいてデータを探索する

ノートブックに基づいてデータを探索する場合、関連する操作は個人開発環境で実行されます。 したがって、この手順の操作を実行する前に、ステップ 4：個人開発環境に入る を完了する必要があります。