Reading from and writing to object storage on WarehousePG#

ストレージの場所またはIcebergカタログを使用してオブジェクトストレージからデータを読み取り、オプションでSpark Connectで高速化し、CTASを使用して新しいデータをオブジェクトストレージに書き込みます。

ストレージの場所を使用したオブジェクトストレージの照会#

ストレージ場所はオブジェクトストレージに直接接続し、Parquet、Delta Lake、およびIcebergファイルをサポートします。コーディネーターで保存場所とPGAAテーブルを作成し、標準のSQLを使用してデータを照会します。 WarehousePGは、最も効率の良い実行モードを自動的に選択し、クエリーの性質に応じて、コーディネーターでクエリーを実行またはセグメントホスト全体に分散します。

!!!注ストレージ場所またはIcebergカタログで作成されたテーブルは読み取り専用で、データは既にオブジェクトストレージに存在する必要があります。新しいデータをオブジェクトストレージに書き込むには、 CTASを使用したオブジェクトストレージへの書き込みを参照してください。

保存場所の構成の詳細については、 Reading tables in object storage を参照してください。

例#

TPC-H SF1データを含むパブリックS3バケットに接続し、3つのテーブルを登録し、分析クエリを実行して、DirectScan純粋なPGAAとCompatScanネイティブヒープテーブルと結合したPGAAの両方をデモンストレーションします。

コーディネーターで保存場所を作成します。

SELECT pgfs.create_storage_location(
  sample-data,
  s3://beacon-analytics-demo-data-us-east-1-prod,
  {"aws_skip_signature": "true"}
);

データを指すPGAAテーブルを作成します。

CREATE TABLE customer () USING PGAA
WITH (pgaa.storage_location = sample-data, pgaa.path = tpch_sf_1/customer);

CREATE TABLE orders () USING PGAA
WITH (pgaa.storage_location = sample-data, pgaa.path = tpch_sf_1/orders);

CREATE TABLE lineitem () USING PGAA
WITH (pgaa.storage_location = sample-data, pgaa.path = tpch_sf_1/lineitem);

分析クエリを実行します。

SELECT
    l_orderkey,
    sum(l_extendedprice * (1 - l_discount)) AS revenue,
    o_orderdate,
    o_shippriority
FROM
    customer,
    orders,
    lineitem
WHERE
    c_mktsegment = BUILDING
    AND c_custkey = o_custkey
    AND l_orderkey = o_orderkey
    AND o_orderdate < DATE 1995-03-15
    AND l_shipdate > DATE 1995-03-15
GROUP BY
    l_orderkey,
    o_orderdate,
    o_shippriority
ORDER BY
    revenue DESC,
    o_orderdate
LIMIT 10;

EXPLAIN を使用して、クエリプランを観察します。 PGAAは最初にDirectScanを試行し、コーディネーター上のSeafowl に実行を完全にオフロードします。プランには、述語プッシュダウン、集計、ソートを含む完全なSeafowl 論理プランが表示されます。 WarehousePGプランナーが関係していないため、 Optimizer 行が表示されないことに注意してください。

                                      QUERY PLAN
--------------------------------------------------------------------------------------------
 `Seafowl`DirectScan: Logical Plan
   Limit:  skip=0, fetch=10
     Sort:  revenue DESC NULLS FIRST, orders.o_orderdate ASC NULLS LAST
       Projection:  lineitem.l_orderkey, sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount) AS revenue, orders.o_orderdate, orders.o_shippriority
         Aggregate:  groupBy=[[lineitem.l_orderkey, orders.o_orderdate, orders.o_shippriority]], aggr=[[sum(lineitem.l_extendedprice * (Int64(1) - lineitem.l_discount))]]
           Filter:  customer.c_mktsegment = Utf8("BUILDING") AND customer.c_custkey = orders.o_custkey AND lineitem.l_orderkey = orders.o_orderkey AND orders.o_orderdate < CAST(Utf8("1995-03-15") AS Date32) AND lineitem.l_shipdate > CAST(Utf8("1995-03-15") AS Date32)
             Cross Join:
               Cross Join:
                 TableScan:  customer
                 TableScan:  orders
               TableScan:  lineitem
