Data Lake Formation:Accéder à DLF avec PyPaimon

Intégration du catalogue DLF

En important le package d'extension pypaimon_dlf2 et en configurant un catalogue DLF, vous pouvez synchroniser automatiquement les métadonnées des tables Paimon vers Alibaba Cloud Data Lake Formation (DLF).

Cette intégration offre les avantages clés suivants :

• Interopérabilité multi-moteurs : Une fois les métadonnées hébergées dans DLF, d'autres moteurs de calcul Alibaba Cloud, tels que MaxCompute, Hologres et Alibaba Cloud EMR, peuvent accéder sans friction à ces données Paimon.

• Gouvernance unifiée : Les fonctionnalités de gestion de data lake de DLF vous permettent de gérer le cycle de vie des tables Paimon et d'optimiser automatiquement leurs formats de stockage.

python3 testdlf.py

root@iZxxx:/opt# python3 testdlf.py
pyarrow.Table
date: string not null
hour: string not null
key: int64 not null
value: string
----
date: [["2024-12-01"],["2024-12-01"]]
hour: [["09"],["08"]]
key: [[2],[1]]
value: [["BBB"],["AAA"]]

Créer une table Paimon DLF

1. Créez un catalogue Paimon DLF.

   Remarque

   Vous devez créer un catalogue pour accéder aux tables Paimon dans DLF.

   # The catalog_options is a dictionary where both keys and values are strings.
   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)

   Le tableau suivant décrit les paramètres.

   Paramètre	Description
   metastore	Définissez la valeur fixe rest, qui indique que vous vous connectez à DLF via le protocole REST Catalog.
   dlf.region	L'ID de la région DLF. Pour plus d'informations, consultez la page Points de terminaison.
   uri	Le point de terminaison REST Catalog DLF. Dans un environnement VPC, utilisez http://${region_id}-vpc.dlf.aliyuncs.com. Dans un environnement public, utilisez https://dlfnext.${region_id}.aliyuncs.com. Pour plus d'informations, consultez la page Points de terminaison.
   warehouse	Le nom du catalogue de données DLF. Vous pouvez consulter ce nom dans la console Data Lake Formation. Pour plus d'informations, consultez la page Catalogue de données.
   dlf.access-key-id	L'AccessKey ID requis pour accéder au service DLF. Pour plus d'informations, consultez la page Créer une paire de clés AccessKey.
   dlf.access-key-secret	L'AccessKey Secret requis pour accéder au service DLF. Pour plus d'informations, consultez la page Créer une paire de clés AccessKey.
   token.provider	Définissez la valeur fixe dlf, qui indique que le service DLF fournit le jeton d'accès.
   max-workers	(Facultatif) Le nombre de threads simultanés pour la lecture des données dans PyPaimon. Doit être un entier supérieur ou égal à 1. La valeur par défaut est 1, ce qui indique une lecture sérielle.

2. Créez une base de données.

   Dans un catalogue Paimon, chaque table appartient à une base de données spécifique. Créez des bases de données pour organiser et gérer vos tables.

   catalog.create_database(
      name='database_name',
      ignore_if_exists=True,    # Specifies whether to ignore the error if the database already exists.
      properties={'key': 'value'}  # Optional. The database properties.
   )

3. Créez un schéma.

   Un schéma comprend les définitions des colonnes, les clés de partition, les clés primaires, les options de table et les commentaires. Les définitions des colonnes sont décrites à l'aide de pyarrow.Schema. Les autres paramètres sont facultatifs. Vous pouvez construire un pyarrow.Schema de l'une des deux manières suivantes.

   PyArrow

   Utilisez la méthode pyarrow.schema. L'exemple de code suivant illustre cette approche.

   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'
   )

   Remarque

   Pour obtenir des informations sur le mappage des types de données entre pyarrow et Paimon, consultez la page Mappage des types de données PyPaimon.

   Pandas

   Si vous disposez de données Pandas, vous pouvez dériver le schéma directement à partir d'un pandas.DataFrame. L'exemple de code suivant illustre cette approche.

   import pandas as pd
   import pyarrow as pa
   from pypaimon import Schema
   # This is sample DataFrame data.
   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)
   # Obtain the pyarrow.Schema from the DataFrame.
   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. Créez et récupérez une table.

   catalog.create_table(
       identifier='database_name.table_name',
       schema=schema,
       ignore_if_exists=True  # Specifies whether to ignore the error if the table already exists.
   )
   table = catalog.get_table('database_name.table_name')
           

Écrire des données dans une table

1. Créez les opérations d'écriture et de validation (commit) de la table.

   # Create table write and commit operations.
   write_builder = table.new_batch_write_builder()
   table_write = write_builder.new_write()
   table_commit = write_builder.new_commit()
   # Write sample Pandas data.
   data = {
       'date': ['2024-12-01', '2024-12-01', '2024-12-02'],
       'hour': ['08', '09', '08'],
       'key': [1, 2, 3],
       'value': ['AAA', 'BBB', 'CCC'],
   }

