DataWorks:Nodo PyODPS 2

Descripción general

PyODPS es el SDK de Python para MaxCompute. Proporciona una interfaz de programación concisa que le permite usar Python para escribir trabajos, consultar tablas y vistas, y gestionar recursos de MaxCompute. Para obtener más información, consulte PyODPS overview. En DataWorks, puede usar un nodo PyODPS para programar y ejecutar tareas de Python e integrarlas con otros trabajos.

Notas

• Si su código requiere paquetes de terceros al ejecutar un nodo PyODPS en un resource group de DataWorks, utilice un serverless resource group y un custom image para instalar los paquetes.

  Nota

  Este método no es compatible si su código usa una UDF que hace referencia a un paquete de terceros. Para obtener información sobre cómo manejar este escenario, consulte UDF example: Use third-party packages in Python UDFs.

• Si su tarea PyODPS necesita acceder a un entorno de red específico, como un origen de datos o servicio en una VPC o un centro de datos local, use un resource group serverless y consulte Network connectivity solutions para establecer la conexión de red necesaria.

• Para obtener más información sobre la sintaxis de PyODPS, consulte PyODPS documentation.

• Los nodos PyODPS están disponibles en dos tipos: PyODPS 2 y PyODPS 3. Se diferencian por su versión subyacente de Python. Los nodos PyODPS 2 usan Python 2 y los nodos PyODPS 3 usan Python 3. Cree el tipo de nodo que corresponda a la versión de Python que utiliza.

• Si la ejecución de SQL en un nodo PyODPS no genera linaje de datos, lo que impide que se muestre en Data Map, puede resolver este problema configurando manualmente los parámetros de programación de DataWorks en el código de la tarea. Para ver el linaje de datos, consulte View data lineage. Para obtener información sobre la configuración de parámetros, consulte Set runtime parameter hints. Puede obtener los parámetros de tiempo de ejecución necesarios mediante el siguiente código de ejemplo.

  import os
  # ...
  # get DataWorks sheduler runtime parameters
  skynet_hints = {}
  for k, v in os.environ.items():
      if k.startswith('SKYNET_'):
          skynet_hints[k] = v
  # ...
  # setting hints while submiting a task
  o.execute_sql('INSERT OVERWRITE TABLE XXXX SELECT * FROM YYYY WHERE ***', hints=skynet_hints)
  # ...

• El registro de salida de un nodo PyODPS admite un tamaño máximo de 4 MB. Evite imprimir grandes volúmenes de datos directamente en el registro. En su lugar, registre solo advertencias esenciales y actualizaciones de progreso.

Límites

• Cuando ejecuta un nodo PyODPS en un resource group exclusivo para programación, se recomienda que los datos procesados localmente dentro del nodo no superen los 50 MB. Esta operación está limitada por las especificaciones del resource group exclusivo para programación. Procesar demasiados datos locales que excedan el umbral del sistema operativo puede causar un error OOM (got killed). Evite escribir demasiado código de procesamiento de datos en nodos PyODPS.

• Cuando ejecuta un nodo PyODPS en un resource group serverless, puede configurar la asignación de CU para el nodo PyODPS según el volumen de datos que se procesarán.

  Nota

  Cuando ejecuta esta tarea en un resource group serverless, la configuración máxima para una sola tarea es 64CU, pero se recomienda no superar 16CU para evitar recursos insuficientes causados por una asignación excesiva de CU, lo cual podría afectar el inicio de la tarea.

• Si encuentra un error Got Killed, indica que el uso de memoria superó el límite y el proceso finalizó. Por lo tanto, evite las operaciones de datos locales tanto como sea posible. Las tareas de SQL y DataFrame iniciadas a través de PyODPS (excepto to_pandas) no están sujetas a esta limitación.

• El código que no es UDF puede usar los paquetes NumPy y Pandas preinstalados. No son compatibles otros paquetes de terceros que contengan código binario.

• Por motivos de compatibilidad, en DataWorks, options.tunnel.use_instance_tunnel se establece en False de forma predeterminada. Si necesita habilitar instance tunnel globalmente, debe establecer manualmente este valor en True.

• La versión subyacente de Python para los nodos PyODPS 2 es la 2.7.

• No se admite la ejecución simultánea de varias tareas de Python dentro de un nodo PyODPS.

• Para imprimir registros en un nodo PyODPS, use print. No se admite el uso de logger.info.

Requisitos previos

Associate a MaxCompute compute engine with your DataWorks workspace.

Procedimiento

