MaxCompute:Desarrollar un trabajo de PyODPS 3

Requisitos previos

Ya creó un nodo de PyODPS 3. Para obtener más información, consulte Create and manage MaxCompute nodes.

Contexto

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

Notas de uso

• Si su código de PyODPS requiere paquetes de terceros, instálelos mediante un serverless resource group y un custom image.

  Nota

  Este método no es compatible si su código incluye una función definida por el usuario (UDF) que hace referencia a un paquete de terceros. Para conocer la configuración correcta, consulte UDF example: Use third-party packages in Python UDFs.

• Para actualizar la versión de PyODPS, use un custom image y ejecute el comando /home/tops/bin/pip3 install pyodps==0.12.1 en un serverless resource group (puede reemplazar 0.12.1 por la versión de PyODPS deseada). También puede usar O&M Assistant para ejecutar el mismo comando en un exclusive resource group for scheduling.

• Si su trabajo de PyODPS necesita acceder a un entorno de red especial, como un origen de datos o un servicio en una VPC o en un centro de datos local (IDC), utilice un serverless resource group y establezca una conexión de red entre dicho grupo y el entorno de destino. Para obtener más información, consulte Network connectivity solutions.

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

• Los nodos de PyODPS están disponibles en dos tipos: PyODPS 2 y PyODPS 3. Cada uno utiliza una versión subyacente distinta de Python: los nodos de PyODPS 2 usan Python 2 y los de PyODPS 3 usan Python 3. Asegúrese de crear el tipo de nodo que coincida con su versión de Python.

• Si la ejecución de SQL en un nodo de PyODPS no genera correctamente la data lineage, lo cual impide que aparezca en Data Map, resuelva el problema estableciendo manualmente los parámetros de programación y de tiempo de ejecución de DataWorks en su código. Para saber cómo consultar el linaje de datos, consulte View data lineage. Para conocer la configuración de parámetros, consulte Set runtime parameter hints. Puede obtener los parámetros de tiempo de ejecución necesarios con 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 tamaño máximo del registro de salida de un nodo de PyODPS es de 4 MB. Evite imprimir resultados de datos extensos en el registro; en su lugar, muestre únicamente la información esencial de alertas y progreso.

Límites

• Cuando ejecute un nodo de PyODPS en un exclusive resource group for scheduling, no procese más de 50 MB de datos locales. Esto se debe a las especificaciones de recursos de dicho grupo. Procesar una gran cantidad de datos locales que supere el umbral del sistema operativo puede provocar un error de falta de memoria (OOM), indicado con el mensaje Got Killed. Evite escribir código de procesamiento de datos extenso directamente en el nodo de PyODPS.

• Cuando ejecute un nodo de PYODPS con un serverless resource group, configure las CU del nodo según la cantidad de datos que necesite procesar.

  Nota

  Al ejecutar una tarea en un serverless resource group, una sola tarea admite una configuración máxima de 64CU. Sin embargo, se recomienda no superar 16CU para evitar la escasez de recursos provocada por un valor de CU excesivo, lo cual podría afectar el inicio de la tarea.

• Un error Got killed indica que el uso de memory superó el límite, lo que provocó la finalización del proceso. Para evitarlo, evite las operaciones con datos locales. Esta limitación no aplica a los trabajos de SQL ni de DataFrame iniciados mediante PyODPS (excepto to_pandas).

• Puede utilizar las bibliotecas Numpy y Pandas preinstaladas para el código que no involucre funciones personalizadas. No se admiten otros paquetes de terceros que contengan código binario.

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

• La definición de bytecode varía entre las versiones menores de Python 3, como Python 3.8 y Python 3.7.

  Actualmente, MaxCompute usa Python 3.7. Se producirá un error de ejecución si utiliza sintaxis de otras versiones de Python 3, como el bloque finally de Python 3.8. Se recomienda usar Python 3.7.

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

• No se admite la ejecución simultánea de varios trabajos de Python dentro de un solo nodo de PyODPS.

Editar código: ejemplo básico

Una vez creado el nodo de PyODPS, edite y ejecute su código. Para obtener más información sobre la sintaxis de PyODPS, consulte Overview of basic operations.

• Punto de entrada de ODPS

  Un nodo de PyODPS de DataWorks proporciona una variable global, llamada odps o o, que funciona como punto de entrada de ODPS. No es necesario definirla manualmente.

  print(odps.exist_table('PyODPS_iris'))

