PyODPS 2节点

操作步骤

1. 在PyODPS 2节点编辑页面，执行如下开发操作。

   PyODPS 2代码示例

   创建PyODPS节点后，您可以进行代码编辑及运行，更多关于PyODPS语法说明，请参见基本操作概述。本文为您介绍以下五种代码示例，您可根据实际业务需要，选择示例内容。

   ODPS入口

   执行SQL

   设置运行参数

   读取运行结果

   DataFrame

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

   print(odps.exist_table('PyODPS_iris'))

   您可以在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 2节点时，需要写死时间，PyODPS节点无法直接替换。

   您还可以通过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 2代码开发

   以下以一个简单示例为您介绍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))

   执行PyODPS任务

   1. 在调试配置的计算资源中，选择配置计算资源、计算配额和DataWorks资源组。

      说明

      • 访问公共网络或VPC网络环境的数据源需要使用与数据源测试连通性成功的调度资源组。详情请参见网络连通方案。

      • 您可以根据任务情况选择配置镜像信息。

   2. 在工具栏单击运行PyODPS任务。

2. 如需定期执行节点任务，请根据业务需求配置调度信息。配置详情请参见调度配置。

   与DataWorks中的SQL节点不同，为了避免影响代码，PyODPS节点不会在代码中替换类似 ${param_name}的字符串，而是在执行代码前，在全局变量中增加一个名为args的dict，调度参数可以在此获取。例如，在参数中设置ds=${yyyymmdd}，则可以通过以下方式在代码中获取该参数。

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

   说明

   如果您需要获取名为ds的分区，则可以使用如下方法。

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

3. 节点任务配置完成后，需对节点进行发布。详情请参见节点/工作流发布。

4. 任务发布后，您可以在运维中心查看周期任务的运行情况。详情请参见运维中心入门。