MaxCompute:SDK Python Open Storage

MaxCompute expose les interfaces de l'API Storage via le SDK Python. Pour plus d'informations, consultez aliyun-odps-python-sdk.

Prérequis

Si vous exécutez le code dans un environnement local, assurez-vous que PyODPS est installé. Pour en savoir plus, consultez la rubrique Installation de PyODPS.

PyODPS est également disponible dans les environnements suivants :

• DataWorks : PyODPS est préinstallé sur les nœuds PyODPS. Développez et exécutez périodiquement des tâches PyODPS directement sur ces nœuds. Pour plus de détails, reportez-vous à la section Utilisation de PyODPS dans DataWorks.

• PAI : PyODPS est préinstallé dans toutes les images PAI intégrées et fonctionne immédiatement avec des composants tels que le composant Python personnalisé dans PAI-Designer. L'utilisation dans les notebooks PAI suit le flux de travail standard de PyODPS. Pour plus d'informations, voir Présentation des opérations de base et DataFrame (non recommandé).

Exemples

Pour consulter des exemples de code complets, rendez-vous sur la page Exemples du SDK Python.

1. Configurez l'environnement pour vous connecter au service MaxCompute

   import os
   from odps import ODPS
   from odps.apis.storage_api import *
   # Ensure that the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is set to your Access Key ID,
   # and the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is set to your Access Key Secret.
   # For security, avoid hardcoding the Access Key ID and Access Key Secret.
   # The endpoint of the MaxCompute service. Only connections from VPC networks are supported.
   o = ODPS(
   		os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
   		os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
   		project='your-default-project',
   		endpoint='your-end-point'
   )
   # The name of the MaxCompute table to access.
   table = "<table to access>"
   # The name of the quota to use for accessing MaxCompute.
   quota_name = "<quota name>"
   # Connects to the MaxCompute service and creates an Arrow-format Storage API client.
   def get_arrow_client():
       odps_table = o.get_table(table)
       client = StorageApiArrowClient(odps=o, table=odps_table, quota_name=quota_name)
   
       return client
   

   Remarque

   Pour obtenir le nom du quota d'un groupe de ressources dédié à l'API Storage (abonnement) :

   • Groupe de ressources dédié à l'API Storage : connectez-vous à la console MaxCompute. Dans le coin supérieur gauche, sélectionnez votre région. Dans le volet de navigation de gauche, choisissez Workspace > Quotas pour afficher les quotas disponibles. Pour en savoir plus, consultez la rubrique Gestion des quotas pour les ressources de calcul.

   • API Storage : connectez-vous à la console MaxCompute. Dans le volet de navigation de gauche, sélectionnez Tenants > Tenant Property pour activer l'API Storage.

2. Lisez les données de la table

   1. Créez une session de lecture pour lire les données depuis MaxCompute

      import logging
      import sys
      from odps.apis.storage_api import *
      from util import *
      
      logger = logging.getLogger(__name__)
      # Creates a read session. The mode parameter specifies the split strategy: 'size' to split by data size, or 'row' to split by row offset.
      def create_read_session(mode):
          client = get_arrow_client()
          req = TableBatchScanRequest(required_partitions=['pt=test_write_1'])
      
          if mode == "size":
              req.split_options = SplitOptions.get_default_options(SplitOptions.SplitMode.SIZE)
          elif mode == "row":
              req.split_options = SplitOptions.get_default_options(SplitOptions.SplitMode.ROW_OFFSET)
      
          resp = client.create_read_session(req)
      
          if resp.status != Status.OK:
              logger.info("Create read session failed")
              return
      
          logger.info("Read session id: " + resp.session_id)
      
      if __name__ == '__main__':
          logging.basicConfig(format='%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s', level=logging.INFO)
          if len(sys.argv) != 2:
              raise ValueError("Please provide split mode: size|row")
      
          mode = sys.argv[1]
          if mode != "row" and mode != "size":
              raise ValueError("Please provide split mode: size|row")
      
          create_read_session(mode)
      

   2. Vérifiez le statut d'une session de lecture

      import logging
      import sys
      import time
      from odps.apis.storage_api import *
      from util import *
      
      logger = logging.getLogger(__name__)
      # Before reading data, ensure the read session is created and ready.
      def check_session_status(session_id):
          client = get_arrow_client()
          req = SessionRequest(session_id=session_id)
          resp = client.get_read_session(req)
      
          if resp.status != Status.OK:
              logger.info("Get read session failed")
              return
      
          # Session creation can be time-consuming. You must wait for the session status to become NORMAL before reading data.
          if resp.session_status == SessionStatus.NORMAL:
              logger.info("Read session id: " + resp.session_id)
          else:
              logger.info("Session status is not expected")
      
      if __name__ == '__main__':
          logging.basicConfig(format='%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s', level=logging.INFO)
          if len(sys.argv) != 2:
              raise ValueError("Please provide session id")
      
          session_id = sys.argv[1]
          check_session_status(session_id)
      

   3. Lisez les données MaxCompute

      # Reads data rows from MaxCompute for a specified session_id and counts the total number of rows.
      import logging
      import sys
      from odps.apis.storage_api import *
      from util import *
      
      logger = logging.getLogger(__name__)
      
      def read_rows(session_id):
          client = get_arrow_client()
          req = SessionRequest(session_id=session_id)
          resp = client.get_read_session(req)
      
          if resp.status != Status.OK and resp.status != Status.WAIT:
              logger.info("Get read session failed")
              return
      
          req = ReadRowsRequest(session_id=session_id)
          if resp.split_count == -1:
              req.row_index = 0
              req.row_count = resp.record_count
          else:
              req.split_index = 0
      
          reader = client.read_rows_arrow(req)
          total_line = 0
          while True:
              record_batch = reader.read()
              if record_batch is None:
                  break
              total_line += record_batch.num_rows
      
          if reader.get_status() != Status.OK:
              logger.info("Read rows failed")
              return
      
          logger.info("Total line is:" + str(total_line))
      
      if __name__ == '__main__':
          logging.basicConfig(format='%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s', level=logging.INFO)
          if len(sys.argv) != 2:
              raise ValueError("Please provide session id")
      
          session_id = sys.argv[1]
          read_rows(session_id)
      

Documents connexes

Pour en savoir plus sur l'API Storage, consultez la rubrique Présentation de l'API Storage.