Data Lake Formation:PyPaimon を使用した DLF へのアクセス

DLF カタログの統合

pypaimon_dlf2 拡張パッケージをインポートして DLF カタログを設定することで、Paimon テーブルのメタデータを Alibaba Cloud Data Lake Formation (DLF) に自動的に同期できます。

この統合により、主に次の利点が得られます。

• 複数エンジン間の相互運用性：メタデータを DLF でホストすることで、MaxCompute、Hologres、Alibaba Cloud EMR などの他の Alibaba Cloud コンピューティングエンジンがこの Paimon データにシームレスにアクセスできます。

• 統合ガバナンス：DLF のデータレイク管理機能により、Paimon テーブルのライフサイクルを管理し、ストレージ形式を自動的に最適化できます。

DLF Paimon テーブルの作成

1. Paimon DLF カタログの作成

   説明

   DLF の Paimon テーブルにアクセスするには、カタログを作成する必要があります。

   # catalog_options は、キーと値の両方が文字列である辞書です。
   catalog_options = {
     'metastore': 'rest',
     'uri': "http://${region_id}-vpc.dlf.aliyuncs.com",
     'warehouse': "${catalog_name}",
     'dlf.region': '${region_id}',
     "token.provider": "dlf",
     'dlf.access-key-id': "xxx",
     'dlf.access-key-secret': "xxxx",
   }
   catalog = CatalogFactory.create(catalog_options)

   次の表に、パラメータの説明を示します。

   パラメータ	説明
   metastore	固定値 rest に設定します。これは、REST Catalog プロトコルを使用して DLF に接続することを示します。
   dlf.region	DLF リージョンの ID。詳細については、「エンドポイント」をご参照ください。
   uri	DLF REST Catalog エンドポイント。VPC 環境では http://${region_id}-vpc.dlf.aliyuncs.com を使用します。パブリックネットワーク環境では https://dlfnext.${region_id}.aliyuncs.com を使用します。詳細については、「エンドポイント」をご参照ください。
   warehouse	DLF データカタログの名前。Data Lake Formation コンソールで名前を確認できます。詳細については、「データカタログ」をご参照ください。
   dlf.access-key-id	DLF サービスにアクセスするために必要な AccessKey ID。詳細については、「AccessKey ペアの作成」をご参照ください。
   dlf.access-key-secret	DLF サービスにアクセスするために必要な AccessKey シークレット。詳細については、「AccessKey ペアの作成」をご参照ください。
   token.provider	固定値 dlf に設定します。これは、DLF サービスがアクセストークンを提供することを示します。
   max-workers	オプション。PyPaimon でデータを読み取る際の同時スレッド数。1 以上の整数である必要があります。デフォルトは 1 で、シリアル読み取りを示します。

2. データベースの作成

   Paimon カタログでは、すべてのテーブルは特定のデータベースに属します。テーブルを整理および管理するためにデータベースを作成します。

   catalog.create_database(
      name='database_name',
      ignore_if_exists=True,    # データベースが既に存在する場合にエラーを無視するかどうかを指定します。
      properties={'key': 'value'}  # オプション。データベースのプロパティ。
   )

3. スキーマの作成

   スキーマには、列定義、パーティションキー、プライマリキー、テーブルオプション、コメントが含まれます。列定義は pyarrow.Schema を使用して記述します。他のパラメータはオプションです。pyarrow.Schema は、次のいずれかの方法で構築できます。

   PyArrow

   pyarrow.schema メソッドを使用します。次にコード例を示します。

   import pyarrow as pa
   from pypaimon import Schema
   pa_schema = pa.schema([
       ('date', pa.string()),
       ('hour', pa.string()),
       ('key', pa.int64()),
       ('value', pa.string())
   ])
   schema = Schema(
       pa_schema=pa_schema,
       partition_keys=['date', 'hour'],
       primary_keys=['date', 'hour', 'key'],
       options={'bucket': '2'},
       comment='my test table'
   )

   説明

   pyarrow と Paimon 間のデータ型マッピングの詳細については、「PyPaimon データ型マッピング」をご参照ください。

   Pandas

   Pandas データがある場合、pandas.DataFrame から直接スキーマを導出できます。次にコード例を示します。

   import pandas as pd
   import pyarrow as pa
   from pypaimon import Schema
   # これはサンプルの DataFrame データです。
   data = {
       'date': ['2024-12-01', '2024-12-01', '2024-12-02'],
       'hour': ['08', '09', '08'],
       'key': [1, 2, 3],
       'value': ['AAA', 'BBB', 'CCC'],
   }
   dataframe = pd.DataFrame(data)
   # DataFrame から pyarrow.Schema を取得します。
   record_batch = pa.RecordBatch.from_pandas(dataframe)
   pa_schema = record_batch.schema
   schema = Schema(
       pa_schema=pa_schema,
       partition_keys=['date', 'hour'],
       primary_keys=['date', 'hour', 'key'],
       options={'bucket': '2'},
       comment='my test table'
   )

4. テーブルの作成と取得

   catalog.create_table(
       identifier='database_name.table_name',
       schema=schema,
       ignore_if_exists=True  # テーブルが既に存在する場合にエラーを無視するかどうかを指定します。
   )
   table = catalog.get_table('database_name.table_name')
           

テーブルへのデータの書き込み