(11 rows)

PGAAは、クエリーにネイティブヒープテーブルとの結合など、Seafowl が処理できない操作が含まれる場合、CompatScanにフォールバックします。例、 customer をネイティブsegments テーブルと結合する場合

CREATE TABLE segments (seg_id INT, seg_name TEXT) DISTRIBUTED BY (seg_id);
INSERT INTO segments VALUES (1, BUILDING), (2, MACHINERY);

EXPLAIN SELECT c_name, seg_name
FROM customer
JOIN segments ON c_mktsegment = seg_name;

PGAAは最初にDirectScanを試行し、次の通知を使用してCompatScanにフォールバックします。

NOTICE:  PGAA DirectScan failed: QueryPlanningError("Error during planning: table default.public.segments not found"). Running the query in compatibility mode.
                                               QUERY PLAN
--------------------------------------------------------------------------------------------------------
 Gather Motion 4:1  (slice1; segments: 4)  (cost=1414.00..246348.40 rows=748960 width=64)
   ->  Hash Join  (cost=1414.00..236986.40 rows=187240 width=64)
         Hash Cond: (customer.c_mktsegment = segments.seg_name)
         ->  Custom Scan (`Seafowl`CompatScan) on customer  (cost=25.00..1131.25 rows=37500 width=64)
               `Seafowl`Plan: Logical Plan
                 Projection:  r1.c_name, r1.c_mktsegment
                   SubqueryAlias:  r1
                     TableScan:  public.customer
               `Seafowl`Query: select r1.c_name, r1.c_mktsegment from public.customer as r1
         ->  Hash  (cost=769.00..769.00 rows=49600 width=32)
               ->  Broadcast Motion 4:4  (slice2; segments: 4)  (cost=0.00..769.00 rows=49600 width=32)
                     ->  Seq Scan on segments  (cost=0.00..149.00 rows=12400 width=32)
 Optimizer: Postgres-based planner
(13 rows)

Gather Motion 4:1 は、Strewnローカスプランを示します。 `Seafowl CompatScanは、4つのセグメントホストの各ホストで並列実行され、それぞれが customerデータのスライスを読み取り、結果がコーディネーターに収集されます。ヒープテーブルとの結合は、WarehousePGプランナーによって処理されます。下部の Optimizer: Postgres-based planner行は、 Seafowl`がすべてを独立して処理するDirectScanとは異なり、 Postgresプランナーがプラン全体を担当したことを確認します。

PGAAがGeneral遺伝子座とStrewn遺伝子座を選択する方法の詳細については、クエリ実行モードを参照してください。

Icebergカタログに接続する#

WarehousePGコーディネーターに外部Iceberg RESTカタログを接続して、クラスター全体でカタログ管理テーブルを照会可能にします。

WarehousePGでPGAAを使用して、 EDB Postgres DistributedPGDがオブジェクトストレージにレプリケートしたデータを読み取ることができます。 PGDは、Iceberg形式のデータを書き込み、Iceberg RESTカタログに登録します。WarehousePGは、データを移動せずにMPPスケールで接続および照会できます。この設定のPGD側については、 PGDを使用した複製を参照してください。

!!!注 pgaa.attach_catalog() は、メンテナンスワーカーと同じ方法で、WarehousePGコーディネーターでのみメタストア同期ワーカーを実行します。 PGAAカタログテーブルはレプリケートテーブルポリシーを使用するため、すべてのノードで同期されたメタデータを利用できます。

資格情報構成やサポートされているカタログタイプを含む完全なセットアップ手順については、 Integrating with Iceberg catalogs を参照してください。

例#

コーディネーターにカタログを登録します。

SELECT pgaa.add_catalog(
  my_catalog,
  iceberg-rest,
  {"url": "https://iceberg-catalog.example.com"}
);

