Azure クラウドストレージからのバッチデータロード

CelerData は、Microsoft Azure Storage からデータをロードするための2つのオプションを提供しています。

• INSERT+FILES() を使用した同期ロード
• Broker Load を使用した非同期ロード

これらのオプションにはそれぞれの利点があり、以下のセクションで詳しく説明します。

ほとんどの場合、使用が簡単な INSERT+FILES() メソッドをお勧めします。

ただし、INSERT+FILES() メソッドは現在、Parquet と ORC ファイル形式のみをサポートしています。そのため、CSV などの他のファイル形式のデータをロードする必要がある場合や、データロード中に DELETE などのデータ変更を行う必要がある場合は、Broker Load を使用できます。

始める前に

ソースデータの準備

CelerData にロードしたいソースデータが、Azure ストレージアカウント内のコンテナに適切に保存されていることを確認してください。

このトピックでは、Azure Data Lake Storage Gen2 (ADLS Gen2) ストレージアカウント (celerdata) 内のコンテナ (celerdata-container) のルートディレクトリに保存されている Parquet 形式のサンプルデータセット (user_behavior_ten_million_rows.parquet) のデータをロードすることを想定しています。

権限の確認

CelerData クラスター内のテーブルにデータをロードするには、これらのテーブルに対して INSERT 権限を持つユーザーとしてのみ可能です。INSERT 権限を持っていない場合は、GRANT に従って、CelerData クラスターに接続するために使用するユーザーに INSERT 権限を付与してください。

接続情報の収集

このトピックの例では、Shared Key 認証方法を使用しています。ADLS Gen2 からデータを読み取る権限を持っていることを確認するために、Azure Data Lake Storage Gen2 > Shared Key (ストレージアカウントのアクセスキー) を読んで、設定する必要がある認証パラメータを理解することをお勧めします。

要するに、Shared Key 認証を使用する場合、次の情報を収集する必要があります。

• ADLS Gen2 ストレージアカウントの名前
• ADLS Gen2 ストレージアカウントの共有キー (アクセスキー)

利用可能なすべての認証方法については、Azure クラウドストレージへの認証 を参照してください。

INSERT+FILES() の使用

このメソッドは、Parquet と ORC ファイル形式のみをサポートしています。

INSERT+FILES() の利点

FILES() は、指定したパス関連のプロパティに基づいてクラウドストレージに保存されているファイルを読み取り、ファイル内のデータのテーブルスキーマを推測し、ファイルからデータをデータ行として返すことができます。

FILES() を使用すると、次のことが可能です。

• SELECT を使用して Azure から直接データをクエリする。
• CREATE TABLE AS SELECT (CTAS) を使用してテーブルを作成し、ロードする。
• INSERT を使用して既存のテーブルにデータをロードする。

典型的な例

SELECT を使用して Azure から直接クエリする

SELECT+FILES() を使用して Azure から直接クエリすることで、テーブルを作成する前にデータセットの内容をプレビューすることができます。例えば：

• データを保存せずにデータセットをプレビューする。
• 最小値と最大値をクエリして、使用するデータ型を決定する。
• NULL 値を確認する。

次の例は、ストレージアカウント celerdata 内のコンテナ celerdata-container に保存されているサンプルデータセット user_behavior_ten_million_rows.parquet をクエリします。

SELECT * FROM FILES
(
    "path" = "abfss://celerdata-container@celerdata.dfs.core.windows.net/user_behavior_ten_million_rows.parquet",
    "format" = "parquet",
    "azure.adls2.storage_account" = "celerdata",
    "azure.adls2.shared_key" = "xxxxxxxxxxxxxxxxxx"
)
LIMIT 3;

システムは次のようなクエリ結果を返します。

NOTE

上記のように返される列名は、Parquet ファイルによって提供されます。

CTAS を使用してテーブルを作成しロードする

これは前の例の続きです。前のクエリは CREATE TABLE AS SELECT (CTAS) にラップされており、スキーマ推測を使用してテーブル作成を自動化します。これは、CelerData がテーブルスキーマを推測し、作成したいテーブルを作成し、そのテーブルにデータをロードすることを意味します。Parquet ファイルを使用する場合、Parquet 形式には列名が含まれているため、FILES() テーブル関数を使用する際にテーブルを作成するために列名や型を指定する必要はありません。

