DataWorks:Nodo PyODPS 3

Introducción

PyODPS es el SDK de Python para MaxCompute. Permite escribir trabajos, consultar tablas y vistas, y gestionar recursos de MaxCompute mediante Python. Para obtener más información, consulte PyODPS overview. En DataWorks, los nodos PyODPS permiten programar y ejecutar tareas de Python junto con otros trabajos.

Notas de uso

• Para invocar un paquete de terceros desde un nodo PyODPS en un resource group de DataWorks, utilice un serverless resource group con un custom image.

  Nota

  Este método no aplica si su código incluye una función definida por el usuario (UDF) que hace referencia a un paquete de terceros. Para conocer el procedimiento correcto, consulte UDF example: Use a third-party package in a Python UDF.

• Para actualizar la versión de PyODPS, puede usar un custom image en un resource group serverless para ejecutar el comando /home/tops/bin/pip3 install pyodps==0.12.1. Puede reemplazar 0.12.1 con la versión de PyODPS deseada. Para un resource group exclusivo de programación, utilice O&M Assistant para ejecutar el mismo comando.

• Si su trabajo de PyODPS necesita acceder a un entorno de red específico, como un origen de datos o servicio en una VPC o una red IDC, utilice un resource group serverless. Para obtener más información sobre cómo conectar el resource group serverless al entorno destino, consulte Network connectivity solutions.

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

• Existen dos tipos de nodos PyODPS: PyODPS 2 (Python 2) y PyODPS 3 (Python 3). Cree el tipo de nodo que corresponda a su versión de Python.

• Si la ejecución de sentencias SQL en un nodo PyODPS no genera el linaje de datos correcto en Data Map, establezca manualmente los parámetros de programación de DataWorks correspondientes en el código del trabajo. Para obtener información sobre cómo ver el linaje de datos, consulte View data lineage. Para obtener información sobre cómo establecer parámetros, consulte Set runtime parameter hints. El siguiente código de ejemplo obtiene los parámetros obligatorios durante la ejecución.

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

• El registro de salida de un nodo PyODPS tiene un tamaño máximo de 4 MB. Evite imprimir grandes volúmenes de datos en el registro. Concéntrese en mostrar información de alertas y progreso.

Límites

• Al ejecutar un nodo PyODPS en un resource group exclusivo de programación, se recomienda que los datos procesados localmente no superen los 50 MB. Este límite lo imponen las especificaciones del resource group exclusivo. Si los datos locales superan el umbral del sistema operativo, puede producirse un error OOM (Got Killed). Evite escribir código de procesamiento de datos excesivo en el nodo PyODPS.

• Al ejecutar un nodo PyODPS en un resource group serverless, configure un número adecuado de CU según el volumen de datos.

  Nota

  En un resource group serverless, un solo trabajo admite un máximo de 64 CUs. Sin embargo, se recomienda no exceder 16 CUs para evitar escasez de recursos que pueda afectar el inicio del trabajo.

• Un error Got killed indica que el proceso superó el límite de memoria. Evite las operaciones de datos locales. Los trabajos de SQL y DataFrame (excepto las operaciones to_pandas) iniciados mediante PyODPS no están sujetos a esta limitación.

• Puede utilizar las bibliotecas NumPy y pandas preinstaladas en código que no involucre funciones definidas por el usuario. No se admiten otros paquetes de terceros que contengan código binario.

• Por motivos de compatibilidad, options.tunnel.use_instance_tunnel está establecido en False de forma predeterminada en DataWorks. Para habilitar instance tunnel globalmente, establezca manualmente este valor en True.

• La definición de bytecode difiere entre versiones menores de Python 3, como Python 3.8 y Python 3.7.

  Actualmente, MaxCompute utiliza Python 3.7. Si usa sintaxis de otras versiones de Python 3, como el finally block de Python 3.8, se producirá un error durante la ejecución. Se recomienda utilizar Python 3.7.

• PyODPS 3 admite la ejecución en un resource group serverless. Para adquirir y utilizar uno, consulte Use a serverless resource group.

• No es posible configurar varios trabajos de Python para que se ejecuten simultáneamente dentro de un solo nodo PyODPS.

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

Requisitos previos

Associate a MaxCompute compute engine with your DataWorks workspace.

Procedimiento