テーブルのメタデータをインポートします。

SELECT pgaa.import_catalog(my_catalog);

オプションで、インポートを特定の名前空間に制限します。

SELECT pgaa.import_catalog(my_catalog, my_schema);

インポートされると、テーブルはWarehousePGセッションから照会できます。例で説明されているのと同じクエリプランパターンが適用されます。 EXPLAIN を使用して、PGAAがDirectScanまたはCompatScanを使用しているか、およびどの遺伝子座が選択されているかを観察します。

Sparkでの加速#

Spark Connectを介してクエリーを外部Apache Sparkクラスターにルーティングします。このモードでは、コーディネーターはクエリーをSparkに直接転送し、セグメントホストは実行に関与せず、SparkはWarehousePGプランナーとは無関係に独自の計画と配布を処理します。

完全なセットアップ手順、GPUアクセラレーションオプション、およびパフォーマンスガイダンスについては、 Accelerating with Spark を参照してください。

例#

pgaa.spark_connect_url を設定し、コーディネーターの実行エンジンを切り替えます。

SET pgaa.executor_engine = spark_connect;
SET pgaa.spark_connect_url = sc://spark-connect-hostname:15002;

例と同じパブリックTPC-Hサンプルデータを使用して、ストレージ場所とPGAAテーブルを作成します。 Spark ConnectはParquetのみをサポートしているため、 pgaa.format を明示的に設定する必要があります。

SELECT pgfs.create_storage_location(
  sample-data,
  s3://beacon-analytics-demo-data-us-east-1-prod,
  {"aws_skip_signature": "true"}
);

CREATE TABLE customer () USING PGAA
WITH (pgaa.storage_location = sample-data, pgaa.path = tpch_sf_1/customer, pgaa.format = parquet);

CREATE TABLE orders () USING PGAA
WITH (pgaa.storage_location = sample-data, pgaa.path = tpch_sf_1/orders, pgaa.format = parquet);

CREATE TABLE lineitem () USING PGAA
WITH (pgaa.storage_location = sample-data, pgaa.path = tpch_sf_1/lineitem, pgaa.format = parquet);

同じTPC-H Q3クエリーを実行します。コーディネーターは実行をSparkに転送します。

SELECT
    l_orderkey,
    sum(l_extendedprice * (1 - l_discount)) AS revenue,
    o_orderdate,
    o_shippriority
FROM
    customer,
    orders,
    lineitem
WHERE
    c_mktsegment = BUILDING
    AND c_custkey = o_custkey
    AND l_orderkey = o_orderkey
    AND o_orderdate < DATE 1995-03-15
    AND l_shipdate > DATE 1995-03-15
GROUP BY
    l_orderkey,
    o_orderdate,
    o_shippriority
ORDER BY
    revenue DESC,
    o_orderdate
LIMIT 10;

EXPLAIN を使用して、クエリプランを観察します。 Sparkがアクティブなエンジンの場合、PGAAはSparkDirectScan ノードを生成します。このプランは、Sparkクラスター内で完全に処理される述語プッシュダウン、結合、集計を含むSpark独自の実行を示します。 Optimizer 行は表示されず、 WarehousePGプランナーが関与していないことを確認します。

                                          QUERY PLAN
