Realtime Compute for Apache Flink:Paimon と StarRocks によるストリーミングレイクハウス

MySQL 変更データキャプチャ (CDC) データソースの準備

このチュートリアルでは、ApsaraDB RDS for MySQL を例に説明します。order_dw という名前のデータベースと、サンプルデータを含む 3 つのテーブルを作成します。

1. (非推奨。ステップ 1 をご参照ください。) ApsaraDB RDS for MySQL インスタンスの作成

   重要

   ApsaraDB RDS for MySQL インスタンスと Realtime Compute for Apache Flink ワークスペースが同じ VPC にあることを確認してください。異なる VPC にある場合は、「How do I access other services across VPCs?」をご参照ください。

2. (非推奨。ステップ 1 をご参照ください。) データベースとアカウントの作成

   order_dw という名前のデータベースを作成します。次に、order_dw データベースに対する読み取りおよび書き込み権限を持つ特権アカウント、または標準アカウントを作成します。

   3 つのテーブルを作成し、データを挿入します。

   CREATE TABLE `orders` (
     order_id bigint not null primary key,
     user_id varchar(50) not null,
     shop_id bigint not null,
     product_id bigint not null,
     buy_fee bigint not null,   
     create_time timestamp not null,
     update_time timestamp not null default now(),
     state int not null
   );
   CREATE TABLE `orders_pay` (
     pay_id bigint not null primary key,
     order_id bigint not null,
     pay_platform int not null, 
     create_time timestamp not null
   );
   CREATE TABLE `product_catalog` (
     product_id bigint not null primary key,
     catalog_name varchar(50) not null
   );
   -- データを準備します
   INSERT INTO product_catalog VALUES(1, 'phone_aaa'),(2, 'phone_bbb'),(3, 'phone_ccc'),(4, 'phone_ddd'),(5, 'phone_eee');
   INSERT INTO orders VALUES
   (100001, 'user_001', 12345, 1, 5000, '2023-02-15 16:40:56', '2023-02-15 18:42:56', 1),
   (100002, 'user_002', 12346, 2, 4000, '2023-02-15 15:40:56', '2023-02-15 18:42:56', 1),
   (100003, 'user_003', 12347, 3, 3000, '2023-02-15 14:40:56', '2023-02-15 18:42:56', 1),
   (100004, 'user_001', 12347, 4, 2000, '2023-02-15 13:40:56', '2023-02-15 18:42:56', 1),
   (100005, 'user_002', 12348, 5, 1000, '2023-02-15 12:40:56', '2023-02-15 18:42:56', 1),
   (100006, 'user_001', 12348, 1, 1000, '2023-02-15 11:40:56', '2023-02-15 18:42:56', 1),
   (100007, 'user_003', 12347, 4, 2000, '2023-02-15 10:40:56', '2023-02-15 18:42:56', 1);
   INSERT INTO orders_pay VALUES
   (2001, 100001, 1, '2023-02-15 17:40:56'),
   (2002, 100002, 1, '2023-02-15 17:40:56'),
   (2003, 100003, 0, '2023-02-15 17:40:56'),
   (2004, 100004, 0, '2023-02-15 17:40:56'),
   (2005, 100005, 0, '2023-02-15 18:40:56'),
   (2006, 100006, 0, '2023-02-15 18:40:56'),
   (2007, 100007, 0, '2023-02-15 18:40:56');

メタデータ管理

Paimon カタログの作成

1. Realtime Compute for Apache Flink コンソールにログインします。

2. 左側のナビゲーションペインで [Metadata Management] に移動し、[Create Catalog] をクリックします。

3. [Built-in Catalog] タブで [Apache Paimon] をクリックし、[Next] をクリックします。

4. 次のパラメーターを設定し、ストレージタイプとして DLF を選択して、[OK] をクリックします。

   パラメーター	説明	必須	備考
   metastore	metastore のタイプ。	はい	この例では DLF を選択します。
   catalog name	DLF データカタログの名前。 重要 RAM ユーザーまたは RAM ロールを使用する場合は、RAM ユーザーまたは RAM ロールに、DLF からデータを読み取り、DLF にデータを書き込むために必要な権限が付与されていることを確認してください。詳細については、「Manage data permissions」をご参照ください。	はい	DLF 2.5 を使用することを推奨します。このバージョンでは、AccessKey ペアなどの情報を入力する必要がなく、既存の DLF データカタログをすばやく選択できます。データカタログの作成方法については、「Data catalog」をご参照ください。 paimoncatalog という名前のデータカタログを作成し、リストから選択します。

