PyODPS概述

PyODPS应用场景请参见：

• DataFrame操作：DataFrame快速入门。

• 读取分区表数据：PyODPS读取分区表数据。

• 参数传递：PyODPS参数传递。

• 使用第三方包：PyODPS使用第三方包。

• 查看一级分区：PyODPS查看一级分区。

• 条件查询：PyODPS条件查询。

• DataFrame Sequence及执行：PyODPS的Sequence及执行操作。

PyODPS支持在本地环境、DataWorks、PAI Notebooks中使用。

• 本地环境：您可以在本地环境安装并使用PyODPS，操作指导可参见通过本地环境使用PyODPS。

• DataWorks：DataWorks的PyODPS节点已安装好了PyODPS，您可以直接在DataWorks的PyODPS节点上开发PyODPS任务并周期性运行，操作指导请参见通过DataWorks使用PyODPS。

• PAI Notebooks：PAI的Python环境也可安装运行PyODPS，其中PAI的内置镜像均已安装好了PyODPS可直接使用，如PAI-Designer的自定义Python组件，在PAI Notebooks中使用PyODPS的方式与通用的使用方式基本一致，可参考基本操作概述、DataFrame概述。

PyODPS作为一个SDK，本身运行于各种客户端，包括PC、DataWorks（数据开发的PyODPS节点）或PAI Notebooks的运行环境。需要注意的是，PyODPS提供了多种方便拉取数据到本地的操作，如tunnel下载操作、execute操作、to_pandas操作等，因此，很多初始使用PyODPS的用户会试图把数据拉取到本地，处理完成后再上传到 MaxCompute上，很多时候这种方式是十分低效的，拉取数据到本地彻底丧失了MaxCompute的大规模并行能力的优势。

以下以一个分词的示例为例，为您对比两种方式的代码区别。

• 示例场景

  用户需要通过分析每天产生的日志字符串来提取一些信息，有一个只有一列的表，类型是string，通过jieba分词可以将中文语句分词，然后再找到想要的关键词存储到信息表里。

• 低效处理代码demo

  import jieba
  t = o.get_table('word_split')
  out = []
  with t.open_reader() as reader:
      for r in reader:
          words = list(jieba.cut(r[0]))
          #
          # 处理逻辑，产生出 processed_data
          #
          out.append(processed_data)
  out_t = o.get_table('words')
  with out_t.open_writer() as writer:
      writer.write(out)

  单机处理数据的思维，逐行读取数据，然后逐行处理数据，再逐行写入目标表。整个流程中，下载上传数据消耗了大量的时间，并且在执行脚本的机器上需要很大的内存处理所有的数据，特别是对于使用DataWorks节点的用户来说，很容易因为超过默认分配的内存值而导致OOM运行报错。

• 高效处理代码demo

  from odps.df import output
  out_table = o.get_table('words')
  df = o.get_table('word_split').to_df()
  
  # 假定需要返回的字段及类型如下
  out_names = ["word", "count"]
  out_types = ["string", "int"]
  
  @output(out_names, out_types)
  def handle(row):
      import jieba
      words = list(jieba.cut(row[0]))
      #
      # 处理逻辑，产生出 processed_data
      #
      yield processed_data
  df.apply(handle, axis=1).persist(out_table.name)

  利用apply实现分布式执行：

  • 复杂逻辑都放在handle这个函数里，这个函数会被自动序列化到服务端作为UDF使用，在服务端调用执行，且因为handle服务端实际执行时也是对每一行进行处理的，所以逻辑上是没有区别的。不同的是，这样写的程序在提交到MaxCompute端执行时，有多台机器同时处理数据，可以节约很多时间。

  • 调用persist接口会将产生的数据直接写到另一张MaxCompute表中，所有的数据产生与消费都在 MaxCompute集群完成，也节约了本地的网络与内存。

  • 在这个例子中也使用到了三方包，MaxCompute是支持自定义函数中使用三方包的（示例中的jieba），所以无需担心代码改动带来的成本，您可以几乎不需要改动主要逻辑就可以享受到MaxCompute的大规模计算能力。