• Ejecutar SQL

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

  • De forma predeterminada, instance tunnel está deshabilitado en DataWorks. Esto significa que instance.open_reader utiliza la interfaz Result, la cual lee un máximo de 10,000 registros. Utilice reader.count para obtener la cantidad de registros. Para iterar por todos los datos, debe deshabilitar limit. Use las siguientes instrucciones para habilitar instance tunnel de manera global y deshabilitar 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 the instance tunnel.

  • También puede agregar tunnel=True a open_reader para habilitar instance tunnel en la llamada actual a open_reader. De igual forma, agregue limit=False para deshabilitar la restricción limit en la llamada actual.

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

• Parámetros de tiempo de ejecución

  • Establezca los parámetros de tiempo de ejecución mediante el parámetro hints, que es de tipo 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 configura sql.settings en la configuración global, estos parámetros de tiempo de ejecución se agregarán a cada ejecución.

    from odps import options
    options.sql.settings = {'odps.sql.mapper.split.size': 16}
    o.execute_sql('select * from PyODPS_iris')  # This call includes hints from the global configuration.

• Resultados de la ejecución

  Una instancia de ejecución de SQL puede ejecutar directamente la operación open_reader en 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 each record.

  • Al ejecutar instrucciones como desc, recupere el resultado de ejecución de SQL sin procesar mediante la propiedad reader.raw.

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

    Nota

    Si utiliza parámetros de programación personalizados, debe codificar de forma fija la hora cuando active directamente la ejecución de un nodo de PyODPS 3 desde la página. El nodo de PyODPS no puede sustituir este valor de manera directa.

• DataFrame

  También puede procesar datos mediante un DataFrame (not recommended).

  • Ejecución

    En el entorno de DataWorks, las operaciones de DataFrame deben activarse explícitamente llamando a un immediately executed method.

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

    Si necesita activar una ejecución inmediata al imprimir, habilite 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())  # An immediate execution is triggered when printing.

  • Imprimir información detallada

    Configure la opción options.verbose. Esta opción está habilitada de forma predeterminada en DataWorks e imprime información detallada durante la ejecución, como la URL de Logview.

Ejemplo

A continuación se muestra cómo utilizar un nodo de PyODPS:

1. Prepare el conjunto de datos y cree la tabla de ejemplo pyodps_iris. Para obtener más detalles, consulte Process data by using DataFrame.

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

3. Introduzca y ejecute el siguiente código en el nodo de PyODPS.

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

   Se devuelve el siguiente resultado:

      sepallength
   0          4.5
   1          5.5
   2          4.9
   3          5.0
   4          6.0

Editar código: ejemplo avanzado

Si el nodo debe ejecutarse periódicamente, defina sus propiedades de programación. Para obtener más información, consulte Configure scheduling properties for a node.

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

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

Para obtener más información sobre el desarrollo de trabajos de PyODPS en otros casos de uso, consulte los siguientes temas:

• Use third-party packages in PyODPS.

• Segment Chinese text with Jieba in a PyODPS node.

• Use a PyODPS node to avoid downloading data to a local computer.

Pasos siguientes

• Determine whether a custom Shell script has run successfully: la lógica para determinar si un script de Python personalizado se ejecutó correctamente es la misma que se aplica a un script de shell. Utilice este método para comprobarlo.

• Deploy a job: si utiliza un espacio de trabajo en modo estándar, despliegue el trabajo en el entorno de producción para que pueda ejecutarse periódicamente.

• O&M for periodically-run jobs: después de desplegar y programar un trabajo en el entorno de producción, gestione su operación y mantenimiento en Operation Center.

• PyODPS FAQ: encuentre respuestas a preguntas frecuentes sobre la ejecución de trabajos de PyODPS para solucionar problemas rápidamente.

Preguntas frecuentes

P: Estoy utilizando un nodo PyODPS3 para recopilar datos de una API de terceros, como Lark, e importarlos a DataWorks. El código se ejecuta sin problemas en mi entorno de desarrollo local, pero reporta un error de tiempo de espera de respuesta cuando se envía al entorno de producción y se ejecuta en Operation Center. ¿Por qué ocurre esto?

R: Vaya a Management Center > Security Settings > Workspace. En la lista blanca del sandbox, agregue el nombre de dominio de la API de terceros para otorgarle acceso al trabajo de PyODPS 3. Por ejemplo:

Agregue el nombre de dominio de la API de Lark open.feishu.cn y establezca el puerto en 443.