5. MySQL の order_dw データベースからすべてのテーブルを同期するために、データカタログに order_dw という名前のデータベースを作成します。

   左側のナビゲーションペインで [Data Exploration] > [Query Editor] を選択し、[New] をクリックして一時的なクエリエディターを開きます。

   -- paimoncatalog データソースを使用します
   USE CATALOG paimoncatalog;
   -- order_dw データベースを作成します
   CREATE DATABASE order_dw;

   返されたメッセージ The following statement has been executed successfully! は、データベースが正常に作成されたことを示しています。

Paimon カタログの使用方法の詳細については、「Manage Paimon catalogs」をご参照ください。

MySQL カタログの作成

1. [Metadata Management] ページで [Create Catalog] をクリックします。

2. [Built-in Catalog] タブで [MySQL] をクリックし、[Next] をクリックします。

3. 次のパラメーターを設定し、[OK] をクリックして mysqlcatalog という名前の MySQL カタログを作成します。

   パラメーター	説明	必須	備考
   catalog name	カタログ名。	はい	任意の名前を入力します。このチュートリアルでは mysqlcatalog を使用します。
   hostname	MySQL データベースの IP アドレスまたはホスト名。	はい	詳細については、「View and manage instance connection endpoints and ports」をご参照ください。ApsaraDB RDS for MySQL インスタンスと Realtime Compute for Apache Flink ワークスペースが同じ VPC にあるため、内部エンドポイントを入力します。
   port	MySQL データベースサービスのポート番号。デフォルト値は 3306 です。	いいえ	詳細については、「View and manage instance connection endpoints and ports」をご参照ください。
   default-database	デフォルトの MySQL データベースの名前。	はい	同期するデータベース名を入力します。このチュートリアルでは order_dw です。
   username	MySQL データベースサービスのユーザー名。	はい	Prepare the MySQL CDC data source で作成したアカウントを使用します。
   password	MySQL データベースサービスのパスワード。	はい	Prepare the MySQL CDC data source で作成したアカウントのパスワードを使用します。

ODS レイヤーの構築：リアルタイムインジェスト

YAML 形式の Flink CDC データインジェストジョブを使用して、MySQL から Paimon にデータを同期し、1 ステップで ODS レイヤーを構築します。

1. YAML データインジェストジョブを作成して起動します。

   1. Realtime Compute for Apache Flink コンソールで [Data Development] > [Data Ingestion] ページに移動し、ods という名前の空の YAML ドラフトジョブを作成します。

   2. 次のコードをエディターにコピーし、ユーザー名やパスワードなどのパラメーターを更新します。

      source:
        type: mysql
        name: MySQL Source
        hostname: rm-bp1e********566g.mysql.rds.aliyuncs.com
        port: 3306
        username: ${secret_values.username}
        password: ${secret_values.password}
        tables: order_dw.\.*  # 正規表現を使用して、order_dw データベース内のすべてのテーブルを読み取ります。
        # (オプション) 増分フェーズで、新しく作成されたテーブルのデータを同期します。
        scan.binlog.newly-added-table.enabled: true
        # (オプション) テーブルとフィールドのコメントを同期します。
        include-comments.enabled: true
        # (オプション) 潜在的な TaskManager の OutOfMemory 問題を回避するため、無制限チャンクの配布を優先します。
        scan.incremental.snapshot.unbounded-chunk-first.enabled: true
        # (オプション) 読み取りを高速化するため、解析フィルターを有効にします。
        scan.only.deserialize.captured.tables.changelog.enabled: true
      sink:
        type: paimon
        name: Paimon Sink
        catalog.properties.metastore: rest
        catalog.properties.uri: http://cn-beijing-vpc.dlf.aliyuncs.com
        catalog.properties.warehouse: paimoncatalog
        catalog.properties.token.provider: dlf
      pipeline:
        name: MySQL to Paimon Pipeline

      パラメーター	説明	必須	例
      catalog.properties.metastore	metastore のタイプ。rest に設定します。	はい	rest
      catalog.properties.token.provider	トークンプロバイダー。DLF に設定します。	はい	DLF
      catalog.properties.uri	DLF Rest Catalog Server のアクセス URI は http://[region-id]-vpc.dlf.aliyuncs.com です。リージョン ID の詳細については、「サービスエンドポイント」をご参照ください。	はい	http://cn-beijing-vpc.dlf.aliyuncs.com
      catalog.properties.warehouse	DLF カタログの名前。	はい	paimoncatalog
      hostname	MySQL データベースの IP アドレスまたはホスト名。詳細については、「View and manage instance connection endpoints and ports」をご参照ください。ApsaraDB RDS for MySQL インスタンスと Realtime Compute for Apache Flink ワークスペースが同じ VPC にあるため、内部エンドポイントを入力します。	はい	rm-bp1e********566g.mysql.rds.aliyuncs.com
      username	MySQL データベースのユーザー名。シークレット管理を使用することを推奨します。詳細については、「Manage variables」をご参照ください。	はい	${secret_values.username}
      password	MySQL データベースのパスワード。シークレット管理を使用することを推奨します。詳細については、「Manage variables」をご参照ください。	はい	${secret_values.password}

      Paimon の書き込みパフォーマンスを最適化する方法については、「Paimon performance tuning」をご参照ください。

   3. 右上隅の [Deploy] をクリックします。

   4. [O&M Center] > [Job O&M] で、新しくデプロイした ODS ジョブの [Actions] 列にある [Start] をクリックし、[ステートレス起動]を選択してジョブを起動します。ジョブの起動設定の詳細については、「Job Startup」をご参照ください。

