开发PyODPS 2任务

前提条件

已创建PyODPS 2节点，详情请参见创建并管理MaxCompute节点。

背景信息

PyODPS是MaxCompute的Python版本的SDK，提供简单方便的Python编程接口，以便您使用Python编写MaxCompute作业、查询MaxCompute表和视图，以及管理MaxCompute资源，详情请参见PyODPS概述。在DataWorks中，您可通过PyODPS节点实现Python任务的调度运行，以及与其他作业的集成操作。

使用限制

• DataWorks建议您在PyODPS节点内获取到本地处理的数据不超过50 MB，该操作受限于DataWorks执行资源的不同规格（包括公共调度资源组和独享调度资源组），处理的本地数据过多并超出操作系统阈值时可能发生OOM（Got killed）错误。请避免在PyODPS节点中写入过多的数据处理代码。详情请参见高效使用PyODPS最佳实践。

• 如果您发现有Got killed报错，即表明内存使用超限，进程被中止。因此，请尽量避免本地的数据操作。通过PyODPS发起的SQL和DataFrame任务（除to_pandas外）不受此限制。

• 非自定义函数代码可以使用平台预装的Numpy和Pandas。不支持其他带有二进制代码的三方包。

• 由于兼容性原因，在DataWorks中，options.tunnel.use_instance_tunnel默认设置为False。如果需要全局开启instance tunnel，需要手动将该值设置为True。

• PyODPS 2节点底层的Python版本为2.7。

编辑代码：简单示例

创建PyODPS节点后，您可以进行代码编辑及运行，更多关于PyODPS语法说明，请参见基本操作概述。

• ODPS入口

  DataWorks的PyODPS节点中，将会包含一个全局的变量odps或o，即ODPS入口，您无需手动定义ODPS入口。

  print(odps.exist_table('PyODPS_iris'))

• 执行SQL

  您可以在PyODPS节点中执行SQL，详情请参见SQL。

  • DataWorks上默认未开启instance tunnel，即instance.open_reader默认使用Result接口（最多一万条记录）。您可以通过reader.count获取记录数。如果您需要迭代获取全部数据，则需要关闭limit限制。您可以通过下列语句在全局范围内打开Instance Tunnel并关闭limit限制。

    options.tunnel.use_instance_tunnel = True
    options.tunnel.limit_instance_tunnel = False  # 关闭limit限制，读取全部数据。
    
    with instance.open_reader() as reader:
      # 通过Instance Tunnel可读取全部数据。

  • 您也可以通过在open_reader上添加tunnel=True，实现仅对本次open_reader开启instance tunnel。同时，您还可以添加 limit=False，实现仅对本次关闭limit限制。

    # 本次open_reader使用Instance Tunnel接口，且能读取全部数据。
    with instance.open_reader(tunnel=True, limit=False) as reader:

  说明

  若您未开启Instance Tunnel，可能导致获取数据格式错误，解决方法请参见Python SDK常见问题。

• 设置运行参数

  • 您可以通过设置hints参数，来设置运行时的参数，参数类型是dict。 Hints参数的详情请参见SET操作。

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

  • 对全局配置设置sql.settings后，每次运行时，都需要添加相关的运行时的参数。

    from odps import options
    options.sql.settings = {'odps.sql.mapper.split.size': 16}
    o.execute_sql('select * from PyODPS_iris')  # 根据全局配置添加hints。

• 读取运行结果

  运行SQL的实例能够直接执行open_reader的操作，有以下两种情况：

  • SQL返回了结构化的数据。

    with o.execute_sql('select * from dual').open_reader() as reader:
    	for record in reader:  # 处理每一个record。

  • 可能执行的是desc等SQL语句，通过reader.raw属性，获取到原始的SQL执行结果。

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

    说明

    如果使用了自定义调度参数，页面上直接触发运行PyODPS 3节点时，需要写死时间，PyODPS节点无法直接替换。

• DataFrame

  您还可以通过DataFrame的方式处理数据。

  • 执行

    在DataWorks的环境里，DataFrame的执行需要显式调用立即执行的方法。

    from odps.df import DataFrame
    iris = DataFrame(o.get_table('pyodps_iris'))
    for record in iris[iris.sepal_width < 3].execute():  # 调用立即执行的方法，处理每条Record。

    如果您需要在Print时调用立即执行，需要开启options.interactive。

    from odps import options
    from odps.df import DataFrame
    options.interactive = True  # 在开始处打开开关。
    iris = DataFrame(o.get_table('pyodps_iris'))
    print(iris.sepal_width.sum())  # Print时会立即执行。

  • 打印详细信息

    通过设置options.verbose选项。在DataWorks上，默认已经处于打开状态，运行过程会打印Logview等详细过程。

示例

以下以一个简单示例为您介绍PyODPS节点的使用：

1. 准备数据集，创建pyodps_iris示例表，具体操作请参见Dataframe数据处理。

2. 创建DataFrame，详情请参见从MaxCompute表创建DataFrame。

3. 在PyODPS节点中输入以下代码并运行。

   from odps.df import DataFrame
   
   # 从ODPS表创建DataFrame。
   iris = DataFrame(o.get_table('pyodps_iris'))
   print(iris.sepallength.head(5))

   返回结果：

      sepallength
   0          4.5
   1          5.5
   2          4.9
   3          5.0
   4          6.0

编辑代码：进阶示例

若节点需要周期性调度，您需要定义节点调度时的相关属性，调度配置详情请参见任务调度属性配置概述。

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

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

更多场景的PyODPS任务开发，请参考：

• PyODPS使用第三方包

• 节点运行ECS机器上的脚本

• PyODPS节点实现结巴中文分词

• PyODPS节点实现避免将数据下载到本地

后续步骤

• 如何判断Shell自定义脚本任务的成功完成：Python自定义脚本任务的成功完成的判断逻辑与Shell节点一致，您可通过该方法进行判断。

• 发布任务：如果您使用的是标准模式的工作空间，需要通过任务发布流程，将任务发布至生产环境后，任务才会周期调度运行。

• 周期任务运维：任务提交发布至生产运维中心调度后，您可通过DataWorks运维中心进行相关运维操作。

• PyODPS常见问题：您可了解PyODPS执行过程中的常见问题，便于出现异常时快速排查解决。