BasePipeline

Pipeline基类定义

class bigflow.pipeline.pipeline_base.PipelineBase(**options)

基类：object

Pipeline基类

• add_archive(file_path, resource_path)

  向Pipeline添加一个压缩文件，使得该文件在运行期自动被解包

  参数:	file_path (str) — 文件路径，目前仅支持HDFS resource_path (str) — 运行期访问该文件解压后的路径

• add_cache_id(cache_id)

  save the ptype cache node id for use

• add_directory(dir_path, resource_path=’’, is_only_python_files_accepted=False)

  向Pipeline添加一个目录，使得该目录下的文件/子目录能够在运行期被访问

  参数:	dir_path (str) — 需要添加的本地目录路径 resource_path (str) — 计算引擎运行时访问该目录的路径, 应是相对路径. 如未提供, 则 dir_path 下的所有 文件和子目录将会在远端的当前目录. is_only_python_files_accepted (bool) — 是否仅添加目录下的.py/.pyc/.egg文件，默认为False

  注解

  如 resource_path 未提供, 且调用了多次 add_directory, 各个目录下的文件或者子目录如果存在重 名, 则为未定义行为. 有可能在添加时就出错, 也有可能在远端启动出错. 为避免该情况, 可以为每个要添 加的 dir_path 设置惟一的 resource_path.

• add_egg_file(file_path)

  向Pipeline添加一个egg文件，使得该文件会在运行期自动被添加到PYTHONPATH中

  参数:	file_path (str) — egg文件路径

• add_file(source, resource_path=None, executable=False, from_bytes=False)

  向Pipeline添加单个文件，使得该文件能够在运行期被访问

  参数:	source (str) — 需要添加的文件路径或者比特数据流 resource_path (str) — 计算引擎运行时访问该文件的本地路径, 应是相对路径. 也即在远端, source 将会被映射 成该 resource_path 路径, 用户程序可以直接用该路径访问. executable (bool) — 若为True，则该文件在运行期会被添加可执行属性 from_bytes (bool) — 若为True, 则表示远端文件的内容即为 source 的比特数据流, 默认为 False, source 应为一个文件路径

  >> pipeline.add_file('/path/to/local/or/hdfs/file', './target_path_a')
  >> def remote_filter(self):
  ..     # This is a dumb example, don't do this in production,
  ..     import os
  ..     return os.path.isfile('./target_path_a')
  >> pc = pipeline.parallelize([1,2,3])
  >> filtered_pc = pc.filter(remote_filter)
  >> print pc.diff(filtered_pc).count().get()
  .. 0

• async_run()

  立刻运行Pipeline并等待作业提交完成

  Raises:	BigflowRuntimeException — 若运行期出错抛出此异常

• config()

  获得job config

  返回:	用户不应当修改此Pipeline job config
  返回类型:	JobConfig

• default_objector()

  返回该Pipeline的默认序列化/反序列化器

  返回:	序列化/反序列化器，用户不应当修改此objector
  返回类型:	Objector

• get(pvalue)

  将一个P类型表示的数据汇聚为内存变量，相当于调用pvalue.get()。改方法隐式调用pvalue.cache() 并立即触发Pipeline.run()

  参数:	pvalue (PType) — P类型实例
  返回:	内存变量
  返回类型:	object

• id()

  得到表示该Pipeline的唯一ID

  返回:	Pipeline ID
  返回类型:	int

• parallelize(dataset, **options)

  将一段内存变量映射为一个P类型实例

  参数:	dataset (object) — 任意类型的内存变量 options — serde: 设置dataset的serde对象
  返回:	表示该内存变量的P类型
  返回类型:	PType

• plan(cache_id)

• read(source)

  将外部存储的数据映射为一个PCollection，并在运行时读取数据

  参数:	source (Source) — 表示外部存储的Source实例
  返回:	读取结果
  返回类型:	PCollection

• reset_all_counters()

  将所有counter清零

  Raises:	error.BigflowRuntimeException — 此方法不允许在 Bigflow变换 的用户自定义方法(UDF)中调用，否则抛出此异常

• reset_counter(name)

  将一个counter清零, 若 name 中不包含 group 部分, 则默认将 Flume group 下面对应的 counter 清零

  参数:	name (str) — counter名称，其说明请参考 counter模块
  Raises:	error.BigflowRuntimeException — 此方法不允许在 Bigflow变换 的用户自定义方法(UDF)中调用，否则抛出此异常

• run()

  立刻运行Pipeline并等待结束

  Raises:	BigflowRuntimeException — 若运行期出错抛出此异常

• set_fini_hook(name, fn)

  向Pipeline设置一个结束钩子，使得Pipeline能够在进程结束前依次执行 :param name: 钩子名称 :type name: str :param fn: 方法名称 :type fn: callable

• set_init_hook(name, fn)

  向Pipeline设置一个初始化钩子，使得Pipeline能够在进程启动时依次执行(按钩子名字排序) :param name: 钩子名称, 用户应使用 str 作为钩子名字, 同时钩子应使用 ascii 码中字符. :type name: str :param fn: 方法名称 :type fn: callable

• write(pcollection, target)

  将一个PCollection映射为外部存储数据，并在运行期写到该外部存储

  参数:	pcollection (PCollection) — 要写出的PCollection target (Target) — 表示外部存储的Target实例