#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright (c) 2015 Baidu, Inc. All Rights Reserved.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""
Bigflow Python
Author: Wang, Cong(wangcong09@baidu.com), panyunhong(panyunhong@baidu.com)
**(UDF)**
"""
def accumulate(pcollection, zero, accumulate_fn, *side_inputs, **options):
"""
PCollectionPObject
IOzeroaccumulate_fn:
zero: Ozero() => O
accumulate_fn: accumulate_fn(O, I) => O (accumulate_fn)
Args:
pcollection (PCollection): PCollection
zero (value or function):
accumulate_fn (function):
*side_inputs: SideInputs
**options:
Returns:
PObject:
aggregatereduceaggregate/reduce
>>> from bigflow import transforms
>>> _p = _pipeline.parallelize([1, 2, 3])
>>> transforms.accumulate(_p, 0, lambda x, y: x + y).get()
6
TODO: Another example
"""
import bigflow.transform_impls.accumulate
return bigflow.transform_impls.accumulate.accumulate(pcollection,
zero, accumulate_fn, *side_inputs, **options)
def aggregate(pcollection, zero, aggregate_fn, combine_fn, *side_inputs, **options):
"""
PCollectionPObject
IOzeroaggregate_fncombine_fn:
zero: Ozero() => O
aggregate_fn: aggregate_fn(O, I) => O (aggregate_fn)
combine_fn: combine_fn(O, O) => O (combine_fn)
aggregatepcollectionzeroO
aggregate_fnO
Ocombine_fn
Args:
pcollection (PCollection): PCollection
zero (value or callable):
aggregate_fn (callable):
combine_fn (callable):
*side_inputs: SideInputs
**options:
Returns:
PObject:
>>> from bigflow import transforms
>>> _p = _pipeline.parallelize(["viva", "la", "vida"])
>>> transforms.aggregate(_p, 0, lambda x, y: x + len(y), lambda x, y: x + y).get() # sum words length
10
"""
import bigflow.transform_impls.aggregate
return bigflow.transform_impls.aggregate.aggregate(pcollection,
zero,
aggregate_fn,
combine_fn,
*side_inputs, **options)
def cartesian(*pcollections, **options):
"""
PCollectionPCollection
Args:
*pcollections: PCollection
**options:
Returns:
PCollection:
>>> from bigflow import transforms
>>> p1 = _pipeline.parallelize([1, 2, 3])
>>> p2 = _pipeline.parallelize(['a', 'b', 'c'])
>>> transforms.cartesian(p1, p2)
[(1, 'a'), (1, 'b'), (1, 'c'), (2, 'a'), (2, 'b'), (2, 'c'), (3, 'a'), (3, 'b'), (3, 'c')]
>>> p3 = _pipeline.parallelize(1)
>>> p4 = _pipeline.parallelize(2)
>>> p3.cartesian(p4).get()
[(1, 2)]
>>> p5 = _pipeline.parallelize([3, 4])
>>> p3.cartesian(p5).get()
[(1, 3), (1, 4)]
>>> p3.cartesian(p4, p5).get()
[(1, 2, 3), (1, 2, 4)]
"""
import bigflow.transform_impls.cartesian
return bigflow.transform_impls.cartesian.cartesian(*pcollections, **options)
def cogroup(*pcollections, **options):
"""
pcollection
cogroupPCollection(k, v)
cogroupkkeyPCollection
PTable
PTablevaluetupletuplePCollection
nPCollectionnPCollectionkey
PCollectionkeyPCollectionPCollection
group_bykey_extractor
group_by_keycogroup
Args:
*pcollections: PCollection
**options:
Returns:
PTable:
>>> from bigflow import transforms
>>> _p1 = _pipeline.parallelize([("A", 1), ("A", 2), ("B", 3)])
>>> _p2 = _pipeline.parallelize([("A", 4)])
>>> _p = transforms.cogroup(_p1, _p2)
# _p{"A": ([1, 2], [4]), "B": ([3], [])} PTablevaluetuple of PCollectionget
>>> _p.apply_values(lambda x, y: transforms.union(x, y)).get()
{"A": [1, 2, 4], "B": [3]}
>>> def distinct_and_join(p, q): # join
... return p.cogroup(q) \\
... .apply_values(lambda a, b: (a.distinct(), b.distinct())) \\
... .apply_values(transforms.cartesian) \\
... .flatten()
>>> _p1 = _pipeline.parallelize([("A", 1), ("A", 2), ("A", 1), ("C", 1)])
>>> _p2 = _pipeline.parallelize([("A", 3), ("A", 3), ("B", 2)])
>>> print distinct_and_join(_p1, _p2).get()
[("A", (1, 3)), ("A", (2, 3))]
>>> # bigflowp.distinct().join(q.distinct())
>>> def semi_join(p, q): # keyjoin
... return p.cogroup(q) \\
... .apply_values(lambda a, b: (a.take(1), b.take(1))) \\
... .apply_values(transforms.cartesian) \\
... .flatten()
>>> print semi_join(_p1, _p2).get()
[("A", (1, 3))]
"""
import bigflow.transform_impls.cogroup
return bigflow.transform_impls.cogroup.cogroup(*pcollections, **options)
def combine(pcollection, fn, **options):
"""
PCollection
Ofn fn([O...]) => O[]
pcollectionfn
O
Ofn
pre_combine=Falsefn
fnfn
Args:
pcollection (PCollection): PCollection
fn (function):
**options:
pre_combine(bool): True
Returns:
PObject:
>>> _p = _pipeline.parallelize([2, 4, 6, 10])
>>> transforms.combine(_p, sum).get()
22
>>> number_list = self._pipeline.parallelize(['2', '3', '4'])
>>> to_lookup = {'2': 2, '3': 3, '4': 4}
>>> def lookup_dict_and_sum(numbers):
... return sum(map(lambda s: to_lookup[s], numbers))
>>> number_list.combine(lookup_dict_and_sum, pre_combine=False).get()
9
>>> number_list.combine(lookup_dict_and_sum).get()
Error may occur (because of pre_combine, 1)
"""
import bigflow.transform_impls.combine
return bigflow.transform_impls.combine.combine(pcollection, fn, **options)
def count(pcollection, **options):
"""
PCollection
Args:
pcollection (PCollection): PCollection
**options:
Returns:
PObject:
>>> from bigflow import transforms
>>> _p = _pipeline.parallelize([1, 2, 3, 4])
>>> transforms.count(_p).get()
4
"""
import bigflow.transform_impls.count
return bigflow.transform_impls.count.count(pcollection, **options)
def distinct(pcollection, **options):
"""
PCollection
Args:
pcollection (PCollection): PCollection
**options:
Returns:
PCollection: PCollection
>>> from bigflow import transforms
>>> _p = _pipeline.parallelize([2, 2, 1, 9, 3, 3])
>>> transforms.distinct(_p).get()
[2, 3, 1, 9]
"""
import bigflow.transform_impls.distinct
return bigflow.transform_impls.distinct.distinct(pcollection, **options)
def extract_keys(ptable, **options):
"""
PTablekey
Args:
ptable (PTable): PTable
**options:
Returns:
PCollection: keyPCollection
>>> from bigflow import transforms
>>> _p = _pipeline.parallelize({"A": [2, 3], "B": [4, 5]})
>>> transforms.extract_keys(_p).get()
["A", "B"]
"""
import bigflow.transform_impls.extract_keys
return bigflow.transform_impls.extract_keys.extract_keys(ptable, **options)
def extract_values(ptable, **options):
"""
PTablevalue
Args:
ptable (PTable): PTable
**options:
Returns:
PCollection: valuePCollection
>>> from bigflow import transforms
>>> _p = _pipeline.parallelize({"A": [2, 3], "B": [4, 5]})
>>> transforms.extract_values(_p).get()
[2, 3, 4, 5]
PTablevalue
>>> _p = _pipeline.parallelize({"A": {"a": [2, 3], "b": [1, 4]}, "B": {"c": [6, 7], "d": [9, 9]}}
>>> print _p
>>> {k0: {k1: [...]}} # PTable
>>> transforms.extract_values(_p).get()
>>> [2, 3, 1, 4, 6, 7, 9, 9]
"""
import bigflow.transform_impls.extract_values
return bigflow.transform_impls.extract_values.extract_values(ptable, **options)
def flatten(ptable, **options):
"""
PTablekeyvalue(key, value)PCollection
Args:
ptable (PTable): PTable
**options:
Returns:
PCollection: (key, value)PCollection
>>> from bigflow import transforms
>>> _p = _pipeline.parallelize({"A": [2, 3], "B": [4, 5]})
>>> transforms.flatten(_p).get()
[("A", 2), ("A", 3), ("B", 4), ("B", 5)]
"""
import bigflow.transform_impls.flatten
return bigflow.transform_impls.flatten.flatten(ptable, **options)
def flatten_values(ptable, **options):
"""
``extract_values(ptable)``
Args:
ptable (PTable): PTable
**options:
Returns:
PCollection: valuePCollection
>>> from bigflow import transforms
>>> _p = _pipeline.parallelize({"A": [2, 3], "B": [4, 5]})
>>> transforms.flatten_values(_p).get()
[2, 3, 4, 5]
"""
import bigflow.transform_impls.flatten_values
return bigflow.transform_impls.flatten_values.flatten_values(ptable, **options)
def filter(pcollection, fn, *side_inputs, **options):
"""
PCollectionPCollection
Ifn fn(I) => bool
Args:
pcollection (PCollection): PCollection
fn (function):
**options:
Returns:
PCollection:
>>> from bigflow import transforms
>>> _p = _pipeline.parallelize([3, 7, 1, 3, 2, 8])
>>> transforms.filter(_p, lambda x: x % 2 == 0).get()
[2, 8]
"""
import bigflow.transform_impls.filter
return bigflow.transform_impls.filter.filter(pcollection, fn, *side_inputs, **options)
def first(pcollection, **options):
"""
PCollection
Args:
pcollection (PCollection): PCollection
**options:
Returns:
PObject: PObject
>>> from bigflow import transforms
>>> _p = _pipeline.parallelize([3, 7, 1, 3, 2, 8])
>>> transforms.first(_p).get()
3
"""
import bigflow.transform_impls.first
return bigflow.transform_impls.first.first(pcollection, **options)
def group_by(pcollection, key_extractor, value_extractor=None, **options):
"""
key_extractorvalue_extractorPCollectionPTable
Args:
pcollection (PCollection): PCollection
key_extractor (function): key
value_extractor (function, optional): value
**options:
Returns:
PTable:
>>> _p = _pipeline.parallelize([("A", 4), ("A", 3), ("B", 2), ("A", 1)])
>>> transforms.group_by(_p, lambda x: x[0], lambda x: x[1]).get()
{"A": [4, 3, 1], "B": [2]}
"""
import bigflow.transform_impls.group_by
return bigflow.transform_impls.group_by.group_by(pcollection,
key_extractor, value_extractor, **options)
def group_by_key(pcollection, **options):
"""
PCollectionkey/valuePCollection
PTable
Args:
pcollection (PCollection): PCollection
**options:
Returns:
PTable:
>>> _p = _pipeline.parallelize([("A", 4), ("A", 3), ("B", 2), ("A", 1)])
>>> transforms.group_by_key(_p).get()
{"A": [4, 3, 1], "B": [2]}
"""
import bigflow.transform_impls.functions
from bigflow import serde
pipeline = pcollection.pipeline()
key_serde = options.get('key_serde', None)
value_serde = options.get('value_serde', None)
if key_serde is None:
key_serde = serde._key_serde(pcollection.serde(), pipeline.default_objector())
if value_serde is None:
value_serde = serde._value_serde(pcollection.serde(), pipeline.default_objector())
options['key_serde'] = key_serde
options['value_serde'] = value_serde
# if key_extractor is None, will use the default key extractor
return group_by(pcollection,
key_extractor=None,
value_extractor=bigflow.core.entity.ExtractValueFn(),
**options
)
def window_into(pcollection, window, **options):
"""
windowPCollectionPTable
Args:
pcollection (PCollection): PCollection
window(Window): Window
Returns:
PTable:
"""
import bigflow.transform_impls.window_into
return bigflow.transform_impls.window_into.window_into(pcollection, window, **options)
def foreach(pvalue, fn, *side_inputs, **options):
"""
PCollection/PObject
foreach
Ifn fn(I) => object()
Args:
pvalue (PCollection or PObject): P
fn (function):
*side_inputs: SideInputs
**options:
Results:
None
>>> from bigflow import lazy_var
>>> r = lazy_var.declare(lambda: redis.Redis(host='x.x.x.x', port=x, db=0))
>>> x = _pipeline.parallelize([("a", "1"), ("b", "4")])
>>> x.foreach(lambda (k, v): r.get().set(k, v))
>>> _pipeline.run() # all the data will be written into redis
"""
import bigflow.transform_impls.foreach
return bigflow.transform_impls.foreach.foreach(pvalue, fn, *side_inputs, **options)
def join(*pcollections, **options):
"""
PCollectionkeyPCollection
(key, (value1, value2, ..., valuen))
Args:
*pcollections: PCollection
**options:
Returns:
PCollection:
>>> x = _pipeline.parallelize([("a", 1), ("b", 4)])
>>> y = _pipeline.parallelize([("a", 2), ("a", 3)])
>>> transforms.join(x, y).get()
[("a", (1, 2)), ("a", (1, 3))]
"""
import bigflow.transform_impls.join
return bigflow.transform_impls.join.join(*pcollections, **options)
def left_join(*pcollections, **options):
"""
PCollectionkeyPCollection
(key, (value 1, value 2, ..., value n))mPCollection
value mNone
Args:
*pcollections: PCollection
**options:
Returns:
PCollection:
>>> x = _pipeline.parallelize([("a", 1), ("b", 4)])
>>> y = _pipeline.parallelize([("a", 2)])
>>> transforms.left_join(x, y).get()
[("a", (1, 2)), ("b", (4, None))]
"""
import bigflow.transform_impls.join
return bigflow.transform_impls.join.left_join(*pcollections, **options)
def right_join(*pcollections, **options):
"""
PCollectionkeyPCollection
(key, (value 1, value 2, ..., value n))mPCollection
value mNone
Args:
*pcollections: PCollection
**options:
Returns:
PCollection:
>>> x = _pipeline.parallelize([("a", 1), ("b", 4)])
>>> y = _pipeline.parallelize([("a", 2), ("a", 3)])
>>> transforms.right_join(x, y).get()
[("a", (1, 2)), ("a", (1, 3))]
"""
import bigflow.transform_impls.join
return bigflow.transform_impls.join.right_join(*pcollections, **options)
def full_join(*pcollections, **options):
"""
PCollectionkeyPCollection
(key, (value 1, value 2, ..., value n))mPCollection
value mNone
Args:
*pcollections: PCollection
**options:
Returns:
PCollection:
>>> x = _pipeline.parallelize([("a", 1)])
>>> y = _pipeline.parallelize([("b", 2)])
>>> transforms.full_join(x, y).get()
[("a", (1, None)), ("b", (None, 2))]
"""
import bigflow.transform_impls.join
return bigflow.transform_impls.join.full_join(*pcollections, **options)
def map(pvalue, fn, *side_inputs, **options):
"""
PCollection
PCollection/PObject
PCollection/PObject
Ifn fn(I) => O
Args:
pvalue (PCollection or PObject): P
fn (function):
*side_inputs: SideInputs
**options:
Results:
PType: PCollection/PObject
>>> from bigflow import transforms
>>> _p = _pipeline.parallelize([1, 3, 5, 7])
>>> transforms.map(_p, lambda x: x + 1).get()
[2, 4, 6, 8]
>>> transforms.map(_p, lambda x: [x, x * 2]).get()
[[1, 2], [3, 6], [5, 10], [7, 14]]
"""
import bigflow.transform_impls.map
return bigflow.transform_impls.map.map(pvalue, fn, *side_inputs, **options)
def flat_map(pvalue, fn, *side_inputs, **options):
"""
PCollectionN
(__iter__())
PCollection
Ifn fn(I) => [O...][]
Args:
pvalue (PCollection or PObject): P
fn (function):
*side_inputs: SideInputs
**options:
Results:
PCollection: PCollection
>>> from bigflow import transforms
>>> _p = _pipeline.parallelize([1, 3, 5, 7])
>>> transforms.flat_map(_p, lambda x: [x, x * 2]).get()
[1, 2, 3, 5, 6, 7, 10, 14]
>>> transforms.flat_map(_p, lambda x: [[x, x * 2]]).get()
[[1, 2], [3, 6], [5, 10], [7, 14]]
>>> transforms.flat_map(_p, lambda x: (x, x * 2)).get()
[1, 2, 3, 6, 5, 10, 7, 14]
>>> transforms.flat_map(_p, lambda x: [(x, x * 2)]).get()
[(1, 2), (3, 6), (5, 10), (7, 14)]
:
>>> transforms.flat_map(_p, lambda x: [])
[]
None
list
list
pythonyieldgenerator
>>> _p = _pipeline.parallelize([3, 5])
>>> def make_partial_sum(x):
... sum = 0
... for i in xrange(1, x + 1):
... sum += i
... yield sum
>>> transforms.flat_map(_p, make_partial_sum)
[1, 3, 6, 1, 3, 6, 10, 15]
list
"""
import bigflow.transform_impls.flat_map
return bigflow.transform_impls.flat_map.flat_map(pvalue, fn, *side_inputs, **options)
def max(pcollection, key=None, **options):
"""
PCollection
Args:
pcollection (PCollection): PCollection
key (function, optional): keyPython``max()`` ``key``
**options:
Returns:
PObject: PObject
>>> from bigflow import transforms
>>> _p = _pipeline.parallelize([3, 7, 1, 3, 2, 8])
>>> transforms.max(_p).get()
8
>>> transforms.max(_p, lambda val: -val).get()
1
"""
import bigflow.transform_impls.max
return bigflow.transform_impls.max.max(pcollection, key, **options)
def max_elements(pcollection, n, key=None, **options):
"""
PCollectionn
Args:
pcollection (PCollection): PCollection
n (int/PObject): 0
key (function, optional): keyPython``max()`` ``key``
**options:
Returns:
PCollection: nPCollectionn=1PCollection
PObject
>>> from bigflow import transforms
>>> _p = _pipeline.parallelize([3, 7, 1, 3, 2, 8])
>>> transforms.max_elements(_p, 2).get()
[8, 7]
>>> n = _p.count().map(lambda x: x - 1) # n is PObject(5)
>>> transforms.max_elements(_p, n, lambda val: -val).get() # key is -val
>>> [1, 3, 3, 2, 7]
"""
import bigflow.transform_impls.max_elements
return bigflow.transform_impls.max_elements.max_elements(pcollection, n, key, **options)
def min(pcollection, key=None, **options):
"""
PCollection
Args:
pcollection (PCollection): PCollection
key (function, optional): keyPython``min()`` ``key``
**options:
Returns:
PObject: PObject
>>> from bigflow import transforms
>>> _p = _pipeline.parallelize([3, 7, 1, 3, 2, 8])
>>> transforms.min(_p).get()
1
>>> transforms.min(_p, lambda val: -val).get()
8
"""
import bigflow.transform_impls.min
return bigflow.transform_impls.min.min(pcollection, key, **options)
def min_elements(pcollection, n, key=None, **options):
"""
PCollectionn
Args:
pcollection (PCollection): PCollection
n (int/PObject): 0
key (function, optional): keyPython``max()`` ``key``
**options:
Returns:
PCollection: nPCollectionn=1PCollection
PObject
>>> from bigflow import transforms
>>> _p = _pipeline.parallelize([3, 7, 1, 3, 2, 8])
>>> transforms.min_elements(_p, 2).get()
[1, 2]
>>> n = _p.count().map(lambda x: x - 1) # n is PObject(5)
>>> transforms.min_elements(_p, n, lambda val: -val).get() # key is -val
>>> [8, 3, 3, 2, 7]
"""
import bigflow.transform_impls.min_elements
return bigflow.transform_impls.min_elements.min_elements(pcollection, n, key, **options)
def reduce(pcollection, fn, *side_inputs, **options):
"""
PCollectionfn
Ifn fn(I1, I2) => I (fn)
Args:
pcollection (PCollection): PCollection
fn (function):
*side_inputs: SideInputs
**options:
Returns:
PObject:
>>> from bigflow import transforms
>>> _p = _pipeline.parallelize([1, 2, 3, 4])
>>> transforms.reduce(_p, lambda x, y: x + y).get()
10
"""
import bigflow.transform_impls.reduce
return bigflow.transform_impls.reduce.reduce(pcollection, fn, *side_inputs, **options)
def sort(pcollection, reverse=False):
"""
PCollection
Args:
pcollection (PCollection): PCollection
reverse (bool): True
Returns:
PCollection:
>>> from bigflow import transforms
>>> _p = _pipeline.parallelize([3, 1, 2, 8])
>>> transforms.sort(_p).get()
[1, 2, 3, 8]
"""
import bigflow.transform_impls.sort
return bigflow.transform_impls.sort.sort(pcollection, reverse)
def sort_by(pcollection, key, reverse=False):
"""
PCollectionkey
Args:
pcollection (PCollection): PCollection
key (function, optional): keyPython``sort()`` ``key`` keyNonekeykey
reverse (bool): True
Returns:
PCollection:
>>> from bigflow import transforms
>>> from bigflow.transform_impls import sort
>>> _p = _pipeline.parallelize([3, 1, 2, 8])
>>> p2 = _pipeline.parallelize([(1, 2), (3, 4), (3, 5), (2, 6), (2, 4)]
>>> transforms.sort_by(_p).get()
>>> transforms.sort_by(p2, lambda rec:[sort.ASC(rec[0]), sort.DESC(rec[1])]).get()
[1, 2, 3, 8]
[(1, 2), (2, 6), (2, 4), (3, 5), (3, 4)]
Note:
1. sort12.0
key
2. sort/sort_by
* accumulate,transform
* aggregate(O+I=>O)
* first take ()
3. sortwrite
http://bigflow.baidu.com/doc/rst/bigflow.output.html
"""
import bigflow.transform_impls.sort
return bigflow.transform_impls.sort.sort_by(pcollection, key, reverse)
def subtract(pcollection1, pcollection2):
"""
PCollection1PCollection2PCollection1PCollection2
Args:
pcollection1 (PCollection): PCollection
pcollection2 (PCollection): PCollection
Returns:
PCollection: PCollection
>>> from bigflow import transforms
>>> a = _pipeline.parallelize([1, 2, 3, 4])
>>> b = _pipeline.parallelize([1, 2, 5])
>>> transforms.subtract(a, b).get()
[3, 4]
"""
import bigflow.transform_impls.subtract
return bigflow.transform_impls.subtract.subtract(pcollection1, pcollection2)
def substract(pcollection1, pcollection2):
"""
subtract
"""
import bigflow.transform_impls.subtract
return bigflow.transform_impls.subtract.subtract(pcollection1, pcollection2)
def sum(pcollection, **options):
"""
PCollection
Args:
pcollection (PCollection): PCollection
**options:
Returns:
PObject: PObject
>>> from bigflow import transforms
>>> _p = _pipeline.parallelize([1, 2, 3, 4])
>>> transforms.sum(_p).get()
10
"""
import bigflow.transform_impls.sum
return bigflow.transform_impls.sum.sum(pcollection, **options)
def take(pcollection, n, **options):
"""
PCollectionn
npcollection
Args:
pcollection (PCollection): PCollection
n (int or PObject):
**options:
Returns:
PCollection: PCollection
>>> from bigflow import transforms
>>> _p = _pipeline.parallelize([1, 2, 3, 4])
>>> transforms.take(_p, 3).get()
[1, 2, 3]
>>> _n = _pipeline.parallelize(2)
>>> transforms.take(_p, _n).get()
[1, 2]
>>> _n = _pipeline.parallelize([1, 2, 3, 4])
>>> transforms.take(_p, 10).get()
[1, 2, 3, 4]
"""
import bigflow.transform_impls.take
return bigflow.transform_impls.take.take(pcollection, n, **options)
def transform(pcollection, first_arg, *other_args, **options):
"""
PCollectionPCollection
transform
`transform(pcollection, initializer, transformer, finalizer, *side_inputs, **options)`
transformPCollection3:
initializer, transformerfinalizer
status(context)emitterPCollection
IOinitializer, transformer, finalizer:
initializer(emitter, *side_inputs) => status(object)
transformer(status, emitter, I, *side_inputs) => status(object)
(transformer)
finalizer(status, emitter, *side_inputs) => None (finalizer)
emitter.emit(O)
Args:
pcollection (PCollection): PCollection
initializer (callable):
transformer (callable):
finalizer (callable):
*side_inputs: SideInputs
**options:
Returns:
PCollection: PCollection
>>> from bigflow import transforms
>>> import copy
>>> def initializer(emitter):
>>> return []
>>>
>>> def transformer(status, emitter, inp):
>>> status.append(copy.deepcopy(inp)) #
>>> return status
>>>
>>> def finalizer(status, emitter):
>>> emitter.emit(status)
>>>
>>> _p = _pipeline.parallelize([1, 2, 3])
>>> _plist = transforms.transform(_p, initializer, transformer, finalizer)
>>> print _plist.count().get() # [1, 2, 3]
1
>>> print _plist.get()
[[1, 2, 3]]
`transform(pcollection, transformer, *side_inputs, **options)`
transformer
:class:`bigflow.base.Transformer `
Transformer.begin_process
Transformer.process
Transformer.end_process
PCollection
[]None
>>> class SumTransformer(base.Transformer):
...
... def begin_process(self):
... self._sum = 0
... return []
...
... def process(self, record):
... self._sum += record
... return []
...
... def end_process(self):
... yield self._sum
...
>>> p1 = _pipeline.parallelize([1, 2, 3])
>>> transforms.transform(p1, SumTransformer).get()
6
>>> class PartialSumTransformer(base.Transformer):
...
... def begin_process(self):
... self._sum = 0
... return []
...
... def process(self, record):
... self._sum += record
... yield self._sum
...
>>> transforms.transform(p1, PartialSumTransformer()),get()
[1, 3, 6]
>>> class ZipTransformer(base.Transformer):
...
... def begin_process(self, *si):
... self.index = 0
... lens = map(len, si)
... self.min_len = min(lens)
... return []
...
... def process(self, inp, *si):
... if self.index < self.min_len:
... yield (inp, ) + tuple(map(lambda x: x[self.index], si))
... self.index += 1
...
>>> p2 = _pipeline.parallelize([4, 5]).sort()
>>> transforms.transform(p1, ZipTransformer(), p2).get()
[(1, 4), (2, 5)]
BigflowPCollection
(aggregate)
"""
import bigflow.transform_impls.transform
return bigflow.transform_impls.transform.transform(pcollection, first_arg, *other_args,
**options)
def union(*pvalues, **options):
"""
PCollection/PObjectPCollection
PCollection
Args:
*pvalues: PCollection/PObject
Returns:
PCollection: PCollection
>>> from bigflow import transforms
>>> _p1 = _pipeline.parallelize([1, 2, 3, 4])
>>> _p2 = _pipeline.parallelize([5, 6, 7, 8])
>>> transforms.union(_p1, _p2).get()
[1, 2, 3, 4, 5, 6, 7, 8]
"""
import bigflow.transform_impls.union
return bigflow.transform_impls.union.union(*pvalues, **options)
def is_empty(pcollection):
"""
PCollection
Args:
pcollection (PCollection): PCollection
Returns:
PObject: PObject
>>> from bigflow import transforms
>>> a = _pipeline.parallelize([1, 2, 3, 4])
>>> transforms.is_empty(a).get()
False
>>> b = _pipeline.parallelize([])
>>> transforms.is_empty(b).get()
True
"""
import bigflow.transform_impls.is_empty
return bigflow.transform_impls.is_empty.is_empty(pcollection)
def intersection(pcollection1, pcollection2, output_duplicated = False):
"""
PCollection1PCollection2PCollection1PCollection2
Args:
pcollection1 (PCollection): 1
pcollection2 (PCollection): 2
Returns:
PCollection:
>>> from bigflow import transforms
>>> a = _pipeline.parallelize([1, 1, 2, 3])
>>> b = _pipeline.parallelize([1, 1, 2, 2, 5])
>>> transforms.intersection(a, b).get()
[1, 2]
>>> transforms.intersection(a, b, output_duplicated = True).get()
[1, 1, 2]
"""
import bigflow.transform_impls.intersection
return bigflow.transform_impls.intersection.intersection(
pcollection1,
pcollection2,
output_duplicated = output_duplicated
)
def diff(pcollection1, pcollection2):
"""
PCollection1PCollection2
Args:
pcollection1 (PCollection): 1
pcollection2 (PCollection): 2
Returns:
PCollection: PCollection
>>> from bigflow import transforms
>>> a = _pipeline.parallelize([1, 1, 2, 3])
>>> b = _pipeline.parallelize([1, 1, 2, 2])
>>> transforms.diff(a, b).get()
[(2, (1, 2)), (3, (1, 0))]
"""
import bigflow.transform_impls.diff
return bigflow.transform_impls.diff.diff(pcollection1, pcollection2)
def to_list_pobject(pvalue, **options):
"""
PCollectionPObjectPObjectlist
Args:
pvalue (PCollection):
**options:
Returns:
PObject: list
>>> from bigflow import transforms
>>> a = _pipeline.parallelize([1, 1, 2, 3])
>>> transforms.to_list_pobject(a).get()
[1, 1, 2, 3]
>>> type(a)
bigflow.pcollection.PCollection
>>> a.map(lambda x: str(type(x))).get()
["", "", "", ""]
>>> b = transforms.to_list_pobject(a)
>>> type(b)
bigflow.pobject.PObject
>>> b.map(lambda x: str(type(x))).get()
""
Note:
transform
PCollectionlistPObject
maplist
Bigflow
apply_values
list
Out-Of-Memory
1. :func:`bigflow.transforms.aggregate `
2. :func:`bigflow.transforms.transform `
"""
import bigflow.transform_impls.to_list_pobject
return bigflow.transform_impls.to_list_pobject.to_list_pobject(pvalue, **options)
def pipe(pvalue, command, **options):
"""
PCollection/PTablecommandPCollection
Args:
pvalue (PCollection/PTable):
command:
**options:
type: pipestreamingbistreamingstreaming
buffer_size: 64MB
input_fields_num: commandfield1
PTablepipePCollectionpipefieldPCollectiontuple
output_fields_num: commandfield1
field_delimiter: streamingfieldtab
Returns:
PCollection: PCollection
Note:
1. pipePCollectionpipe
2. pipePTablepipePTableKey
Key
>>> from bigflow import transforms
>>> p = _pipeline.parallelize({
>>> 'key1_a': {
>>> 'key2_a': ['value1', 'value2'],
>>> 'key2_b': ['value3', 'value4']
>>> },
>>> 'key2_b': {
>>> 'key2_c': ['value5', 'value6']
>>> }
>>> })
>>> transforms.pipe(p, 'cat').get()
catcolumntab
key1_a\tkey2_a\tvalue1
key1_a\tkey2_a\tvalue2
key1_a\tkey2_b\tvalue3
key1_a\tkey2_b\tvalue4
key1_b\tkey2_c\tvalue5
key1_b\tkey2_c\tvalue6
3. PTableapply_valuespipeapply
Key
>>> from bigflow import transforms
>>> p = _pipeline.parallelize([1, 1, 2, 3])
>>> transforms.pipe(p, 'cat').get()
['1', '1', '2', '3']
>>> from bigflow import transforms
>>> p = _pipeline.parallelize({'A': [1, 2], 'B': [2, 3]})
>>> transforms.pipe(p, 'cat').get()
['A\t1', 'A\t2', 'B\t2', 'B\t3']
>>> from bigflow import transforms
>>> p = _pipeline.parallelize([(1, 'a'), (1, 'a'), (2, 'b'), (3, 'c')])
>>> transforms.pipe(p, 'cat', type='bistreaming', input_fields_num=2, output_fields_num=2).get()
[('1', 'a'), ('1', 'a'), ('2', 'b'), ('3', 'c')]
"""
import bigflow.transform_impls.pipe
return bigflow.transform_impls.pipe.pipe(pvalue, command, **options)
def idl_to_str(pcollection, **options):
"""
PCollectionidlidl packetHeartbeatEOF
Args:
pcollection (PCollection):
**options:
log_type: idllog_textlog_binlog_text
Returns:
PCollection: PCollection
"""
from bigflow import serde
from bigflow.core import entity
log_type = options.get("log_type", "log_text")
idl_serde = serde.IdlPacketSerde(log_type=log_type)
deserialize = entity.SerdeWrapper(idl_serde, is_serialize=False)
return pcollection.map(deserialize, serde=serde.StrSerde()) \
.filter(lambda x: x is not None)
def str_to_idl(pcollection, **options):
"""
PCollectionidlstr
Args:
pcollection (PCollection):
**options:
log_type: idllog_textlog_binlog_text
Returns:
PCollection: PCollection
"""
from bigflow import serde
from bigflow.core import entity
log_type = options.get("log_type", "log_text")
idl_serde = serde.IdlPacketSerde(log_type=log_type)
serialize = entity.SerdeWrapper(idl_serde, is_serialize=True)
return pcollection.map(serialize, serde=serde.StrSerde())
def to_pobject(pvalue, **options):
"""
PCollection/PTablePObjectPObjectlist/dict
Args:
pvalue (PCollection/PTable):
**options:
Returns:
PObject: list/dict
>>> from bigflow import transforms
>>> a = _pipeline.parallelize([1, 1, 2, 3])
>>> transforms.to_pobject(a).get()
[1, 1, 2, 3]
>>> from bigflow import transforms
>>> b = _pipeline.parallelize({'e': 'f', 'g': 'h'})
>>> transforms.to_pobject(b).get()
{'e': 'f', 'g': 'h'}
"""
from bigflow import pobject
from bigflow import pcollection
from bigflow import ptable
def _to_pobject(item):
if isinstance(item, pobject.PObject):
return item
elif isinstance(item, pcollection.PCollection):
return to_list_pobject(item)
elif isinstance(item, ptable.PTable):
return to_list_pobject(item.apply_values(lambda x: _to_pobject(x)) \
.flatten()) \
.map(lambda x: dict(x))
else:
raise ValueError("to_pobject only applied on PCollections/PObject/PTable")
return _to_pobject(pvalue)
def make_tuple(*pobjects, **options):
"""
PObjectPObject(tuple)
PObject
:func:`bigflow.transforms.cartesian(self, *pvalues, **options)
`
Args: *pobjects (PObject) PObjectsPObject.
Returns:
PObject: PObject(tuple), tuplennPObject
>>> p1 = _pipeline.parallelize(1)
>>> p2 = _pipeline.parallelize(2)
>>> transforms.make_tuple(p1, p2).get()
(1, 2)
>>> p3 = _pipeline.parallelize([3, 4])
>>> transforms.make_tuple(p1, p3).get()
!!! AssertionError
>>> transforms.make_tuple(p1, p1, p2).get()
(1, 1, 2)
"""
from bigflow.transform_impls import make_tuple
return make_tuple.make_tuple(*pobjects, **options)