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 権限を付与してください。

接続情報の収集

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

簡単に言うと、共有キー認証を実践する場合、次の情報を収集する必要があります:

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

利用可能なすべての認証方法については、Authenticate to Azure cloud storage を参照してください。

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 形式には列名が含まれているためです。

データベースを作成し、それに切り替えます:

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 列に設定されています。このデータに対するユースケースは異なるかもしれないので、ソートキーとして 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;

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