1. テーブル書き込みとコミット操作を作成します。

   # テーブル書き込みとコミット操作を作成します。
   write_builder = table.new_batch_write_builder()
   table_write = write_builder.new_write()
   table_commit = write_builder.new_commit()
   # Pandas のサンプルデータを書き込みます。
   data = {
       'date': ['2024-12-01', '2024-12-01', '2024-12-02'],
       'hour': ['08', '09', '08'],
       'key': [1, 2, 3],
       'value': ['AAA', 'BBB', 'CCC'],
   }

2. 次のいずれかの方法でテーブルにデータを書き込むことができます。

   大規模なデータセットには PyArrow を使用します。通常数ギガバイト以下の小規模なデータセットには、Pandas の方が効率的です。

   PyArrow

   データは pyarrow.Table または pyarrow.RecordBatch として書き込めます。pyarrow.RecordBatch はストリーム処理により適しています。

   • 方法 1：pyarrow.Table を書き込む

     # フィールドを作成します。
     fields = [
         pa.field('date', pa.string()),
         pa.field('hour', pa.string()),
         pa.field('key', pa.int64()),
         pa.field('value', pa.string())
     ]
     # フィールドからスキーマを作成します。
     schema = pa.schema(fields)
     # テーブルを作成します。
     pa_table = pa.Table.from_arrays(data, schema)
     # データを書き込みます。
     table_write.write_arrow(pa_table)

   • 方法 2：pyarrow.RecordBatch を書き込む

     # フィールドを作成します。
     fields = [
         pa.field('date', pa.string()),
         pa.field('hour', pa.string()),
         pa.field('key', pa.int64()),
         pa.field('value', pa.string())
     ]
     # フィールドからスキーマを作成します。
     schema = pa.schema(fields)
     # RecordBatch を作成します。
     record_batch = pa.RecordBatch.from_arrays(data, schema)
     # データを書き込みます。
     table_write.write_arrow_batch(record_batch)

   Pandas

   pandas.DataFrame からデータを書き込むことができます。

   import pandas as pd
   dataframe = pd.DataFrame(data)
   table_write.write_pandas(dataframe)

3. データをコミットし、リソースを閉じます。

   # データをコミットします。
   table_commit.commit(table_write.prepare_commit())
   # リソースを閉じます。
   table_write.close()
   table_commit.close()

テーブルからのデータの読み取り

1. ReadBuilder を作成します。

   read_builder = table.new_read_builder()

2. PredicateBuilder を使用して、フィルター条件を構築しプッシュダウンします。

   • たとえば、date が 2024-12-01 であるデータのみを読み取ることができます。

     predicate_builder = read_builder.new_predicate_builder()
     predicate = predicate_builder.equal('date', '2024-12-01')
     read_builder = read_builder.with_filter(predicate)

   • たとえば、key と value 列のみを射影できます。

     read_builder = read_builder.with_projection(['key', 'value'])

   説明

   サポートされているフィルター条件の詳細については、「PyPaimon フィルター条件」をご参照ください。

3. splits を取得します。

   table_scan = read_builder.new_scan()
   splits = table_scan.plan().splits()

4. splits をさまざまな出力形式に変換します。

   Apache Arrow

   • すべてのデータを pyarrow.Table に読み込みます。

     table_read = read_builder.new_read()
     pa_table = table_read.to_arrow(splits)
     print(pa_table)
     # サンプル出力：
     # pyarrow.Table
     # key: int64 not null
     # value: string
     # ----
     # key: [[1],[2]]
     # value: [["AAA"],["BBB"]]

   • データを pyarrow.RecordBatchReader に読み込み、バッチを反復処理します。

     table_read = read_builder.new_read()
     for batch in table_read.to_arrow_batch_reader(splits):
         print(batch)
     # サンプル出力：
     # pyarrow.RecordBatch
     # key: int64
     # value: string
     # ----
     # key: [1,2]
     # value: ["AAA","BBB"]

   Pandas

   データを pandas.DataFrame に読み込みます。

   table_read = read_builder.new_read()
   df = table_read.to_pandas(splits)
   print(df)
   # サンプル出力：
   #    key  value
   # 0   1  AAA
   # 1   2  BBB

   DuckDB

   重要

   DuckDB をインストールする必要があります。pip install duckdb を実行してインストールできます。

   データをインメモリ DuckDB テーブルに変換し、クエリを実行します。

   table_read = read_builder.new_read()
   duckdb_con = table_read.to_duckdb(splits, 'duckdb_table')
   print(duckdb_con.query("SELECT * FROM duckdb_table").fetchdf())
   # サンプル出力：
   #    key  value
   # 0   1  AAA
   # 1   2  BBB
   print(duckdb_con.query("SELECT * FROM duckdb_table WHERE key = 1").fetchdf())
   # サンプル出力：
   #    key  value
   # 0   1  AAA

   Ray

   重要

   Ray をインストールする必要があります。pip install ray を実行してインストールできます。

   table_read = read_builder.new_read()
   ray_dataset = table_read.to_ray(splits)
   # ray_dataset に関する情報を出力します。
   print(ray_dataset)
   # サンプル出力：
   # MaterializedDataset(num_blocks=1, num_rows=2, schema={key: int64, value: string})
   # ray_dataset の最初の 2 つのレコードを出力します。
   print(ray_dataset.take(2))
   # サンプル出力：
   # [{'key': 1, 'value': 'AAA'}, {'key': 2, 'value': 'BBB'}]
   # ray_dataset 全体を Pandas DataFrame に変換し、結果を出力します。
   print(ray_dataset.to_pandas())
   # サンプル出力：
   #    key  value
   # 0   1  AAA
   # 1   2  BBB