1. En la página de edición del nodo PyODPS 2, realice las siguientes operaciones de desarrollo.

   Ejemplos de código PyODPS 2

   Después de crear un nodo PyODPS, puede editar y ejecutar código. Para obtener más información sobre la sintaxis de PyODPS, consulte PyODPS documentation. Este tema describe los siguientes cinco ejemplos de código. Puede seleccionar ejemplos según sus necesidades empresariales.

   Entrada ODPS

   En los nodos PyODPS de DataWorks, se incluye una variable global odps o o como entrada ODPS. No necesita definir manualmente la entrada ODPS.

   print(odps.exist_table('PyODPS_iris'))

   Ejecutar SQL

   Puede ejecutar SQL en un nodo PyODPS. Para obtener más información, consulte Execute SQL statements.

   • De forma predeterminada, instance tunnel no está habilitado en DataWorks, lo que significa que instance.open_reader usa la interfaz Result de forma predeterminada (hasta 10,000 registros). Puede usar reader.count para obtener el número de registros. Si necesita iterar sobre todos los datos, debe deshabilitar la restricción limit. Puede usar las siguientes instrucciones para habilitar Instance Tunnel globalmente y deshabilitar la restricción limit.

     options.tunnel.use_instance_tunnel = True
     options.tunnel.limit_instance_tunnel = False  # Disable limit restriction, read all data.
     
     with instance.open_reader() as reader:
       # Use Instance Tunnel to read all data.

   • También puede agregar tunnel=True a open_reader para habilitar instance tunnel solo para la llamada actual de open_reader. También puede agregar limit=False para deshabilitar la restricción limit solo para la llamada actual.

     # The open_reader uses the Instance Tunnel interface to read all data.
     with instance.open_reader(tunnel=True, limit=False) as reader:

   Establecer parámetros de tiempo de ejecución

   • Puede establecer parámetros de tiempo de ejecución configurando el parámetro hints, que es de tipo dict. Para obtener más información sobre el parámetro hints, consulte SET operations.

     o.execute_sql('select * from PyODPS_iris', hints={'odps.sql.mapper.split.size': 16})

   • Después de configurar sql.settings globalmente, debe agregar los parámetros de tiempo de ejecución relevantes cada vez que ejecute una tarea.

     from odps import options
     options.sql.settings = {'odps.sql.mapper.split.size': 16}
     o.execute_sql('select * from PyODPS_iris')  # Add hints based on global configurationhints。

   Leer resultados de ejecución

   Una instancia que ejecuta SQL puede realizar directamente la operación open_reader. Son posibles los siguientes dos escenarios:

   • La instrucción SQL devuelve datos estructurados.

     with o.execute_sql('select * from dual').open_reader() as reader:
     	for record in reader:  # Process eachrecord。

   • La instrucción SQL puede ser una instrucción como desc. Puede usar la propiedad reader.raw para obtener el resultado de ejecución SQL sin procesar.

     with o.execute_sql('desc dual').open_reader() as reader:
     	print(reader.raw)

     Nota

     Si usa parámetros de programación personalizados, debe codificar de forma rígida los valores de tiempo cuando active directamente un nodo PyODPS 2 en la página, ya que un nodo PyODPS no puede reemplazar directamente los parámetros.

   DataFrame

   También puede usar DataFrame para procesar datos.

   • Ejecución

     En el entorno de DataWorks, la ejecución de DataFrame requiere llamadas explícitas a methods that trigger immediate execution.

     from odps.df import DataFrame
     iris = DataFrame(o.get_table('pyodps_iris'))
     for record in iris[iris.sepal_width < 3].execute():  # Execute immediately and process each Record.

     Si desea activar la ejecución inmediata al usar Print, debe habilitar options.interactive.

     from odps import options
     from odps.df import DataFrame
     options.interactive = True  # Enable the switch at the beginning.
     iris = DataFrame(o.get_table('pyodps_iris'))
     print(iris.sepal_width.sum())  # Executes immediately on print.

   • Imprimir información detallada

     Puede configurar la opción options.verbose. En DataWorks, esta opción está habilitada de forma predeterminada y se imprime información detallada, como la URL de Logview, durante la ejecución.

   Desarrollo de código PyODPS 2

   El siguiente ejemplo sencillo muestra cómo usar un nodo PyODPS:

   1. Prepare un conjunto de datos y cree la tabla de ejemplo pyodps_iris. Para obtener más información, consulte Create a DataFrame.

   2. Cree un DataFrame. Para obtener más información, consulte Create a DataFrame.

   3. Introduzca el siguiente código en el nodo PyODPS.

      from odps.df import DataFrame
      
      # Create DataFrame from ODPS table to create DataFrame.
      iris = DataFrame(o.get_table('pyodps_iris'))
      print(iris.sepallength.head(5))

   Ejecutar una tarea PyODPS

   1. En Run Configuration, configure Compute Resource, Compute Resource and Quota y DataWorks Resource Group.

      Nota

      • Para acceder a orígenes de datos en un entorno de red pública o VPC, use un resource group de programación que haya pasado la prueba de conectividad para el origen de datos. Para obtener más información, consulte Network connectivity solutions.

      • Puede configurar la información de Image según los requisitos de la tarea.

   2. En la barra de herramientas, haga clic en Run para ejecutar la tarea PyODPS.

2. Si necesita ejecutar la tarea del nodo periódicamente, configure los ajustes de programación según sus necesidades empresariales. Para obtener más información, consulte Configure schedule settings.

   A diferencia de los nodos SQL en DataWorks, los nodos PyODPS no reemplazan cadenas como ${param_name} en el código para evitar afectarlo. En su lugar, antes de ejecutar el código, se agrega un dict llamado args a las variables globales, desde el cual se pueden recuperar los parámetros de programación. Por ejemplo, si establece ds=${yyyymmdd} en Parameter, puede recuperar el parámetro en su código de la siguiente manera.

   print('ds=' + args['ds'])
   ds=20240930

   Nota

   Si necesita obtener la partición llamada ds, puede usar el siguiente método.

   o.get_table('table_name').get_partition('ds=' + args['ds'])

3. Después de completar la configuración de la tarea del nodo, debe desplegar el nodo. Para obtener más información, consulte Deploy a node.

4. Después de desplegar la tarea, puede ver el estado de ejecución de la tarea programada en Operation Center. Para obtener más información, consulte View scheduled tasks.

Ejecutar un nodo mediante un rol asociado

Puede associate a RAM role to run a node, lo que le permite ejecutar tareas de nodo con un rol RAM específico para un control de permisos detallado y gestión de seguridad.

Siguiente paso

PyODPS FAQ: Puede obtener información sobre problemas comunes durante la ejecución de PyODPS para una resolución rápida de problemas.