--------------------------------------------------------------------------------------------
SparkDirectScan: Logical Plan
  AdaptiveSparkPlan:  isFinalPlan=false
  +- TakeOrderedAndProject: (limit=10, orderBy=[revenue DESC, o_orderdate ASC])
      +- HashAggregate: (keys=[l_orderkey, o_orderdate, o_shippriority], functions=[sum(...)])
        +- HashAggregate: (keys=[l_orderkey, o_orderdate, o_shippriority], functions=[partial_sum(...)])
            +- Project:  [o_orderdate, o_shippriority, l_orderkey, l_extendedprice, l_discount]
              +- SortMergeJoin:  [o_orderkey], [l_orderkey], Inner
                  :- Sort [o_orderkey ASC NULLS FIRST]
                  :  +- Exchange hashpartitioning(o_orderkey, 200)
                  :     +- BroadcastHashJoin [c_custkey], [o_custkey], Inner, BuildLeft
                  :        :- BroadcastExchange
                  :        :  +- Filter (c_mktsegment = BUILDING)
                  :        :     +- FileScan parquet [c_custkey, c_mktsegment]
                  :        :           Location: s3a://beacon-analytics-demo-data-us-east-1-prod/tpch_sf_1/customer
                  :        +- Filter (o_orderdate < 1995-03-15)
                  :           +- FileScan parquet [o_orderkey, o_custkey, o_orderdate, o_shippriority]
                  :                 Location: s3a://beacon-analytics-demo-data-us-east-1-prod/tpch_sf_1/orders
                  +- Sort [l_orderkey ASC NULLS FIRST]
                    +- Exchange hashpartitioning(l_orderkey, 200)
                        +- Filter (l_shipdate > 1995-03-15)
                          +- FileScan parquet [l_orderkey, l_extendedprice, l_discount, l_shipdate]
                                Location: s3a://beacon-analytics-demo-data-us-east-1-prod/tpch_sf_1/lineitem
(22 rows)

完全なSpark実行プランを検査するには、ブラウザーでhttp://<spark-host>:4040 のSpark UIを開きます。 [ SQL ]タブに移動して、ステージ境界、シャッフル操作、ステージごとのメトリックを含むクエリプランの視覚的な内訳を表示します。

CTASを使用したオブジェクトストレージへの書き込み#

CREATE TABLE AS SELECT CTASを使用して、スタンドアロンストレージ場所またはIcebergカタログをターゲットに、他のPGAAインストールと同じ構文でクエリ結果をオブジェクトストレージに書き込みます。単一のライタ、WarehousePG 7.5より後のコーディネーター、またはWarehousePG 7.5以前の1つの指定されたセグメントが、すべてのセグメントに並列して書き込むのではなく、データを書き込みます。 PGAAは、結果のテーブルに、クラスター全体で一貫性を保つ分散ポリシーを提供するため、他のPGAAテーブルと同様に照会および結合できます。

CTAS構文、保存場所、およびIcebergカタログの詳細については、 Writing to object storage を参照してください。

例#

分散WarehousePGテーブルを作成し、CTASを使用してオブジェクトストレージにエクスポートし、結果を元のテーブルに対して結合できることを確認します。

コーディネーターに保存場所を作成します。資格情報には、宛先バケットの読み取りおよび書き込み権限が必要です。

SELECT pgfs.create_storage_location(
  my_lake_data,
  s3://your-bucket-name/path,
  {"aws_region": "us-east-1", "aws_access_key_id": "...", "aws_secret_access_key": "..."}
);

次のファンクションを実行して、資格情報に書き込みアクセスがあることを確認します。成功した場合はNULL を結果ます。

SELECT pgaa.test_storage_location(my_lake_data, true);

分散ソーステーブルを作成し、CTASでエクスポートします。

CREATE TABLE orders_local (order_id int, amount numeric) DISTRIBUTED BY (order_id);
INSERT INTO orders_local SELECT i, i * 10 FROM generate_series(1, 100) AS i;

CREATE TABLE orders_export
USING PGAA
WITH (
    pgaa.storage_location = my_lake_data,
    pgaa.path = path/in/s3/orders_export,
    pgaa.format = iceberg
) AS
SELECT * FROM orders_local;

行数を確認し、元の分散テーブルに対して新しいテーブルを結合します。

SELECT count(*) FROM orders_export;

SELECT count(*)
FROM orders_export e
JOIN orders_local l ON e.order_id = l.order_id;

Reading from and writing to object storage on WarehousePG

目次

Reading from and writing to object storage on WarehousePG#

ストレージの場所を使用したオブジェクトストレージの照会#

例#

Icebergカタログに接続する#

例#

Sparkでの加速#

例#

CTASを使用したオブジェクトストレージへの書き込み#

例#