2. MySQL から Paimon に同期した 3 つのテーブルのデータを確認します。

   Realtime Compute for Apache Flink コンソールで [Data Development] > [Data Exploration] ページに移動します。[Query Editor] タブで次のコードをエディターにコピーし、ステートメントを選択してから、右上隅の [Run] をクリックします。

   SELECT * FROM paimoncatalog.order_dw.orders ORDER BY order_id;

   クエリの実行後、orders テーブルから 7 件の注文レコードが返されます。結果には、order_id (100001 ～ 100007)、user_id、shop_id、product_id、buy_fee (1000 ～ 5000)、create_time、update_time、state 列が含まれます。state の値は、すべてのレコードで 1 です。

DWD レイヤーの構築：テーマ別ワイドテーブル

1. dwd_orders ワイドテーブルの作成

   Realtime Compute for Apache Flink コンソールで、データ開発 > データ探索 ページに移動します。[クエリエディター] タブで、次のコードをエディターにコピーし、ステートメントを選択して、右上隅の [実行] をクリックします。

   CREATE TABLE paimoncatalog.order_dw.dwd_orders (
       order_id BIGINT,
       order_user_id STRING,
       order_shop_id BIGINT,
       order_product_id BIGINT,
       order_product_catalog_name STRING,
       order_fee BIGINT,
       order_create_time TIMESTAMP,
       order_update_time TIMESTAMP,
       order_state INT,
       pay_id BIGINT,
       pay_platform INT COMMENT 'プラットフォーム 0: phone、1: pc',
       pay_create_time TIMESTAMP,
       PRIMARY KEY (order_id) NOT ENFORCED
   ) WITH (
       'merge-engine' = 'partial-update', -- 部分更新マージエンジンを使用してワイドテーブルを生成します
       'changelog-producer' = 'lookup' -- lookup 変更ログプロデューサーを使用して、低レイテンシーの変更ログを生成します
   );

   The following statement has been executed successfully! というメッセージは、作成が成功したことを示します。