2. Vous pouvez écrire des données dans la table de l'une des manières suivantes :

   Pour les grands ensembles de données, utilisez PyArrow. Pour les ensembles de données plus petits (généralement quelques gigaoctets ou moins), Pandas peut s'avérer plus efficace.

   PyArrow

   Vous pouvez écrire les données sous forme de pyarrow.Table ou de pyarrow.RecordBatch. pyarrow.RecordBatch convient mieux au traitement en flux continu.

   • Méthode 1 : Écrire un pyarrow.Table

     # Create fields.
     fields = [
         pa.field('date', pa.string()),
         pa.field('hour', pa.string()),
         pa.field('key', pa.int64()),
         pa.field('value', pa.string())
     ]
     # Create a schema from the fields.
     schema = pa.schema(fields)
     # Create a table.
     pa_table = pa.Table.from_arrays(data, schema)
     # Write the data.
     table_write.write_arrow(pa_table)

   • Méthode 2 : Écrire un pyarrow.RecordBatch

     # Create fields.
     fields = [
         pa.field('date', pa.string()),
         pa.field('hour', pa.string()),
         pa.field('key', pa.int64()),
         pa.field('value', pa.string())
     ]
     # Create a schema from the fields.
     schema = pa.schema(fields)
     # Create a RecordBatch.
     record_batch = pa.RecordBatch.from_arrays(data, schema)
     # Write the data.
     table_write.write_arrow_batch(record_batch)

   Pandas

   Vous pouvez écrire des données à partir d'un pandas.DataFrame.

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

3. Validez les données et libérez les ressources.

   # Commit the data.
   table_commit.commit(table_write.prepare_commit())
   # Close the resources.
   table_write.close()
   table_commit.close()

Lire des données depuis une table

1. Créez un ReadBuilder.

   read_builder = table.new_read_builder()

2. Utilisez un PredicateBuilder pour construire et appliquer des conditions de filtre (pushdown).

   • Par exemple, vous pouvez lire uniquement les données où date est égal à 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)

   • Par exemple, vous pouvez projeter uniquement les colonnes key et value.

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

   Remarque

   Pour plus d'informations sur les conditions de filtre prises en charge, consultez la page Conditions de filtre PyPaimon.

3. Obtenez les splits.

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

4. Convertissez les splits en différents formats de sortie.

   Apache Arrow

   • Lisez toutes les données dans un pyarrow.Table.

     table_read = read_builder.new_read()
     pa_table = table_read.to_arrow(splits)
     print(pa_table)
     # Sample output:
     # pyarrow.Table
     # key: int64 not null
     # value: string
     # ----
     # key: [[2],[1]]
     # value: [["BBB"],["AAA"]]

   • Lisez les données dans un pyarrow.RecordBatchReader et itérez sur les lots.

     table_read = read_builder.new_read()
     for batch in table_read.to_arrow_batch_reader(splits):
         print(batch)
     # Sample output:
     # pyarrow.RecordBatch
     # key: int64
     # value: string
     # ----
     # key: [1,2]
     # value: ["AAA","BBB"]

   Pandas

   Lisez les données dans un pandas.DataFrame.

   table_read = read_builder.new_read()
   df = table_read.to_pandas(splits)
   print(df)
   # Sample output:
   #    key  value
   # 0   1  AAA
   # 1   2  BBB

   DuckDB

   Important

   Vous devez installer DuckDB. Vous pouvez exécuter pip install duckdb pour l'installer.

   Convertissez les données en une table DuckDB en mémoire et interrogez-la.

   table_read = read_builder.new_read()
   duckdb_con = table_read.to_duckdb(splits, 'duckdb_table')
   print(duckdb_con.query("SELECT * FROM duckdb_table").fetchdf())
   # Sample output:
   #    key  value
   # 0   1  AAA
   # 1   2  BBB
   print(duckdb_con.query("SELECT * FROM duckdb_table WHERE key = 1").fetchdf())
   # Sample output:
   #    key  value
   # 0   1  AAA

   Ray

   Important

   Vous devez installer Ray. Vous pouvez exécuter pip install ray pour l'installer.

   table_read = read_builder.new_read()
   ray_dataset = table_read.to_ray(splits)
   # Print information about ray_dataset.
   print(ray_dataset)
   # Sample output:
   # MaterializedDataset(num_blocks=1, num_rows=2, schema={key: int64, value: string})
   # Print the first two records in ray_dataset.
   print(ray_dataset.take(2))
   # Sample output:
   # [{'key': 1, 'value': 'AAA'}, {'key': 2, 'value': 'BBB'}]
   # Convert the entire ray_dataset to a Pandas DataFrame and print the result.
   print(ray_dataset.to_pandas())
   # Sample output:
   #    key  value
   # 0   1  AAA
   # 1   2  BBB