スキーマレスクエリによるデータレイク上の Parquet データの読み取り - MaxCompute

このドキュメントでは、E-MapReduce のサーバーレス Spark クラスターを例に、MaxCompute でスキーマレスクエリを使用して Spark SQL で生成された Parquet ファイルを読み取り、計算完了後に UNLOAD コマンドで結果を OSS に書き戻す方法を説明します。

前提条件

EMR Serverless Spark ワークスペースを作成済みであること。

本文では、中国東部 1 (杭州) リージョンを例に、作成するワークスペース名を schemaless_test、OSS バケット名を oss-mc-test と仮定します。
MaxCompute プロジェクトを作成済みであること。

手順1：サーバーレス Spark による Parquet データの生成

E-MapReduce コンソールにログインします。左側のナビゲーションペインで、[EMR Serverless] > [Spark] を選択します。
[Spark] ページで、ワークスペース名をクリックして [EMR Serverless Spark] ページに移動し、左側の データ開発 をクリックします。
Spark SQL タスクを新規作成し、次の SQL コマンドを入力して Parquet フォーマットのテーブルを作成してデータを書き込み、[実行] をクリックします。

重要
下記のコマンドを実行する前に、ページ右上で選択されているデータディレクトリとデータベースに関連付けられたパスが、対象の OSS バケットのパスであることを確認してください。
```
CREATE TABLE example_table01 (
    id INT,
    name STRING,
    age INT
) USING PARQUET;
INSERT INTO example_table01 VALUES
(1, 'Alice', 30),
(2, 'Bob', 25),
(3, 'Charlie', 35),
(4, 'David', 40),
(5, 'Eve', 32),
(6, 'Frank', 28),
(7, 'Grace', 33),
(8, 'Hannah', 29),
(9, 'Ian', 36),
(10, 'Julia', 31);
SELECT * FROM example_table01;
```
実行が成功すると、DLF データレイク構築コンソールの[メタデータ管理] ページで生成されたexample_table01 テーブルを確認できます。[メタデータ管理] ページで [データテーブル] タブを選択すると、作成された example_table01 テーブルが表示されます。テーブルの形式は Parquet です。

また、OSS バケットのディレクトリで生成された Parquet ファイルを確認することもできます。OSS の example_table01/ ディレクトリには、_SUCCESS マーカーファイルと 2 つの .snappy.parquet データファイルが表示され、Parquet データが正常に書き込まれたことがわかります。

手順2：スキーマレスクエリによるデータの読み取り

MaxCompute で Parquet ファイルを読み取る方法については、「Feature: Schemaless Query」をご参照ください。

Parquet テーブルに書き込む際に_SUCCESSという名前のファイルが生成されるため、file_pattern_blacklist パラメーターを使用して_SUCCESS ファイルをブラックリストに追加し、読み取り対象から除外する必要があります。このパラメーターを追加しない場合、エラーが報告されます。

SELECT * FROM 
LOCATION 'oss://oss-cn-hangzhou-internal.aliyuncs.com/oss-mc-test/example_table01/' 
(
  'file_format'='parquet',
  'file_pattern_blacklist'='.*_SUCCESS.*'
);

返される結果は次のとおりです。

+------------+------------+------------+
| id         | name       | age        |
+------------+------------+------------+
| 1          | Alice      | 30         |
| 2          | Bob        | 25         |
| 3          | Charlie    | 35         |
| 4          | David      | 40         |
| 5          | Eve        | 32         |
| 6          | Frank      | 28         |
| 7          | Grace      | 33         |
| 8          | Hannah     | 29         |
| 9          | Ian        | 36         |
| 10         | Julia      | 31         |
+------------+------------+------------+

手順3：SQL による計算の実行

年齢が 30 歳を超える人数を問い合わせます。

SELECT COUNT(*) FROM 
LOCATION 'oss://oss-cn-hangzhou-internal.aliyuncs.com/oss-mc-test/example_table01/' 
(
  'file_format'='parquet',
  'file_pattern_blacklist'='.*_SUCCESS.*'
)
WHERE age>30;

返される結果は次のとおりです。

+------------+
| _c0        |
+------------+
| 6          |
+------------+

手順4：UNLOAD コマンドによる計算結果の OSS への書き戻し

MaxCompute では、MaxCompute プロジェクトから外部ストレージの OSS にデータをエクスポートして、他の計算エンジンで利用できます。詳細については、「UNLOAD」をご参照ください。

MaxCompute で次のコマンドを実行し、「手順3：SQL による計算の実行」の計算結果を Parquet フォーマットで OSS にエクスポートします。

以下のコードを実行する前に、OSS バケット oss-mc-test に unload_schemaless ディレクトリを作成する必要があります。

UNLOAD FROM  
(
  SELECT COUNT(*) FROM 
  LOCATION 'oss://oss-cn-hangzhou-internal.aliyuncs.com/oss-mc-test/example_table01/' 
  ('file_format'='parquet','file_pattern_blacklist'='.*_SUCCESS.*')
  WHERE age>30
)
INTO
LOCATION 'oss://oss-cn-hangzhou-internal.aliyuncs.com/oss-mc-test/unload_schemaless/'
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' 
WITH SERDEPROPERTIES ('odps.properties.rolearn'='acs:ram::<uid>:role/AliyunODPSDefaultRole') 
STORED AS PARQUET 
PROPERTIES('mcfed.parquet.compression'='SNAPPY')
;

OSS コンソールにログインし、UNLOAD が成功したことを確認します。