1. Desarrolle su código en la página del editor del nodo PyODPS 3.

   Ejemplos de código PyODPS 3

   Una vez creado un nodo PyODPS, puede editar y ejecutar su código. Para obtener más información sobre la sintaxis de PyODPS, consulte Basic operations. Los siguientes ejemplos abarcan cinco escenarios comunes. Elija el que se ajuste a sus necesidades.

   Punto de entrada ODPS

   Cada nodo PyODPS en DataWorks incluye una variable global de punto de entrada ODPS, odps o o, que no necesita definir manualmente.

   print(odps.exist_table('PyODPS_iris'))

   Ejecutar SQL

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

   • De forma predeterminada, instance tunnel está deshabilitado en DataWorks, por lo que instance.open_reader utiliza la interfaz Result y devuelve un máximo de 10,000 registros. Puede usar reader.count para obtener el número de registros. Para iterar sobre todos los datos, deshabilite el limit. Las siguientes sentencias habilitan globalmente instance tunnel y deshabilitan el limit.

     options.tunnel.use_instance_tunnel = True
     options.tunnel.limit_instance_tunnel = False  # Disable the limit to read all data.
     
     with instance.open_reader() as reader:
       # All data can be read through Instance Tunnel.

   • También puede habilitar instance tunnel para una sola llamada a open_reader agregando tunnel=True a la llamada de open_reader. También puede agregar limit=False para deshabilitar la restricción de limit en dicha llamada.

     # Use Instance Tunnel for this open_reader call and read all data.
     with instance.open_reader(tunnel=True, limit=False) as reader:

   Establecer parámetros de ejecución

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

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

   • Si establece configuraciones globales mediante sql.settings, los parámetros de ejecución correspondientes se agregan a cada ejecución.

     from odps import options
     options.sql.settings = {'odps.sql.mapper.split.size': 16}
     o.execute_sql('select * from PyODPS_iris')  # Hints are added based on the global settings.

   Leer resultados de ejecución

   Puede invocar open_reader directamente en una instancia de ejecución de SQL. Esto admite dos escenarios:

   • La sentencia SQL devuelve datos estructurados.

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

   • Si ejecuta sentencias SQL como desc, puede usar la propiedad reader.raw para obtener los resultados sin procesar de la ejecución SQL.

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

     Nota

     Si utiliza parámetros de programación personalizados y ejecuta un nodo PyODPS 3 directamente desde la interfaz, debe codificar el valor de tiempo de forma fija, ya que el nodo no puede sustituir variables durante la ejecución.

   DataFrame

   También puede procesar datos mediante DataFrame (Deprecated).

   • Ejecución

     En el entorno de DataWorks, las operaciones de DataFrame requieren una llamada explícita a un immediate execution method.

     from odps.df import DataFrame
     iris = DataFrame(o.get_table('pyodps_iris'))
     for record in iris[iris.sepal_width < 3].execute():  # Call an immediate execution method to process each record.

     Si necesita desencadenar la ejecución inmediata al imprimir, debe habilitar options.interactive.

     from odps import options
     from odps.df import DataFrame
     options.interactive = True  # Enable the option at the beginning.
     iris = DataFrame(o.get_table('pyodps_iris'))
     print(iris.sepal_width.sum())  # This triggers immediate execution.

   • Imprimir información detallada

     options.verbose está habilitado de forma predeterminada en DataWorks, lo que provoca que se imprima información detallada, como la URL de Logview, durante la ejecución.

   Desarrollo de código PyODPS 3

   El siguiente ejemplo muestra cómo utilizar un nodo PyODPS:

   1. Prepare un conjunto de datos creando la tabla de ejemplo pyodps_iris. Para obtener más información, consulte Process DataFrame data.

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

   3. Introduzca el siguiente código en el nodo PyODPS y ejecútelo.

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

   Ejecutar el trabajo de PyODPS****

   1. En el panel Run Configuration, dentro de la sección Compute Resource, configure el Compute Resource, la computing quota y el DataWorks Resource Group.

      Nota

      • Para acceder a un origen de datos a través de una red pública o una VPC, debe utilizar un resource group de programación que haya pasado una prueba de conectividad con dicho origen. Para obtener más información, consulte Network connectivity solutions.

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

   2. En el cuadro de diálogo de parámetros de la barra de herramientas, seleccione el origen de datos de MaxCompute creado y haga clic en Run para ejecutar el trabajo de PyODPS.

2. Para ejecutar el nodo periódicamente, configure sus propiedades de programación según sus requisitos empresariales. Para obtener más información, consulte Configure node scheduling.

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

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

   Nota

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

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

3. Una vez configurado el nodo, debe desplegarlo. Para obtener más información, consulte Node and workflow deployment.

4. Después de desplegar el trabajo, puede consultar su estado en Operation Center. Para obtener más información, consulte Get started with Operation Center.

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 de RAM específico para lograr un control de permisos detallado y una gestión de seguridad adecuada.

Siguientes pasos

FAQ about PyODPS: Problemas comunes durante la ejecución de PyODPS y cómo resolverlos.