MaxCompute:Prise en main

Prérequis

• Vous avez activé MaxCompute.

• Vous avez créé un projet MaxCompute.

• Vous avez créé un espace de travail DataWorks et lié des ressources de calcul à cet espace.

Procédure

1. Créez un nœud PyODPS.

   Cette rubrique s'appuie sur un nœud PyODPS dans DataWorks. Pour plus de détails, consultez la rubrique Développer une tâche PyODPS 3.

   • Nous prenons ici l'exemple d'un nœud PyODPS 3. La version Python sous-jacente pour les nœuds PyODPS 3 est la 3,7.

   • Un nœud PyODPS peut traiter au maximum 50 Mo de données locales et utiliser jusqu'à 1 Go de mémoire lors de l'exécution. Si une tâche dépasse ces limites, le système l'interrompt. Évitez donc d'écrire du code Python traitant de grands volumes de données dans une tâche PyODPS.

   • L'écriture et le débogage du code dans DataWorks sont moins efficaces. Pour améliorer votre productivité, nous vous recommandons de développer votre code localement dans IntelliJ IDEA.

   1. Créez un workflow.

      Accédez à la page Data Studio, cliquez avec le bouton droit sur Business Flow, puis sélectionnez Create Workflow.

   2. Créez un nœud PyODPS.

      Cliquez avec le bouton droit sur le nouveau workflow, choisissez Create Node > MaxCompute > PyODPS 3, saisissez un nom pour le nœud, puis cliquez sur Submit.

2. Modifiez le nœud PyODPS.

   1. Rédigez le code.

      Saisissez le code de test dans l'éditeur du nœud PyODPS. L'exemple suivant illustre l'utilisation des API PyODPS pour les opérations sur les tables. Pour plus d'informations sur les opérations relatives aux tables et aux instructions SQL, consultez les rubriques Tables et SQL.

      from odps import ODPS
      
      # In DataWorks PyODPS nodes, a global variable (`o` or `odps`) is available by default as the MaxCompute entry. You can use it directly without defining it manually.
      table = o.create_table('my_new_table', 'num bigint, id string', if_not_exists=True)
      
      # Insert data into the non-partitioned table my_new_table.
      records = [[111, 'aaa'],
                [222, 'bbb'],
                [333, 'ccc'],
                [444, '中文']]
      o.write_table(table, records)
      
      # Read data from the non-partitioned table my_new_table.
      for record in o.read_table(table):
          print(record[0],record[1])
      
      # Read data from the table by executing an SQL statement.
      result = o.execute_sql('select * from my_new_table;',hints={'odps.sql.allow.fullscan': 'true'})
      
      # Read the SQL execution results.
      with result.open_reader() as reader:    
          for record in reader:            
              print(record[0],record[1])
      
      # Drop the table to release resources.
      table.drop()

   2. Exécutez le code.

      Une fois le code modifié, cliquez sur l'icône . Lorsque l'exécution est terminée, consultez les résultats dans l'onglet Runtime Log. La sortie de journal suivante indique que l'opération a réussi.

      2023-07-21 15:06:41 INFO ========================================================
      Executing user script with PyODPS 0.11.2.3
      Tunnel session created: <TableUploadSession xxx                                          >
      Tunnel session created: <TableDownloadSession xxx                                        >
      111 aaa
      222 bbb
      333 ccc
      444 中文
      Tunnel session created: <InstanceDownloadSession id=xxx          project_name=xxx                              >
      111 aaa
      222 bbb
      333 ccc
      444 中文
      2023-07-21 15:06:49 INFO ========================================================
      2023-07-21 15:06:49 INFO Exit code of the Shell command 0
      2023-07-21 15:06:49 INFO --- Invocation of Shell command completed ---
      2023-07-21 15:06:49 INFO Shell run successfully!
      2023-07-21 15:06:49 INFO Current task status: FINISH
      2023-07-21 15:06:49 INFO Cost time is: 7.507s
      /home/admin/alisatasknode/taskinfo/xxx              xxx        .log-END-EOF