NOTE

スキーマ推測を使用する場合の CREATE TABLE の構文では、レプリカの数を設定することはできません。CelerData クラシッククラスターを使用している場合は、テーブルを作成する前にレプリカの数を設定してください。以下の例は、単一レプリカのシステム用です：

ADMIN SET FRONTEND CONFIG ('default_replication_num' = "1");

データベースを作成し、切り替えます：

CREATE DATABASE IF NOT EXISTS mydatabase;
USE mydatabase;

CTAS を使用してテーブルを作成し、ストレージアカウント celerdata 内のコンテナ celerdata-container に保存されているサンプルデータセット user_behavior_ten_million_rows.parquet のデータをテーブルにロードします：

CREATE TABLE user_behavior_inferred AS
SELECT * FROM FILES
(
    "path" = "abfss://celerdata-container@celerdata.dfs.core.windows.net/user_behavior_ten_million_rows.parquet",
    "format" = "parquet",
    "azure.adls2.storage_account" = "celerdata",
    "azure.adls2.shared_key" = "xxxxxxxxxxxxxxxxxx"
);

テーブルを作成した後、DESCRIBE を使用してそのスキーマを表示できます：

DESCRIBE user_behavior_inferred;

システムは次のようなクエリ結果を返します。

推測されたスキーマと手動で作成されたスキーマを比較します：

• データ型
• NULL 許可
• キーフィールド

宛先テーブルのスキーマをよりよく制御し、クエリパフォーマンスを向上させるために、本番環境では手動でテーブルスキーマを指定することをお勧めします。

テーブルをクエリして、データがロードされたことを確認します。例：

SELECT * from user_behavior_inferred LIMIT 3;

システムは次のようなクエリ結果を返し、データが正常にロードされたことを示します。

INSERT を使用して既存のテーブルにロードする

挿入するテーブルをカスタマイズしたい場合があります。例えば、以下のような点です：

• 列のデータ型、NULL 許可設定、またはデフォルト値
• キーの種類と列
• データのパーティショニングとバケッティング

NOTE

最も効率的なテーブル構造を作成するには、データの使用方法と列の内容に関する知識が必要です。このトピックではテーブル設計については扱いません。テーブル設計に関する情報は、Table types を参照してください。

この例では、テーブルがどのようにクエリされるか、Parquet ファイル内のデータに関する知識に基づいてテーブルを作成しています。Parquet ファイル内のデータに関する知識は、Azure でファイルを直接クエリすることで得ることができます。

• Azure でのデータセットのクエリにより、Timestamp 列が datetime データ型に一致するデータを含んでいることが示されているため、以下の DDL で列型が指定されています。
• Azure でデータをクエリすることで、データセットに NULL 値がないことがわかるため、DDL では列を NULL 許可として設定していません。
• 予想されるクエリの種類に基づいて、ソートキーとバケッティング列は UserID 列に設定されています。このデータに対するユースケースは異なるかもしれないので、ItemID をソートキーとして使用することを選択するかもしれません。

データベースを作成し、切り替えます：

CREATE DATABASE IF NOT EXISTS mydatabase;
USE mydatabase;

手動でテーブルを作成します（Azure からロードしたい Parquet ファイルと同じスキーマを持つことをお勧めします）：

CREATE TABLE user_behavior_declared
(
    UserID int(11),
    ItemID int(11),
    CategoryID int(11),
    BehaviorType varchar(65533),
    Timestamp datetime
);

テーブルを作成した後、INSERT INTO SELECT FROM FILES() を使用してロードできます：

INSERT INTO user_behavior_declared
SELECT * FROM FILES
(
    "path" = "abfss://celerdata-container@celerdata.dfs.core.windows.net/user_behavior_ten_million_rows.parquet",
    "format" = "parquet",
    "azure.adls2.storage_account" = "celerdata",
    "azure.adls2.shared_key" = "xxxxxxxxxxxxxxxxxx"
)

