MaxCompute:Utiliser PyODPS dans un environnement local

Prérequis

Avant de commencer, vérifiez que vous disposez des éléments suivants :

• PyODPS installé localement et variables d'environnement configurées

• La table d'exemple pyodps_iris créée dans votre projet. Consultez la section Exemples d'utilisation pour savoir comment télécharger le jeu de données, créer la table et y écrire des données.

Prise en main

1. Ouvrez un éditeur Python et créez un fichier .py.

   Si aucun éditeur Python n'est installé, créez un fichier avec l'extension .py à l'aide de n'importe quel éditeur de texte.

2. Écrivez votre code PyODPS. Les sections suivantes présentent les principales fonctionnalités à l'aide d'exemples exécutables.

3. Enregistrez le fichier et exécutez-le avec la commande python <your-file>.py.

Initialiser le point d'entrée ODPS

La classe ODPS est le point d'entrée de toutes les opérations PyODPS. Initialisez-la avec vos identifiants et les détails de votre projet avant d'appeler toute autre API.

import os
from odps import ODPS

o = ODPS(
    # Read credentials from environment variables
    access_id=os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    secret_access_key=os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),

    # (Not recommended) To specify credentials explicitly instead:
    # access_id='your-aliyun-access-key-id',
    # secret_access_key='your-aliyun-access-key-secret',

    project='<your-default-project>',
    endpoint='<your-endpoint>',
)

Remplacez les espaces réservés suivants :

Après initialisation, l'objet o prend en charge les opérations list, get, exist, create et delete sur les ressources MaxCompute. Pour une référence complète, consultez la section Présentation des opérations de base.

Exécuter du code SQL

MaxCompute prend en charge deux modes d'exécution SQL : le mode traditionnel et MaxCompute Query Acceleration (MCQA). Les deux modes acceptent les instructions DDL (Data Definition Language) et DML (Data Manipulation Language). MCQA met en cache les résultats des tâches et renvoie les résultats mis en cache lors des exécutions répétées pour accélérer le processus. Pour plus de détails sur la facturation, consultez la section Coûts de calcul (paiement à l'utilisation).

Les méthodes execute_sql() et run_sql() ne prennent pas en charge tous les types d'instructions SQL. Pour les instructions autres que DDL et DML, utilisez directement la méthode correspondante. Par exemple, utilisez create_table() pour les instructions CREATE TABLE et utilisez run_xflow() ou execute_xflow() pour les commandes API.

# Create a table using the dedicated method
o.create_table('my_t', 'num bigint, id string', if_not_exists=True)

# Run a SELECT query
result = o.execute_sql('SELECT * FROM pyodps_iris LIMIT 3')
with result.open_reader() as reader:
    for record in reader:
        print(record)

DataFrame

PyODPS fournit une API DataFrame pour le traitement des données. Contrairement à pandas, un DataFrame PyODPS est évalué de manière lazy : il ne s'exécute que lorsque vous appelez une méthode d'exécution immédiate telle que execute() ou persist().

from odps.df import DataFrame

# Load the pyodps_iris table as a DataFrame
iris = DataFrame(o.get_table('pyodps_iris'))

# Filter rows and print results — .execute() triggers the actual computation
for record in iris[iris.sepalwidth < 3].execute():
    print(record)

Par défaut, PyODPS n'affiche pas les détails d'exécution tels que l'URL Logview. Pour activer la sortie détaillée :

from odps import options
options.verbose = True

Pour plus d'exemples DataFrame, consultez la section DataFrame (non recommandé).

Définir les paramètres d'exécution (hints)

Transmettez un dictionnaire hints à tout appel SQL pour définir les paramètres d'exécution par requête :

o.execute_sql('SELECT * FROM pyodps_iris', hints={'odps.sql.mapper.split.size': 16})

Pour appliquer les hints à chaque appel SQL de la session, configurez-les globalement :

from odps import options

# All subsequent execute_sql() calls include these hints automatically
options.sql.settings = {'odps.sql.mapper.split.size': 16}
o.execute_sql('SELECT * FROM pyodps_iris')

Exemple complet

L'exemple suivant couvre le flux de travail complet : connexion à MaxCompute, création d'une table, écriture des données, relecture, interrogation de la table pyodps_iris et nettoyage.

1. Créez un fichier nommé test-pyodps-local.py.

2. Ajoutez le code suivant :

   import os
   from odps import ODPS
   
   o = ODPS(
       access_id=os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
       secret_access_key=os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
       project='<your-default-project>',
       endpoint='<your-endpoint>',
   )
   
   # Create a non-partitioned table with two columns
   table = o.create_table('my_new_table', 'num bigint, id string', if_not_exists=True)
   
   # Write four records to the table
   records = [[111, 'aaa'],
              [222, 'bbb'],
              [333, 'ccc'],
              [444, 'Chinese']]
   o.write_table(table, records)
   
   # Read back the records
   for record in o.read_table(table):
       print(record[0], record[1])
   
   # Query the pyodps_iris table using SQL
   result = o.execute_sql('SELECT * FROM pyodps_iris LIMIT 3;', hints={'odps.sql.allow.fullscan': 'true'})
   
   # Drop the table to release resources
   table.drop()
   
   print('Read data from the pyodps_iris table using open_reader:')
   
   # Print query results
   with result.open_reader() as reader:
       for record in reader:
           print(record[0], record[1])

3. Exécutez le script :

   python test-pyodps-local.py

   Le résultat attendu est le suivant :

   111 aaa
   222 bbb
   333 ccc
   444 Chinese
   Read data from the pyodps_iris table using open_reader:
   4.9 3.0
   4.7 3.2
   4.6 3.1

Étapes suivantes

• Présentation des opérations de base — opérations sur les tables, chargement et téléchargement de données, etc.

• DataFrame (non recommandé) — traitement avancé des données avec l'API DataFrame

• Utiliser un nœud PyODPS pour la segmentation de mots chinois Jieba — un exemple de bout en bout dans DataWorks