2. ODS 変更データの消費

   Realtime Compute for Apache Flink コンソールで、[データ開発] > [ETL]ページに移動します。dwd という名前の新しい SQL ストリーミングジョブを作成し、次のコードを SQL エディターにコピーして [デプロイ] をクリックし、[ステートレス起動] を使用してジョブを開始します。

   この SQL ジョブは、ディメンションテーブル結合で orders テーブルと product_catalog テーブルを結合します。次に、その結果と orders_pay テーブルからのデータを dwd_orders テーブルに書き込みます。Paimon の部分更新マージメカニズムは、同じ order_id を持つ orders テーブルと orders_pay テーブルのレコードを結合します。

   SET 'execution.checkpointing.max-concurrent-checkpoints' = '3';
   SET 'table.exec.sink.upsert-materialize' = 'NONE';
   SET 'execution.checkpointing.interval' = '10s';
   SET 'execution.checkpointing.min-pause' = '10s';
   -- 現在、Paimon は単一ジョブ内で同一テーブルへの複数の INSERT 文をサポートしていません。そのため、ここでは UNION ALL を使用します。
   INSERT INTO paimoncatalog.order_dw.dwd_orders 
   SELECT 
       o.order_id,
       o.user_id,
       o.shop_id,
       o.product_id,
       dim.catalog_name,
       o.buy_fee,
       o.create_time,
       o.update_time,
       o.state,
       NULL,
       NULL,
       NULL
   FROM
       paimoncatalog.order_dw.orders o 
       LEFT JOIN paimoncatalog.order_dw.product_catalog FOR SYSTEM_TIME AS OF proctime() AS dim
       ON o.product_id = dim.product_id
   UNION ALL
   SELECT
       order_id,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       pay_id,
       pay_platform,
       create_time
   FROM
       paimoncatalog.order_dw.orders_pay;

3. dwd_orders のデータ確認

   Realtime Compute for Apache Flink コンソールで、[データ開発] > [データ探索] ページに移動します。[クエリエディター] タブで、次のコードをエディターにコピーし、ステートメントを選択してから、右上隅にある [実行] をクリックします。

   SELECT * FROM paimoncatalog.order_dw.dwd_orders ORDER BY order_id;

   クエリが正常に実行されると、dwd_orders ワイドテーブルから 7 件の注文レコードが返されます。結果には 12 のフィールドが含まれます：order_id、order_user_id、order_shop_id、order_product_id、order_product_catalog_name、order_fee、order_create_time、order_update_time、order_state、pay_id、pay_platform、pay_create_time。order_id は 100001 ～ 100007 の範囲で、注文金額は 1000 ～ 5000 の範囲です。また、すべての作成時刻は 2023-02-15 です。

DWS レイヤーの構築：メトリック計算

1. DWS レイヤーの集計テーブル dws_users と dws_shops を作成します。

   Realtime Compute for Apache Flink コンソールで、[データ開発] > [データ探索] ページに移動します。 [クエリエディター] タブで、次のコードをエディターにコピーし、ステートメントを選択して、右上隅にある [実行] をクリックします。

   -- ユーザーディメンションの集計テーブル。
   CREATE TABLE paimoncatalog.order_dw.dws_users (
       user_id STRING,
       ds STRING,
       paid_buy_fee_sum BIGINT COMMENT '当日の支払総額',
       PRIMARY KEY (user_id, ds) NOT ENFORCED
   ) WITH (
       'merge-engine' = 'aggregation', -- 集計マージエンジンを使用して集計テーブルを生成します。
       'fields.paid_buy_fee_sum.aggregate-function' = 'sum' -- paid_buy_fee_sum のデータを合計して結果を集計します。
       -- dws_usersテーブルは下流のストリーミングジョブで消費されないため、変更ログプロデューサーを指定する必要はありません。
   );
   -- ショップディメンションの集計テーブル。
   CREATE TABLE paimoncatalog.order_dw.dws_shops (
       shop_id BIGINT,
       ds STRING,
       paid_buy_fee_sum BIGINT COMMENT '当日の支払総額',
       uv BIGINT COMMENT '当日のユニーク購入ユーザー数',
       pv BIGINT COMMENT '当日のユーザーによる総購入数',
       PRIMARY KEY (shop_id, ds) NOT ENFORCED
   ) WITH (
       'merge-engine' = 'aggregation', -- 集計マージエンジンを使用して集計テーブルを生成します。
       'fields.paid_buy_fee_sum.aggregate-function' = 'sum', -- paid_buy_fee_sum のデータを合計して結果を集計します。
       'fields.uv.aggregate-function' = 'sum', -- uv のデータを合計して結果を集計します。
       'fields.pv.aggregate-function' = 'sum' -- pv のデータを合計して結果を集計します。
       -- dws_shopsテーブルは下流のストリーミングジョブで消費されないため、変更ログプロデューサーを指定する必要はありません。
   );
   -- ユーザーディメンションとショップディメンションの集計テーブルを同時に計算するため、ユーザーとショップを主キーとする中間テーブルを作成します。
   CREATE TABLE paimoncatalog.order_dw.dwm_users_shops (
       user_id STRING,
       shop_id BIGINT,
       ds STRING,
       paid_buy_fee_sum BIGINT COMMENT 'あるユーザーがあるショップで当日に支払った合計金額',
       pv BIGINT COMMENT 'あるユーザーがあるショップで当日に購入した回数',
       PRIMARY KEY (user_id, shop_id, ds) NOT ENFORCED
   ) WITH (
       'merge-engine' = 'aggregation', -- 集計マージエンジンを使用して集計テーブルを生成します。
       'fields.paid_buy_fee_sum.aggregate-function' = 'sum', -- paid_buy_fee_sum のデータを合計して結果を集計します。
       'fields.pv.aggregate-function' = 'sum', -- pv のデータを合計して結果を集計します。
       'changelog-producer' = 'lookup', -- lookup 変更ログプロデューサーを使用して、低レイテンシーの変更ログを生成します。
       -- DWM レイヤーの中間テーブルは通常、上流のアプリケーションから直接クエリされないため、書き込みパフォーマンスを最適化できます。
       'file.format' = 'avro', -- Avro の行ベースストレージフォーマットを使用すると、より効率的な書き込みパフォーマンスが得られます。
       'metadata.stats-mode' = 'none' -- 統計を省略すると OLAP クエリのコストは増加しますが (継続的なストリーム処理には影響なし)、書き込みパフォーマンスは向上します。
   );

   The following statement has been executed successfully! というメッセージは、作成が成功したことを示します。