ロードが完了したら、テーブルをクエリしてデータがロードされたことを確認します。例：

SELECT * from user_behavior_declared LIMIT 3;

システムは次のようなクエリ結果を返し、データが正常にロードされたことを示します。

ロード進捗の確認

information_schema.loads ビューから INSERT ジョブの進捗をクエリできます。この機能は v3.1 以降でサポートされています。例：

SELECT * FROM information_schema.loads ORDER BY JOB_ID DESC;

複数のロードジョブを送信した場合は、ジョブに関連付けられた LABEL でフィルタリングできます。例：

SELECT * FROM information_schema.loads WHERE LABEL = 'insert_659411dd-a560-11ee-af93-000d3a544fa3';

loads ビューで提供されるフィールドに関する情報は、Information Schema を参照してください。

NOTE

INSERT は同期コマンドです。INSERT ジョブがまだ実行中の場合、その実行ステータスを確認するには別のセッションを開く必要があります。

Broker Load の使用

非同期の Broker Load プロセスは、Azure への接続を確立し、データを取得し、CelerData にデータを保存する役割を果たします。

このメソッドは、Parquet、ORC、CSV、JSON ファイル形式をサポートしています。

Broker Load の利点

• Broker Load は、ロード中に UPSERT や DELETE 操作などのデータ変換やデータ変更をサポートします。
• Broker Load はバックグラウンドで実行され、クライアントはジョブが続行するために接続を維持する必要がありません。
• Broker Load は長時間実行されるジョブに適しており、デフォルトのタイムアウトは4時間です。
• Parquet と ORC ファイル形式に加えて、Broker Load は CSV と JSON ファイル形式をサポートしています。

典型的な例

データベースとテーブルを作成し、Azure からサンプルデータセット user_behavior_ten_million_rows.parquet を取得するロードプロセスを開始し、データロードの進捗と成功を確認します。

データベースとテーブルの作成

CelerData クラスターに接続します。次に、データベースを作成し、切り替えます：

CREATE DATABASE IF NOT EXISTS mydatabase;
USE mydatabase;

手動でテーブルを作成します（Azure からロードしたい Parquet ファイルと同じスキーマを持つことをお勧めします）：

CREATE TABLE user_behavior
(
    UserID int(11),
    ItemID int(11),
    CategoryID int(11),
    BehaviorType varchar(65533),
    Timestamp datetime
);

Broker Load の開始

次のコマンドを実行して、サンプルデータセット user_behavior_ten_million_rows.parquet から user_behavior テーブルにデータをロードする Broker Load ジョブを開始します：

LOAD LABEL user_behavior
(
    DATA INFILE("abfss://celerdata-container@celerdata.dfs.core.windows.net/user_behavior_ten_million_rows.parquet")
    INTO TABLE user_behavior
    FORMAT AS "parquet"
)
WITH BROKER
(
    "azure.adls2.storage_account" = "celerdata",
    "azure.adls2.shared_key" = "xxxxxxxxxxxxxxxxxx"
)
PROPERTIES
(
    "timeout" = "3600"
);

このジョブには4つの主要なセクションがあります：

• LABEL: ロードジョブの状態をクエリする際に使用される文字列。
• LOAD 宣言: ソース URI、ソースデータ形式、および宛先テーブル名。
• BROKER: ソースの接続詳細。
• PROPERTIES: タイムアウト値およびロードジョブに適用するその他のプロパティ。

詳細な構文とパラメータの説明については、BROKER LOAD を参照してください。

ロード進捗の確認

information_schema.loads ビューをクエリして、ロードの進捗を追跡します：

SELECT * FROM information_schema.loads;

複数のロードジョブを送信した場合は、ジョブに関連付けられた LABEL でフィルタリングできます：

SELECT * FROM information_schema.loads WHERE LABEL = 'user_behavior';

loads ビューで提供されるフィールドに関する情報は、Information Schema を参照してください。

ロードジョブが完了したことを確認した後、宛先テーブルの一部をチェックして、データが正常にロードされたかどうかを確認できます。例：

SELECT * from user_behavior LIMIT 3;

システムは次のようなクエリ結果を返し、データが正常にロードされたことを示します。