統合レイクハウス分析: MaxLake を使用した複数のユースケースに対応するデータ処理とオーケストレーション - MaxCompute

ユースケースの概要

このトピックでは、MaxLake を使用して統合データレイクハウスアーキテクチャを実装する方法について説明します。これには、データインジェスト、データウェアハウジング、および複数の分析ユースケースのオーケストレーションが含まれます。このワークフローでは、IoV (Internet of Vehicles) データを例として、車載端末からレポートされた GPS 位置データに基づいて車両の走行距離と速度を分析する方法を示します。複数のコンピュートエンジンを統合することで、単一のデータソースから多面的な価値を引き出し、リアルタイムのオンライン分析処理 (OLAP) レポート、動的マスキングによるチーム間の安全なデータ共有、AI モデルトレーニングといった多様なユースケースをサポートできます。全体のアーキテクチャとプロセスを以下に示します。

自動データ検出 (ODS レイヤー)

生データは Object Storage Service (OSS) に格納されます。この例では、車両端末が GPS 位置情報を含む生ファイル (parquet および orc) を時間単位でパーティション分割してアップロードします。

MaxCompute で CONNECTION を作成し、外部ストレージの認証情報を管理します。
次に、DataScan タスクを使用してこれらのファイルを自動的に検出し、構造化された外部テーブル (ODS レイヤー) として登録します。

増分データ処理 (DWD および ADS レイヤー)

ODS レイヤーから無効なデータをフィルター処理し、フィールド形式を統一します。例として、時間値のフォーマットや経度・緯度座標の検証などがあります。

複数のユースケースに対応するマルチエンジンオーケストレーション

オンライン分析処理 (OLAP) + AI：処理済みデータに StarRocks エンジンを接続し、サブ秒レベルのリアルタイムクエリとダッシュボード化 (例：全車両の総走行距離の可視化) を実現します。
チーム間の安全なコラボレーション：MaxCompute のきめ細かいセキュリティを使用して、Spark エンジンを使用するサードパーティチームとデータを共有します。
AI トレーニング：クレンジングされた高品質データは、PyTorch などのフレームワークを使用したモデルトレーニングの基盤となります。
車両 ID などの機密フィールドには動的データマスキングが適用され、コラボレーション中のプライバシーコンプライアンスを確保します。

操作手順

テストデータのアップロード
1. Object Storage Service (OSS) コンソールにログインします。
2. 左側のナビゲーションウィンドウで、バケット をクリックします。
  バケット ページで、バケットの作成 をクリックします。
  この例では、バケット名は vehicle-raw です。
3. バケット ページで、対象の バケット名 をクリックして オブジェクト ページに移動します。
  アップロード をクリックして、解凍したテストデータファイル Maxlake_example_parquet.zip をアップロードします。
権限の付与
1. RAM ユーザー (root アカウントではない) を使用して接続を管理する場合、テナントレベルの Connection_Admin ロールが必要です。詳細については、「テナントレベルのロール権限付与」をご参照ください。
2. Alibaba Cloud アカウント、またはテナントレベルの Super_Administrator および Admin ロールを持つユーザーは、Connection_Admin ロールを付与できます。
  1. MaxCompute コンソールにログインし、左上隅でリージョンを選択します。
  2. 左側のナビゲーションウィンドウで、構成の管理 > テナント管理 を選択します。
  3. テナント管理 ページで、ロール管理 タブをクリックします。
  4. ロール管理 タブで、Connection_Admin と Datascan_Admin を選択し、操作列の 新しい権限付与 をクリックします。
  5. 追加の承認 ダイアログボックスで、権限を付与するユーザーを追加し、OK をクリックします。

MaxCompute での CONNECTION の作成

MaxCompute コンソールにログインし、左上隅でリージョンを選択します。
左側のナビゲーションウィンドウで、マックスレイク > データ接続 を選択します。
データ接続（コネクション） ページで、接続を作成する をクリックします。

接続の作成 ダイアログボックスで、次のパラメーターを設定し、OK をクリックします。

パラメーター	説明
データ接続名	データレイク接続の名前。名前はテナント内で一意である必要があります。
RAMRoleARN	OSS へのアクセス権限を持つ RAM ロールの RAMRoleARN を選択します。カスタムロールを作成し、その RAMRoleARN を指定できます。詳細については、「STS モードの権限付与」をご参照ください。
データ接続の説明	データレイク接続の説明。

DataScan タスクの作成