2. DWD レイヤーのテーブル dwd_orders から変更データを消費します。

   Realtime Compute for Apache Flink コンソールで、[[データ開発 > ETL]] ページに移動し、dwm という名前の新しい SQL ストリーミングジョブを作成します。 次のコードを SQL エディターにコピーし、[デプロイ] をクリックしてから、[ステートレス開始] を使用してジョブを開始します。

   この SQL ジョブは、dwd_orders テーブルのデータを dwm_users_shops テーブルに書き込みます。Paimon の事前集約データマージメカニズムは、order_fee を自動的に合計して、ユーザーがショップで支払った合計金額を算出します。また、値 1 を合計して、ユーザーがそのショップで購入した回数を算出します。

   SET 'execution.checkpointing.max-concurrent-checkpoints' = '3';
   SET 'table.exec.sink.upsert-materialize' = 'NONE';
   SET 'execution.checkpointing.interval' = '10s';
   SET 'execution.checkpointing.min-pause' = '10s';
   INSERT INTO paimoncatalog.order_dw.dwm_users_shops
   SELECT
       order_user_id,
       order_shop_id,
       DATE_FORMAT (pay_create_time, 'yyyyMMdd') as ds,
       order_fee,
       1 -- 1 つの入力レコードが 1 回の購入を表します
   FROM paimoncatalog.order_dw.dwd_orders
   WHERE pay_id IS NOT NULL AND order_fee IS NOT NULL;

3. DWM レイヤーのテーブル dwm_users_shops から変更データをリアルタイムで消費します。

   Realtime Compute for Apache Flink コンソールで、[データ開発] > [ETL] ページに移動します。dws という名前の新しい SQL ストリーミングジョブを作成します。次のコードを SQL エディターにコピーし、[デプロイ] をクリックして、[ステートレス起動] を使用してジョブを開始します。

   この SQL ジョブは、dwm_users_shops テーブルのデータを dws_users テーブルと dws_shops テーブルに書き込みます。Paimon の事前集約データマージメカニズムを使用して、dws_users テーブルでは各ユーザーの合計支出額 (paid_buy_fee_sum) を算出します。dws_shops テーブルでは、ショップの総売上 (paid_buy_fee_sum)、購入ユーザー数 (1 の合計)、総購入回数 (pv) を算出します。

   SET 'execution.checkpointing.max-concurrent-checkpoints' = '3';
   SET 'table.exec.sink.upsert-materialize' = 'NONE';
   SET 'execution.checkpointing.interval' = '10s';
   SET 'execution.checkpointing.min-pause' = '10s';
   -- DWD ジョブとは異なり、ここでの各 INSERT 文は異なる Paimon テーブルに書き込むため、同じジョブに含めることができます。
   BEGIN STATEMENT SET;
   INSERT INTO paimoncatalog.order_dw.dws_users
   SELECT 
       user_id,
       ds,
       paid_buy_fee_sum
   FROM paimoncatalog.order_dw.dwm_users_shops;
   -- 主キーはショップ ID です。人気のあるショップのデータは、他のショップよりもはるかに大きくなる可能性があります。
   -- したがって、ローカルマージを使用してメモリ内でデータを事前集計してから Paimon に書き込み、データスキューを軽減します。
   INSERT INTO paimoncatalog.order_dw.dws_shops /*+ OPTIONS('local-merge-buffer-size' = '64mb') */
   SELECT
       shop_id,
       ds,
       paid_buy_fee_sum,
       1, -- 1 つの入力レコードが、このショップでのあるユーザーのすべての購入を表します
       pv
   FROM paimoncatalog.order_dw.dwm_users_shops;
   END;

