MaxCompute:Open Storage Python SDK

MaxCompute, Storage API arayüzlerini Python SDK aracılığıyla sunar. Daha fazla bilgi için bkz. aliyun-odps-python-sdk.

Önkoşullar

Kodu yerel ortamda çalıştırıyorsanız PyODPS'in yüklü olduğundan emin olun. Daha fazla bilgi için bkz. PyODPS'i Yükleme.

PyODPS aşağıdaki ortamlarda da kullanılabilir:

• DataWorks: PyODPS, PyODPS düğümlerinde önceden yüklenmiştir. Bu düğümlerde PyODPS görevlerini doğrudan geliştirebilir ve periyodik olarak çalıştırabilirsiniz. Daha fazla bilgi için bkz. DataWorks'te PyODPS Kullanma.

• PAI: PyODPS, tüm yerleşik PAI görüntülerinde önceden yüklenmiştir ve PAI-Designer'daki özel Python bileşeni gibi bileşenlerde kutudan çıkar çıkmaz çalışır. PAI Notebooks'ta kullanım, standart PyODPS iş akışını takip eder. Daha fazla bilgi için bkz. Temel İşlemlere Genel Bakış ve DataFrame (Önerilmez).

Örnekler

Tam kod örnekleri için bkz. Python SDK Örnekleri.

1. MaxCompute hizmetine bağlanmak için ortamı yapılandırın

   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
   

   Not

   Storage API özel kaynak grubu (abonelik) için kota adını almak için:

   • Storage API özel kaynak grubu: MaxCompute konsolu üzerinden oturum açın. Sol üst köşeden bölgenizi seçin. Sol gezinme bölmesinde Workspace > Quotas öğelerini seçerek mevcut kotaları görüntüleyin. Daha fazla bilgi için bkz. Hesaplama kaynakları için kotaları yönetme.

   • Storage API: MaxCompute konsolu üzerinden oturum açın. Sol gezinme bölmesinde Tenants > Tenant Property öğelerini seçerek Storage API'yi etkinleştirin.

2. Tablo verilerini okuyun

   1. MaxCompute'den veri okumak için bir okuma oturumu oluşturun

      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. Bir okuma oturumunun durumunu kontrol edin

      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. MaxCompute verilerini okuyun

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

İlgili belgeler

Storage API hakkında daha fazla bilgi için bkz. Storage API'ye Genel Bakış.