[MaxCompute コンソール]にログインし、左上隅でリージョンを選択します。
左側のナビゲーションウィンドウで、マックスレイク > データ検出 を選択します。
データ検出 ページで、データ発見タスクの作成 をクリックします。

タスクの作成 ダイアログボックスで、次のパラメーターを設定し、作成をクリックします。

基本構成
パラメーター
説明
タスク名
タスクの名前。名前はテナント内で一意である必要があります。
タスクサイクル
5 分

湖のデータ設定

パラメーター

説明

接続

外部ストレージのアクセス認証情報として CONNECTION を選択します。

場所

データファイルが格納されている OSS パスを入力します。

フォーマット：oss://<Bucket name>/<OSS path>/
OSS バケットと MaxCompute DataScan タスクは、同じリージョンにあり、同じ Alibaba Cloud アカウントに属している必要があります。
例：oss://vehicle-raw/Maxlake_example_parquet

カタログ設定

パラメーター

説明

項目

外部テーブルが作成される MaxCompute プロジェクトを選択します。

スキーマ

スキーマを選択します。

スキーマに、DataScan が検出すると予想されるフォルダー/ファイルと同じ名前のユーザー作成テーブルが既に含まれていないことを確認してください。名前の競合が存在する場合、外部テーブルは作成されません。

データ検出タスクは 5 分ごとに実行されます。タスクが実行されるのを待ちます。成功したら、IoV テーブルが作成され、パーティションが自動的に追加されていることを確認します。
説明
- スキャンが失敗した場合は、OSS バケットに .DS_Store のようなシステムファイルがないか確認してください。これらは解析エラーの原因となる可能性があります。手動で削除する必要があります。
- 将来のバージョンでは、このようなファイルの自動除外がサポートされる予定です。

SQL データ処理スクリプト。MaxCompute SQL エディターで次の SQL スクリプトを実行します。

ODS レイヤー (生データ) の検証

-- ビジネスシナリオの設計
-- シナリオ: IoV プラットフォームがリアルタイムの車両位置データ (GPS) を取得し、各車両の時間ごとの走行距離と平均速度を計算します。

-- データソース: 車載端末が数秒ごとに位置、速度などのデータをレポートします。
-- ODS: レポートされた生データを直接格納します。
-- DWD: データをクレンジング、解析、変換して個別のレポートレコードにします。
-- ADS: 各車両の時間ごとの総走行距離と平均速度を計算します。

-- ODS: 生のレポートデータテーブル
-- IoV の生の位置データを時間単位でパーティション分割して格納します。

SHOW PARTITIONS ods_vehicle_gps_raw;
SET odps.sql.allow.fullscan=true;

SELECT  * FROM ods_vehicle_gps_raw WHERE dt='2025-09-17' AND hh='23';

-- サンプル結果
+------------+------------+-------------+------------+------------+------------+-------------+------------+------------+
| vin        | device_id  | report_time | lng        | lat        | speed      | raw_payload | dt         | hh         | 
+------------+------------+-------------+------------+------------+------------+-------------+------------+------------+
| VIN001     | DEV001     | 2025-09-16 00:00:00 | 120.00023573730152 | 30.39975989605289 | 73.3843581906447 | mock_payload | 2025-09-17 | 23         | 
| VIN002     | DEV002     | 2025-09-16 00:00:00 | 120.00517998985256 | 30.33811818824062 | 67.43035716350673 | mock_payload | 2025-09-17 | 23         | 
| VIN003     | DEV003     | 2025-09-16 00:00:00 | 120.24295999679852 | 30.143229002199707 | 40.8918776553552 | mock_payload | 2025-09-17 | 23         | 
| VIN001     | DEV001     | 2025-09-16 00:30:00 | 120.24754980497414 | 30.373484773735274 | 49.50436236779409 | mock_payload | 2025-09-17 | 23         | 
| VIN002     | DEV002     | 2025-09-16 00:30:00 | 120.00510501582413 | 30.42486370328109 | 55.8400627485663 | mock_payload | 2025-09-17 | 23         | 
| VIN003     | DEV003     | 2025-09-16 00:30:00 | 120.36073125682805 | 30.065016013833237 | 61.82996654036919 | mock_payload | 2025-09-17 | 23         | 
| VIN002     | DEV002     | 2025-09-16 19:30:00 | 120.29488938268968 | 30.12474152125639 | 66.48209032904454 | mock_payload | 2025-09-17 | 23         | 
| VIN002     | DEV002     | 2025-09-16 19:00:00 | 120.35157954057287 | 30.459823299646295 | 76.36574370617315 | mock_payload | 2025-09-17 | 23         | 
| VIN001     | DEV001     | 2025-09-16 19:30:00 | 120.3113710027241 | 30.33402715522518 | 62.601762741153024 | mock_payload | 2025-09-17 | 23         | 
+------------+------------+-------------+------------+------------+------------+-------------+------------+------------+