4. dws_users テーブルと dws_shops テーブルのデータを確認します。

   Realtime Compute for Apache Flink コンソールで、[データ開発] > [データ探索] ページに移動します。[クエリエディター] タブで、次のコードをエディターにコピーし、ステートメントを選択して、右上隅の [実行] をクリックします。

   -- dws_users テーブルのデータを表示します
   SELECT * FROM paimoncatalog.order_dw.dws_users ORDER BY user_id;

   クエリの実行後、dws_users テーブルから、user_id、ds、paid_buy_fee_sum 列を含む 3 件のレコードが返されます。返されるレコードは user_001 / 20230215 / 8000、user_002 / 20230215 / 5000、user_003 / 20230215 / 5000 です。

   -- dws_shops テーブルのデータを表示します
   SELECT * FROM paimoncatalog.order_dw.dws_shops ORDER BY shop_id;

   dws_shops テーブルのクエリ後、結果には 4 行 5 列が含まれます：shop_id、ds、paid_buy_fee_sum、uv、pv。shop_id は 12345 ～ 12348 で、ds はすべてのレコードで 20230215 です。paid_buy_fee_sum は 5000、4000、7000、2000、uv は 1、1、2、2、pv はそれぞれ 1、1、3、2 です。

業務データベースからの変更のキャプチャ

ストリーミングデータレイクハウスを構築した後、業務データベースからの変更をキャプチャする機能をテストします。

1. MySQL の order_dw データベースに次のデータを挿入します。

   INSERT INTO orders VALUES
   (100008, 'user_001', 12345, 3, 3000, '2023-02-15 17:40:56', '2023-02-15 18:42:56', 1),
   (100009, 'user_002', 12348, 4, 1000, '2023-02-15 18:40:56', '2023-02-15 19:42:56', 1),
   (100010, 'user_003', 12348, 2, 2000, '2023-02-15 19:40:56', '2023-02-15 20:42:56', 1);
   INSERT INTO orders_pay VALUES
   (2008, 100008, 1, '2023-02-15 18:40:56'),
   (2009, 100009, 1, '2023-02-15 19:40:56'),
   (2010, 100010, 0, '2023-02-15 20:40:56');

2. dws_users テーブルと dws_shops テーブルのデータを表示します。Realtime Compute for Apache Flink コンソールで、データ開発 > データ探索 ページに移動します。[クエリエディタ] タブで、次のコードをエディタにコピーし、ステートメントを選択して、右上隅の [実行] をクリックします。

   • dws_users テーブル

     SELECT * FROM paimoncatalog.order_dw.dws_users ORDER BY user_id;

     クエリでは、user_id、ds、paid_buy_fee_sum 列を含む 3 行が返されます。返されるレコードは user_001 / 20230215 / 11000、user_002 / 20230215 / 6000、user_003 / 20230215 / 7000 です。

   • dws_shops テーブル

     SELECT * FROM paimoncatalog.order_dw.dws_shops ORDER BY shop_id;

     クエリでは、shop_id、ds、paid_buy_fee_sum、uv、pv の 5 列を含む 4 行が返されます。shop_id は 12345、12346、12347、12348 で、ds はすべてのレコードで 20230215 です。paid_buy_fee_sum は 8000、4000、7000、5000、uv は 1、1、2、3、pv はそれぞれ 2、1、3、4 です。