DWD レイヤー (データクレンジング) の生成

-- DWD レイヤー (dwd_vehicle_gps)
-- ODS レイヤーに基づいて、このレイヤーは無効なデータをフィルター処理し、時間フォーマットや経度・緯度の検証など、フィールド形式を統一します。
CREATE TABLE IF NOT EXISTS dwd_vehicle_gps (
  vin          STRING COMMENT '一意の車両識別子',
  event_time   DATETIME COMMENT 'レポート時間',
  lng          DOUBLE COMMENT '経度',
  lat          DOUBLE COMMENT '緯度',
  speed        DOUBLE COMMENT '速度 (km/h)',
  loc_valid    BOOLEAN COMMENT '位置が有効かどうかを示します'
)
PARTITIONED BY (dt STRING, hh STRING);

-- 変換 SQL
INSERT OVERWRITE TABLE dwd_vehicle_gps PARTITION (dt='2025-09-17', hh='23')
SELECT
vin,
TO_DATE(report_time,'yyyy-MM-dd HH:mi:ss') AS event_time,
lng, lat,
speed,
CASE WHEN lng BETWEEN 70 AND 140 AND lat BETWEEN 10 AND 60 THEN TRUE ELSE FALSE END AS loc_valid
FROM ods_vehicle_gps_raw
WHERE dt='2025-09-17' AND hh='23'
AND speed >= 0
AND vin IS NOT NULL;

SELECT  * FROM dwd_vehicle_gps WHERE dt='2025-09-17' AND hh='23';

-- サンプル結果
+------------+------------+------------+------------+------------+-----------+------------+------------+
| vin        | event_time | lng        | lat        | speed      | loc_valid | dt         | hh         | 
+------------+------------+------------+------------+------------+-----------+------------+------------+
| VIN001     | 2025-09-16 00:00:00 | 120.00023573730152 | 30.39975989605289 | 73.3843581906447 | true      | 2025-09-17 | 23         | 
| VIN002     | 2025-09-16 00:00:00 | 120.00517998985256 | 30.33811818824062 | 67.43035716350673 | true      | 2025-09-17 | 23         | 
| VIN003     | 2025-09-16 00:00:00 | 120.24295999679852 | 30.143229002199707 | 40.8918776553552 | true      | 2025-09-17 | 23         | 
| VIN001     | 2025-09-16 00:30:00 | 120.24754980497414 | 30.373484773735274 | 49.50436236779409 | true      | 2025-09-17 | 23         | 
| VIN003     | 2025-09-16 00:30:00 | 120.36073125682805 | 30.065016013833237 | 61.82996654036919 | true      | 2025-09-17 | 23         | 
| VIN001     | 2025-09-16 05:00:00 | 120.13891993725622 | 30.39267490566367 | 53.99676876794396 | true      | 2025-09-17 | 23         | 
| VIN003     | 2025-09-16 05:30:00 | 120.04798104849084 | 30.012209889484666 | 65.01092831837522 | true      | 2025-09-17 | 23         | 
| VIN002     | 2025-09-16 20:00:00 | 120.42721760246307 | 30.051330581564144 | 79.73892066615583 | true      | 2025-09-17 | 23         | 
| VIN003     | 2025-09-16 20:00:00 | 120.47715870033818 | 30.302941456112517 | 58.61057150112957 | true      | 2025-09-17 | 23         | 
| VIN001     | 2025-09-16 20:30:00 | 120.3067564206695 | 30.179763514166588 | 47.77533756931095 | true      | 2025-09-17 | 23         | 
+------------+------------+-------------+------------+------------+------------+-------------+------------+------------+

ADS レイヤー (集約) の生成

-- ADS レイヤー (ads_vehicle_hourly_stat)
-- 各車両の時間ごとの総走行距離 (単純な位置差分計算を使用) と平均速度を計算します。
CREATE TABLE IF NOT EXISTS ads_vehicle_hourly_stat (
  vin           STRING COMMENT '一意の車両識別子',
  stat_hour     STRING COMMENT '統計時間 (yyyy-MM-dd HH)',
  total_distance DOUBLE COMMENT '総走行距離 (km)',
  avg_speed     DOUBLE COMMENT '平均速度 (km/h)'
)
PARTITIONED BY (dt STRING, hh STRING);

-- 統計変換 SQL
-- 注: ここではデモンストレーションのために簡略化された数式が使用されています。実際のシナリオでは、ハバーサイン公式を使用できます。
-- 経度と緯度の差に基づく単純な距離の近似 (デモンストレーション用)。
SET odps.sql.type.system.odps2=true;

WITH ordered AS (
  SELECT
  vin, event_time, lng, lat, speed,
  ROW_NUMBER() OVER (PARTITION BY vin ORDER BY event_time) AS rn
  FROM dwd_vehicle_gps
  WHERE dt='2025-09-17' AND hh='23' AND loc_valid = TRUE
),
with_prev AS (
  SELECT
  a.vin, a.event_time, a.speed,
  -- ユークリッド距離を使用した単純な近似。ここでは、経度または緯度の 1 度の差が約 111 km であるとします。これは正確ではありませんが、このデモには十分です。
  ABS(a.lng - b.lng)*111 AS dx,
  ABS(a.lat - b.lat)*111 AS dy
  FROM ordered a
  LEFT JOIN ordered b
  ON a.vin = b.vin AND a.rn = b.rn + 1
)
INSERT OVERWRITE TABLE ads_vehicle_hourly_stat PARTITION (dt='2025-09-17', hh='23')
  SELECT
  vin,
  '2025-09-17 23' AS stat_hour,
  SUM( SQRT( COALESCE(dx,0)*COALESCE(dx,0) + COALESCE(dy,0)*COALESCE(dy,0) ) ) AS total_distance,
  AVG(speed) AS avg_speed
  FROM with_prev
  GROUP BY vin;

-- 車両 VIN001 の走行距離と平均速度をクエリします。
SET odps.sql.allow.fullscan=true;
SELECT * FROM ads_vehicle_hourly_stat WHERE vin='VIN001'
  ORDER BY stat_hour DESC;
  
-- 結果
+------------+---------------+--------------------+-------------------+------------+------------+
| vin        | stat_hour     | total_distance     | avg_speed         | dt         | hh         | 
+------------+---------------+--------------------+-------------------+------------+------------+
| VIN001     | 2025-09-17 23 | 1510.7384548492398 | 59.33624859907179 | 2025-09-17 | 23         | 
+------------+---------------+--------------------+-------------------+------------+------------+

SET odps.sql.allow.fullscan=true;
SELECT
vin AS vehicle_id,
stat_hour AS stat_time,
CONCAT(CAST(ROUND(total_distance, 2) AS STRING), ' km') AS mileage,
CONCAT(CAST(ROUND(avg_speed, 2) AS STRING), ' km/h') AS avg_speed
FROM ads_vehicle_hourly_stat WHERE vin='VIN001'
  ORDER BY stat_hour DESC;
  
-- 結果
+------------+---------------+------------+------------+
| vehicle_id | stat_time     | mileage    | avg_speed  | 
+------------+---------------+------------+------------+
| VIN001     | 2025-09-17 23 | 1510.74 km | 59.34 km/h | 
+------------+---------------+------------+------------+

チーム間のロールを確立し、クエリ権限を付与します。

-- チーム間のコラボレーションと安全なデータ共有。Spark エンジンをデータ分析に使用するサードパーティチームに対して、動的データマスキングを有効にします。
CREATE role thirdparty;
GRANT CreateInstance, List ON project <project_name> TO ROLE thirdparty;
GRANT SELECT ON TABLE ods_vehicle_gps_raw TO ROLE thirdparty;

ADD USER RAM$<your Alibaba Cloud account>;
GRANT thirdparty TO RAM$<your Alibaba Cloud account>;

車両の総走行距離テーブルを非識別化します。先頭と末尾の文字のみを保持し、他の文字をアスタリスク (*) に置き換えることで、一意の車両 ID をマスクします。

-- プロジェクトのデータマスキング機能を有効にします。
setproject odps.data.masking.policy.enable=true;

-- 車両の総走行距離テーブルを非識別化します。先頭と末尾の文字のみを保持し、他の文字をアスタリスク (*) に置き換えることで、一意の車両 ID をマスクします。
CREATE data masking policy IF NOT EXISTS masking_vin
TO role (thirdparty)
USING MASKED_STRING_UNMASKED_BA(1, 1);

apply data masking policy masking_vin bind TO
TABLE ads_vehicle_hourly_stat COLUMN vin;

パラメーター	説明
タスク名	タスクの名前。名前はテナント内で一意である必要があります。
タスクサイクル	5 分