dampr module
Dampr is a light-weight MapReduce library for single machine computation. It supports a number of useful features such as map and reduce side joins, associative reduces, aggregations, multiprocessing, and more. It operates by default out-of-core, allowing processing of 100s of gbs to tbs of data on a single machine.
While the underlying engine uses the MapReduce paradigm for processing, Dampr is best utilized via it's DSL which provides higher level functionality for complex workflows.
It loosely attempts to replicate interfaces such as Spark or Scalding for ease of pickup.
""" Dampr is a light-weight MapReduce library for single machine computation. It supports a number of useful features such as map and reduce side joins, associative reduces, aggregations, multiprocessing, and more. It operates by default out-of-core, allowing processing of 100s of gbs to tbs of data on a single machine. While the underlying engine uses the MapReduce paradigm for processing, Dampr is best utilized via it's DSL which provides higher level functionality for complex workflows. It loosely attempts to replicate interfaces such as Spark or Scalding for ease of pickup. """ import logging from .dampr import Dampr, PMap, PReduce, PJoin, ARReduce from .base import BlockMapper, BlockReducer from .dataset import Dataset __all__ = ["Dampr", "PMap", "PReduce", "PJoin", "ARReduce", "BlockMapper", "BlockReducer"] def setup_logging(debug=False): """ Convenience function for enabling logging """ loglevel = logging.DEBUG if debug else logging.INFO logging.basicConfig(level=loglevel, format='%(asctime)s %(levelname)s %(message)s')
Classes
class ARReduce
Associative Reducer operators.
class ARReduce(object): """ Associative Reducer operators. """ def __init__(self, pmap): self.pmap = pmap def reduce(self, binop, reduce_buffer=1000, **options): """ Reduces a grouped dataset by an associative binary operator. This will do a partial reduce in the map stage, completing the reduction in the reduce stage. It is often substantially faster than the more general group_by. `reduce_buffer` is a constant which tells Dampr how much temporary storage to keep in memory on the map side reductions. For example with `1000`, Dampr will keep 1000 unique keys in memory. In the case where a new key would spill over the buffer size, Dampr will flush the buffer to disk and create a new buffer. By increasing the `reduce_buffer`, you can increase efficiency while sacrificing memory. >>> Dampr.memory([1,2,3,4,5]).a_group_by(lambda x: 1).reduce(lambda x, y: x + y).read() [(1, 15)] """ def _reduce(key, vs): acc = next(vs) for v in vs: acc = binop(acc, v) return acc red = Reduce(_reduce) options.update({"binop": binop, "reduce_buffer": reduce_buffer}) # We add the associative aggregator to the combiner during map pm = self.pmap.checkpoint(True, combiner=PartialReduceCombiner(red), options=options) return PReduce(pm.source, pm.pmer).reduce(_reduce) def first(self, **options): """ Returns the first item found for a given key. >>> Dampr.memory([1,2,3,4,5]).a_group_by(lambda x: x % 2).first().read() [(0, 2), (1, 1)] """ return self.reduce(lambda x, _y: x, **options) def sum(self, **options): """ Simple sum of values by key. >>> Dampr.memory([1,2,3,4,5]).a_group_by(lambda x: x % 2).sum().read() [(0, 6), (1, 9)] """ return self.reduce(lambda x, y: x + y, **options)
Ancestors (in MRO)
- ARReduce
- __builtin__.object
Instance variables
var pmap
Methods
def __init__(
self, pmap)
def __init__(self, pmap): self.pmap = pmap
def first(
self, **options)
Returns the first item found for a given key.
Dampr.memory([1,2,3,4,5]).a_group_by(lambda x: x % 2).first().read() [(0, 2), (1, 1)]
def first(self, **options): """ Returns the first item found for a given key. >>> Dampr.memory([1,2,3,4,5]).a_group_by(lambda x: x % 2).first().read() [(0, 2), (1, 1)] """ return self.reduce(lambda x, _y: x, **options)
def reduce(
self, binop, reduce_buffer=1000, **options)
Reduces a grouped dataset by an associative binary operator. This will do a partial reduce in the map stage, completing the reduction in the reduce stage. It is often substantially faster than the more general group_by.
reduce_buffer
is a constant which tells Dampr how much temporary storage
to keep in memory on the map side reductions. For example with 1000
, Dampr
will keep 1000 unique keys in memory. In the case where a new key would spill
over the buffer size, Dampr will flush the buffer to disk and create a new
buffer. By increasing the reduce_buffer
, you can increase efficiency while
sacrificing memory.
>>> Dampr.memory([1,2,3,4,5]).a_group_by(lambda x: 1).reduce(lambda x, y: x + y).read() [(1, 15)]
def reduce(self, binop, reduce_buffer=1000, **options): """ Reduces a grouped dataset by an associative binary operator. This will do a partial reduce in the map stage, completing the reduction in the reduce stage. It is often substantially faster than the more general group_by. `reduce_buffer` is a constant which tells Dampr how much temporary storage to keep in memory on the map side reductions. For example with `1000`, Dampr will keep 1000 unique keys in memory. In the case where a new key would spill over the buffer size, Dampr will flush the buffer to disk and create a new buffer. By increasing the `reduce_buffer`, you can increase efficiency while sacrificing memory. >>> Dampr.memory([1,2,3,4,5]).a_group_by(lambda x: 1).reduce(lambda x, y: x + y).read() [(1, 15)] """ def _reduce(key, vs): acc = next(vs) for v in vs: acc = binop(acc, v) return acc red = Reduce(_reduce) options.update({"binop": binop, "reduce_buffer": reduce_buffer}) # We add the associative aggregator to the combiner during map pm = self.pmap.checkpoint(True, combiner=PartialReduceCombiner(red), options=options) return PReduce(pm.source, pm.pmer).reduce(_reduce)
def sum(
self, **options)
Simple sum of values by key.
Dampr.memory([1,2,3,4,5]).a_group_by(lambda x: x % 2).sum().read() [(0, 6), (1, 9)]
def sum(self, **options): """ Simple sum of values by key. >>> Dampr.memory([1,2,3,4,5]).a_group_by(lambda x: x % 2).sum().read() [(0, 6), (1, 9)] """ return self.reduce(lambda x, y: x + y, **options)
class BlockMapper
Custom BlockMapper. User's can specify how a Mapper instance can consume a stream of key-values, allowing for custom logic.
class BlockMapper(Mapper, Streamable): """ Custom BlockMapper. User's can specify how a Mapper instance can consume a stream of key-values, allowing for custom logic. """ def start(self): """ Sets up instance variables for when a Mapper begins consumption of a Map Block. """ pass def add(self, key, value): """ Logic for how to handle new key-value pairs. This function is required to return an iterator regardless of whether it yields data: this gives a more flexible definition. """ raise NotImplementedError() def finish(self): """ Mapping is finished. In the case of aggregations, this should yield out all remaining key-values to consume. """ return () def map(self, *datasets): assert len(datasets) == 1 return self.stream(datasets[0].read()) def stream(self, kvs): self.start() for key, value in kvs: for out in self.add(key, value): yield out for out in self.finish(): yield out
Ancestors (in MRO)
- BlockMapper
- dampr.base.Mapper
- dampr.base.Streamable
- __builtin__.object
Methods
def add(
self, key, value)
Logic for how to handle new key-value pairs. This function is required to return an iterator regardless of whether it yields data: this gives a more flexible definition.
def add(self, key, value): """ Logic for how to handle new key-value pairs. This function is required to return an iterator regardless of whether it yields data: this gives a more flexible definition. """ raise NotImplementedError()
def finish(
self)
Mapping is finished. In the case of aggregations, this should yield out all remaining key-values to consume.
def finish(self): """ Mapping is finished. In the case of aggregations, this should yield out all remaining key-values to consume. """ return ()
def map(
self, *datasets)
def map(self, *datasets): assert len(datasets) == 1 return self.stream(datasets[0].read())
def start(
self)
Sets up instance variables for when a Mapper begins consumption of a Map Block.
def start(self): """ Sets up instance variables for when a Mapper begins consumption of a Map Block. """ pass
def stream(
self, kvs)
def stream(self, kvs): self.start() for key, value in kvs: for out in self.add(key, value): yield out for out in self.finish(): yield out
class BlockReducer
Custom BlockReducer. User's can specify how a Reducer instance can consume a stream of key-valueiters, allowing for custom logic.
class BlockReducer(Reducer): """ Custom BlockReducer. User's can specify how a Reducer instance can consume a stream of key-valueiters, allowing for custom logic. """ def start(self): pass def add(self, k, it): raise NotImplementedError() def finish(self): return () def reduce(self, *datasets): assert len(datasets) == 1 self.start() for k, vs in self.yield_groups(datasets[0]): for nk, nv in self.add(k, vs): yield nk, nv for nk, nv in self.finish(): yield nk, nv
Ancestors (in MRO)
- BlockReducer
- dampr.base.Reducer
- __builtin__.object
Methods
def add(
self, k, it)
def add(self, k, it): raise NotImplementedError()
def finish(
self)
def finish(self): return ()
def group_datasets(
self, dataset)
def group_datasets(self, dataset): if len(dataset) > 1: dataset = MergeDataset(dataset) elif len(dataset) == 1: dataset = dataset[0] else: dataset = EmptyDataset() return dataset
def reduce(
self, *datasets)
def reduce(self, *datasets): assert len(datasets) == 1 self.start() for k, vs in self.yield_groups(datasets[0]): for nk, nv in self.add(k, vs): yield nk, nv for nk, nv in self.finish(): yield nk, nv
def start(
self)
def start(self): pass
def yield_groups(
self, dataset)
def yield_groups(self, dataset): return self.group_datasets(dataset).grouped_read()
class Dampr
Entrypoint into the Dampr processing functions.
class Dampr(object): """ Entrypoint into the Dampr processing functions. """ def __init__(self, graph=None, runner=None): if graph is None: graph = Graph() self.graph = graph if runner is None: runner = MTRunner self.runner = runner @classmethod def memory(cls, items, partitions=50): """ Create an in-memory dataset from the provided items. `partitions` define how many initial functions there will be. >>> Dampr.memory([1,2,3,4,5]) """ mi = MemoryInput(list(enumerate(items)), partitions) source, ng = Graph().add_input(mi) return PMap(source, Dampr(ng)) @classmethod def read_input(cls, *datasets): """ Reads from the provided datasets. When provided multiple datasets, it treats each as a separate partition. `read_input` can also take Chunker, which lazily returns a set of Datasets to operate over. >>> from dampr.inputs import MemoryInput >>> Dampr.read_input(MemoryInput(enumerate(range(5,10)), 2)) """ if len(datasets) == 1 and isinstance(datasets, Chunker): datasets = datasets[0] elif len(datasets) > 1: datasets = CatDataset(datasets) else: datasets = datasets[0] source, ng = Graph().add_input(datasets) return PMap(source, Dampr(ng)) @classmethod def text(cls, fname, chunk_size=16*1024**2): """ Reads a file or directory of files into Dampr. Each record is assumed to be newline delimited. When fname is a directory, it will walk the directory, collecting all files within it as part of the collection. `text` accepts globs. `chunk_size` describes how big each map portion will be. Returns a PMap object. >>> Dampr.text('/tmp', chunk_size=64*1024**2) """ return cls.read_input(PathInput(fname, chunk_size)) @classmethod def json(cls, *args, **kwargs): """ Convenience function which reads newline-delimited json records. """ return cls.text(*args, **kwargs).map(json.loads) @classmethod def from_dataset(cls, dataset): """ Typically not used, this will read the raw outputs of a Dampr stage as an input. """ assert isinstance(dataset, Chunker) source, ng = Graph().add_input(dataset) return PMap(source, Dampr(ng)) @classmethod def run(self, *pmers, **kwargs): """ Runs a graph or set of graphs. >>> foo = Dampr.memory([1,2,3,4,5]) >>> bar = Dampr.memory([6,7,8,9,10]) >>> left, right = Dampr.run(foo, bar) >>> left.read() [1, 2, 3, 4, 5] >>> right.read() [6, 7, 8, 9, 10] """ sources = [] graph = None for i, pmer in enumerate(pmers): if isinstance(pmer, PMap): pmer = pmer.checkpoint() elif isinstance(pmer, PJoin): pmer = pmer.reduce(lambda l, r: (list(l), list(r))) if i == 0: graph = pmer.pmer.graph else: graph = pmer.pmer.graph.union(graph) sources.append(pmer.source) name = kwargs.pop('name', 'dampr/{}'.format(random.random())) ds = pmer.pmer.runner(name, graph, **kwargs).run(sources) return [ValueEmitter(d) for d in ds] def _add_mapper(self, *args, **kwargs): output, ng = self.graph.add_mapper(*args, **kwargs) return output, Dampr(ng) def _add_reducer(self, *args, **kwargs): output, ng = self.graph.add_reducer(*args, **kwargs) return output, Dampr(ng) def _add_sink(self, *args, **kwargs): output, ng = self.graph.add_sink(*args, **kwargs) return output, Dampr(ng)
Ancestors (in MRO)
- Dampr
- __builtin__.object
Instance variables
var graph
var runner
Methods
def __init__(
self, graph=None, runner=None)
def __init__(self, graph=None, runner=None): if graph is None: graph = Graph() self.graph = graph if runner is None: runner = MTRunner self.runner = runner
def from_dataset(
cls, dataset)
Typically not used, this will read the raw outputs of a Dampr stage as an input.
@classmethod def from_dataset(cls, dataset): """ Typically not used, this will read the raw outputs of a Dampr stage as an input. """ assert isinstance(dataset, Chunker) source, ng = Graph().add_input(dataset) return PMap(source, Dampr(ng))
def json(
cls, *args, **kwargs)
Convenience function which reads newline-delimited json records.
@classmethod def json(cls, *args, **kwargs): """ Convenience function which reads newline-delimited json records. """ return cls.text(*args, **kwargs).map(json.loads)
def memory(
cls, items, partitions=50)
Create an in-memory dataset from the provided items. partitions
define how
many initial functions there will be.
>>> Dampr.memory([1,2,3,4,5])
@classmethod def memory(cls, items, partitions=50): """ Create an in-memory dataset from the provided items. `partitions` define how many initial functions there will be. >>> Dampr.memory([1,2,3,4,5]) """ mi = MemoryInput(list(enumerate(items)), partitions) source, ng = Graph().add_input(mi) return PMap(source, Dampr(ng))
def read_input(
cls, *datasets)
Reads from the provided datasets. When provided multiple datasets, it treats each as a separate partition.
read_input
can also take Chunker, which lazily returns a set of Datasets to
operate over.
>>> from dampr.inputs import MemoryInput >>> Dampr.read_input(MemoryInput(enumerate(range(5,10)), 2))
@classmethod def read_input(cls, *datasets): """ Reads from the provided datasets. When provided multiple datasets, it treats each as a separate partition. `read_input` can also take Chunker, which lazily returns a set of Datasets to operate over. >>> from dampr.inputs import MemoryInput >>> Dampr.read_input(MemoryInput(enumerate(range(5,10)), 2)) """ if len(datasets) == 1 and isinstance(datasets, Chunker): datasets = datasets[0] elif len(datasets) > 1: datasets = CatDataset(datasets) else: datasets = datasets[0] source, ng = Graph().add_input(datasets) return PMap(source, Dampr(ng))
def run(
self, *pmers, **kwargs)
Runs a graph or set of graphs.
foo = Dampr.memory([1,2,3,4,5]) bar = Dampr.memory([6,7,8,9,10]) left, right = Dampr.run(foo, bar) left.read() [1, 2, 3, 4, 5] right.read() [6, 7, 8, 9, 10]
@classmethod def run(self, *pmers, **kwargs): """ Runs a graph or set of graphs. >>> foo = Dampr.memory([1,2,3,4,5]) >>> bar = Dampr.memory([6,7,8,9,10]) >>> left, right = Dampr.run(foo, bar) >>> left.read() [1, 2, 3, 4, 5] >>> right.read() [6, 7, 8, 9, 10] """ sources = [] graph = None for i, pmer in enumerate(pmers): if isinstance(pmer, PMap): pmer = pmer.checkpoint() elif isinstance(pmer, PJoin): pmer = pmer.reduce(lambda l, r: (list(l), list(r))) if i == 0: graph = pmer.pmer.graph else: graph = pmer.pmer.graph.union(graph) sources.append(pmer.source) name = kwargs.pop('name', 'dampr/{}'.format(random.random())) ds = pmer.pmer.runner(name, graph, **kwargs).run(sources) return [ValueEmitter(d) for d in ds]
def text(
cls, fname, chunk_size=16777216)
Reads a file or directory of files into Dampr. Each record is assumed to be newline delimited.
When fname is a directory, it will walk the directory, collecting all
files within it as part of the collection. text
accepts globs.
chunk_size
describes how big each map portion will be.
Returns a PMap object.
>>> Dampr.text('/tmp', chunk_size=64*1024**2)
@classmethod def text(cls, fname, chunk_size=16*1024**2): """ Reads a file or directory of files into Dampr. Each record is assumed to be newline delimited. When fname is a directory, it will walk the directory, collecting all files within it as part of the collection. `text` accepts globs. `chunk_size` describes how big each map portion will be. Returns a PMap object. >>> Dampr.text('/tmp', chunk_size=64*1024**2) """ return cls.read_input(PathInput(fname, chunk_size))
class PJoin
Performs different types of joins between two grouped datasets.
class PJoin(PBase): """ Performs different types of joins between two grouped datasets. """ def __init__(self, source, pmer, right): super(PJoin, self).__init__(source, pmer) self.right = right def run(self, name=None, **kwargs): return self.reduce(lambda l, r: (list(l), list(r))).run(name, **kwargs) def reduce(self, aggregate, many=False): """ Performs an inner join between two datasets. The aggregate function will be provied two arguments: the left iterator and the right iterator. >>> left = Dampr.memory([("foo", 13), ("bar", 14)]).group_by(lambda x: x[0]) >>> right = Dampr.memory([("bar", "baller"), ("baz", "bag")]).group_by(lambda x: x[0]) >>> left.join(right).reduce(lambda lit, rit: (list(lit), list(rit))).read() [('bar', ([('bar', 14)], [('bar', 'baller')]))] If `many` is True, reduce will flatten the output into seperate records: >>> left.join(right).reduce(lambda lit, rit: list(lit) + list(rit), many=True).read() [('bar', ('bar', 14)), ('bar', ('bar', 'baller'))] """ def _reduce(k, left, right): return aggregate(left, right) source, pmer = self.pmer._add_reducer([self.source, self.right], KeyedInnerJoin(_reduce, many)) return PMap(source, pmer) def left_reduce(self, aggregate): """ Performs a left join on two datasets. In the case where the right dataset is missing the join key, it will call the aggregate function with an empty iterator. >>> left = Dampr.memory([("foo", 13), ("bar", 14)]).group_by(lambda x: x[0]) >>> right = Dampr.memory([("bar", "baller"), ("baz", "bag")]).group_by(lambda x: x[0]) >>> left.join(right).left_reduce(lambda lit, rit: (list(lit), list(rit))).read() [('bar', ([('bar', 14)], [('bar', 'baller')])), ('foo', ([('foo', 13)], []))] """ def _reduce(k, left, right): return aggregate(left, right) source, pmer = self.pmer._add_reducer([self.source, self.right], KeyedLeftJoin(_reduce)) return PMap(source, pmer) def _cross(self, crosser): def _cross(k1, v1, k2, v2): return k1, crosser(v1, v2) source, pmer = self.pmer._add_reducer([self.source, self.right], KeyedCrossJoin(_cross)) return PMap(source, pmer).map(lambda x: x[1])
Ancestors (in MRO)
- PJoin
- dampr.dampr.PBase
- __builtin__.object
Instance variables
var right
Methods
def __init__(
self, source, pmer, right)
def __init__(self, source, pmer, right): super(PJoin, self).__init__(source, pmer) self.right = right
def left_reduce(
self, aggregate)
Performs a left join on two datasets. In the case where the right dataset is missing the join key, it will call the aggregate function with an empty iterator.
>>> left = Dampr.memory([("foo", 13), ("bar", 14)]).group_by(lambda x: x[0]) >>> right = Dampr.memory([("bar", "baller"), ("baz", "bag")]).group_by(lambda x: x[0]) >>> left.join(right).left_reduce(lambda lit, rit: (list(lit), list(rit))).read() [('bar', ([('bar', 14)], [('bar', 'baller')])), ('foo', ([('foo', 13)], []))]
def left_reduce(self, aggregate): """ Performs a left join on two datasets. In the case where the right dataset is missing the join key, it will call the aggregate function with an empty iterator. >>> left = Dampr.memory([("foo", 13), ("bar", 14)]).group_by(lambda x: x[0]) >>> right = Dampr.memory([("bar", "baller"), ("baz", "bag")]).group_by(lambda x: x[0]) >>> left.join(right).left_reduce(lambda lit, rit: (list(lit), list(rit))).read() [('bar', ([('bar', 14)], [('bar', 'baller')])), ('foo', ([('foo', 13)], []))] """ def _reduce(k, left, right): return aggregate(left, right) source, pmer = self.pmer._add_reducer([self.source, self.right], KeyedLeftJoin(_reduce)) return PMap(source, pmer)
def read(
self, k=None, **kwargs)
Shorthand for run() followed by a read()
def read(self, k=None, **kwargs): """ Shorthand for run() followed by a read() """ return self.run(**kwargs).read(k)
def reduce(
self, aggregate, many=False)
Performs an inner join between two datasets. The aggregate function will be provied two arguments: the left iterator and the right iterator.
>>> left = Dampr.memory([("foo", 13), ("bar", 14)]).group_by(lambda x: x[0]) >>> right = Dampr.memory([("bar", "baller"), ("baz", "bag")]).group_by(lambda x: x[0]) >>> left.join(right).reduce(lambda lit, rit: (list(lit), list(rit))).read() [('bar', ([('bar', 14)], [('bar', 'baller')]))]
If many
is True, reduce will flatten the output into seperate records:
>>> left.join(right).reduce(lambda lit, rit: list(lit) + list(rit), many=True).read() [('bar', ('bar', 14)), ('bar', ('bar', 'baller'))]
def reduce(self, aggregate, many=False): """ Performs an inner join between two datasets. The aggregate function will be provied two arguments: the left iterator and the right iterator. >>> left = Dampr.memory([("foo", 13), ("bar", 14)]).group_by(lambda x: x[0]) >>> right = Dampr.memory([("bar", "baller"), ("baz", "bag")]).group_by(lambda x: x[0]) >>> left.join(right).reduce(lambda lit, rit: (list(lit), list(rit))).read() [('bar', ([('bar', 14)], [('bar', 'baller')]))] If `many` is True, reduce will flatten the output into seperate records: >>> left.join(right).reduce(lambda lit, rit: list(lit) + list(rit), many=True).read() [('bar', ('bar', 14)), ('bar', ('bar', 'baller'))] """ def _reduce(k, left, right): return aggregate(left, right) source, pmer = self.pmer._add_reducer([self.source, self.right], KeyedInnerJoin(_reduce, many)) return PMap(source, pmer)
def run(
self, name=None, **kwargs)
def run(self, name=None, **kwargs): return self.reduce(lambda l, r: (list(l), list(r))).run(name, **kwargs)
class PMap
Represents most mapping processes. Internally, it collects consecutive mapping operations and fuses them together into a single operation, speeding up many types of operations.
This class shouldn't be intialized manually.
class PMap(PBase): """ Represents most mapping processes. Internally, it collects consecutive mapping operations and fuses them together into a single operation, speeding up many types of operations. This class shouldn't be intialized manually. """ def __init__(self, source, pmer, agg=None): super(PMap, self).__init__(source, pmer) self.agg = [] if agg is None else agg def run(self, name=None, **kwargs): """ Run the defined graph. """ if len(self.agg) > 0: return self.checkpoint().run(name, **kwargs) else: return super(PMap, self).run(name, **kwargs) def _add_mapper(self, mapper): assert isinstance(mapper, Streamable) # If we have a fusable mapper, just add it to the set. return PMap(self.source, self.pmer, self.agg + [mapper]) def _add_map(self, f): return self._add_mapper(Map(f)) def sample(self, prob): """ Samples data with a given probability. For example: `graph.sample(0.1)` will uniformly sample 10% of the data in the collection. """ assert 0 <= prob <= 1.0 def _sample(k, v): if get_rand().random() < prob: yield k, v return self._add_map(_sample) def checkpoint(self, force=False, combiner=None, options=None): """ Checkpoint forces Dampr to fuse all cached maps and add it as a MR stage. This is useful when sharing the results of a computation with multiple other graphs. Without checkpoint(), Dampr would execute the shared graph multiple times rather than reuse the results of the computation: >>> evens = Dampr.memory([1,2,3,4,5]).filter(lambda x: x % 2 == 0).checkpoint() >>> summed = evens.group_by(lambda x: 1).sum() >>> multiplied = evens.group_by(lambda x: 1).reduce(lambda x, y: x * y) """ if len(self.agg) > 0 or force: aggs = [Map(_identity)] if len(self.agg) == 0 else self.agg[:] name = ' -> ' .join('{}'.format(str(a)) for a in aggs) name = 'Stage {}: %s' % (name) source, pmer = self.pmer._add_mapper([self.source], fuse(aggs), combiner=combiner, name=name, options=options) return PMap(source, pmer) return self def custom_mapper(self, mapper, name=None, **options): """ Custom Mapper provides a low-level interface to the underlying map function. Users can provide any instance which adhere's to the Mapper interface, allowing for powerful or specific implementations. This is typically used very rarely and has some caveats such as not fusing with other Mappers. Similarly, the implementer needs to understand more of the nuances associated with keys. >>> from dampr.base import Map >>> Dampr.memory([1,2,3,4,5]).custom_mapper(Map(lambda k, x: [(k, x+1)])).read() [2, 3, 4, 5, 6] """ if isinstance(mapper, Streamable): return self._add_mapper(mapper) assert isinstance(mapper, Mapper) name = name if name is not None else str(mapper) me = self.checkpoint() source, pmer = me.pmer._add_mapper([me.source], mapper, name=name, options=options) return PMap(source, pmer) def custom_reducer(self, reducer, name=None, **options): """ Allows the user to provide any Reducer which adheres to the Reducer interface. This is a very powerful, low-level interface which should be avoided when possible as it's easy to write bugs. >>> Dampr.memory([1,2,3,4,5]).custom_reducer(Reduce(lambda k, x: [(k, sum(x))])).read() [[(0, 1)], [(1, 2)], [(2, 3)], [(3, 4)], [(4, 5)]] """ assert isinstance(reducer, Reducer) me = self.checkpoint(force=True) name = name if name is not None else str(reducer) new_source, pmer = me.pmer._add_reducer([me.source], reducer, name=name, options=options) return PMap(new_source, pmer) def partition_map(self, f, **options): """ Provides a medium-level interface for writing custom functionality for mapping a partition. This can be used for creating custom logic in mappers for the sake of performance or additional functionality. Note: partition_map will still execute even if the partition is empty! Make sure the logic handles empty sets! `f` is a function that takes in an iterator of items in the partition and yields out group keys and new values. >>> def plus_one(items): ... for num in items: ... yield num, num + 1 ... >>> Dampr.memory([1,2,3,4,5]).partition_map(plus_one).read() [2, 3, 4, 5, 6] """ return self.custom_mapper(StreamMapper(f), **options) def partition_reduce(self, f): """ `partition_reduce` is a medium-level function that allows for more complex logic during reductions. It can be useful in certain cases where reductions over sets of reduced values is convenient, such as returning only the top K items in a dataset Note: `partition_reduce` will still execute on empty partitions! Make sure the logic handles cases where a partition is empty! >>> def largest_number(it): ... largest = float('-inf') ... for group_key, its in it: ... for value in its: ... largest = max(largest, value) ... yield "Largest", largest ... >>> Dampr.memory([1,2,3,4,5]).partition_reduce(largest_number).read(n_partitions=1) [('Largest', 5)] """ return self.custom_reducer(StreamReducer(f)) def len(self): """ Counts the number of items in the collection. >>> Dampr.memory([1,2,3,4,5]).len().read() [5] """ def _map_count(items): count = 0 for _ in items: count += 1 yield 1, count def _reduce_count(groups): count = 0 not_empty = False for _, counts in groups: not_empty = True for c in counts: count += c if not_empty: yield 1, count return self \ .partition_map(_map_count) \ .partition_reduce(_reduce_count) \ .map(lambda x: x[1]) def map(self, f): """ Maps elements in the underlying collection using function `f`: >>> Dampr.memory([1,2,3,4,5]).map(lambda x: x + 1).read() [2, 3, 4, 5, 6] """ def _map(k, v): yield k, f(v) return self._add_map(_map) def filter(self, f): """ Filters items from a collection based on a predicate f. A predicate return True keeps the item. >>> Dampr.memory([1,2,3,4,5]).filter(lambda x: x % 2 == 1).read() [1, 3, 5] """ def _filter(k, v): if f(v): yield k, v return self._add_map(_filter) def flat_map(self, f): """ Maps elements in the underlying collection using function f, flattening the results >>> Dampr.memory([1,2,3,4,5]).flat_map(range).read() [0, 0, 1, 0, 1, 2, 0, 1, 2, 3, 0, 1, 2, 3, 4] """ def _flat_map(k, v): for vi in f(v): yield k, vi return self._add_map(_flat_map) def group_by(self, key, vf=lambda x: x): """ Groups a collections of X by a key function, optionally mapping X to Y using `vf`. Returns a Reducer object for different types of aggregations >>> Dampr.memory([1,2,3,4,5]).group_by(lambda x: x % 2).reduce(lambda k, it: sum(it)).read() [(0, 6), (1, 9)] """ def _group_by(_key, value): yield key(value), vf(value) pm = self._add_map(_group_by).checkpoint() return PReduce(pm.source, pm.pmer) def a_group_by(self, key, vf=lambda x: x): """ Groups a collection of X by a key function and optionally mapping X to Y using `vf`. It differs from `group_by` by requiring an associative reduction operator: by forcing this restriction, Dampr is able to perform a partial reduction of the collection during the mapping sequence which can dramatically speed up the performace of the reduce stage. When possible, use a_group_by over the more general group_by. >>> Dampr.memory([1,2,3,4,5]).a_group_by(lambda x: x % 2).reduce(lambda x, y: x+y).read() [(0, 6), (1, 9)] """ def _a_group_by(_key, value): yield key(value), vf(value) # We don't checkpoint here! pm = self._add_map(_a_group_by) return ARReduce(pm) def fold_by(self, key, binop, value=lambda x: x, **options): """ Shortcut for a_group_by(key, value).reduce(binop) """ return self.a_group_by(key, value).reduce(binop, **options) def sort_by(self, key, **options): """ Sorts the results by a given key function. >>> Dampr.memory([1,2,3,4,5]).filter(lambda x: x % 2 == 1).sort_by(lambda x: -x).read() [5, 3, 1] """ def _sort_by(_key, value): yield key(value), value return self._add_map(_sort_by).checkpoint(options=options) def join(self, other): """ Joins two independent computations, returning a Joining class. This is a powerful and expensive operation which can merge two Dampr collections together. """ assert isinstance(other, PBase) me = self.checkpoint(True) if isinstance(other, PMap): other = other.checkpoint(True) pmer = Dampr(me.pmer.graph.union(other.pmer.graph)) return PJoin(me.source, pmer, other.source) def count(self, key=lambda x: x, **options): """ Counts each item X in the collection by its key function. >>> Dampr.memory([1,2,3,4,5]).count(lambda x: x % 2).read() [(0, 2), (1, 3)] """ return self.a_group_by(key, lambda v: 1) \ .reduce(operator.add, **options) def mean(self, key=lambda x: 1, value=lambda x: x, **options): """ Finds the mean of X, grouped by its key function and optionally mapped by a value function: >>> ages = [("Andrew", 33), ("Alice", 42), ("Andrew", 12), ("Bob", 51)] >>> Dampr.memory(ages).mean(lambda x: x[0], lambda v: v[1]).read() [('Alice', 42.0), ('Andrew', 22.5), ('Bob', 51.0)] """ def _mean_binop(x, y): return x[0] + y[0], x[1] + y[1] def _average(x): return (x[0], x[1][0] / float(x[1][1])) return self.a_group_by(key, lambda v: (value(v), 1)) \ .reduce(_mean_binop, **options) \ .map(_average) def inspect(self, prefix="", exit=False): """ Inspect is a debug function which prints each item X that flows through it. It's valueable in inspecting intermediate results of a pipeline without changing the values internally. """ def _inspect(k, v): print("{}: {}".format(prefix, v)) yield k, v ins = self._add_map(_inspect) if exit: ins.run() sys.exit(0) return ins def cached(self, **options): """ The cached function runs a graph and stores it in memory. This is useful for small datasets to gain extra performance in subsequent computations. >>> Dampr.memory([1,2,3,4,5,6]).mean(lambda x: x % 2).cached().read() [(0, 4.0), (1, 3.0)] """ # Run the pipeline, load it into memory, and create a new graph options['memory'] = True return self.checkpoint(options=options) def sink(self, path): """ Since writes each X in a collection to a given path. The path will create a directory and write each map or reduce partition into the directory as partitions. Sink assumes each X in the collection is already a unicode string. >>> Dampr.memory(["foo", "bar", "baz"]).sink("/tmp/foo").run() >>> open("/tmp/foo/0").read() >>> open("/tmp/foo/1").read() >>> open("/tmp/foo/2").read() """ aggs = [Map(_identity)] if len(self.agg) == 0 else self.agg[:] name = ' -> ' .join('{}'.format(a.mapper.__name__) for a in aggs) name = 'Stage {}: %s' % (name) source, pmer = self.pmer._add_sink([self.source], fuse(aggs), path=path, name=name, options=None) return PMap(source, pmer) def sink_tsv(self, path): """ A convenience function which takes a tuple or list, creates a simple tab-delimited output, and sinks it to the provided path: >>> Dampr.memory([("Hank Aaron", 755)]).sink_tsv("/tmp/foo").run() >>> open("/tmp/foo/0").read() """ return self.map(lambda x: u'\t'.join(unicode(p) for p in x)).sink(path) def sink_json(self, path): """ A convenience function which takes a simple python object and serializes it to a line-delimited json to the given path: >>> Dampr.memory([{"name": "Hank Aaron", "home runs": 755}]).sink_json("/tmp/foo").run() >>> open("/tmp/foo/0").read() """ return self.map(json.dumps).sink(path) def cross_right(self, other, cross, memory=False): """ Produces the cross product between two datasets during the map stage. This is an incredibly expensive operation when done against two large datasets. When it is known that the right dataset is small, setting `memory` to `True` will cache the right dataset in memory, speeding up the computation significantly. Two items, Xi and Yi, are joined with the cross function. >>> left = Dampr.memory([1,2,3,4,5]) >>> right = Dampr.memory(['foo', 'bar']) >>> left.cross_right(right, lambda x, y: (x, y)).read() [(1, 'foo'), (1, 'bar'), (2, 'foo'), (2, 'bar'), (3, 'foo'), (3, 'bar'), (4, 'foo'), (4, 'bar'), (5, 'foo'), (5, 'bar')] """ assert isinstance(other, PMap) return other.cross_left(self, lambda xi, yi: cross(yi, xi), memory) def cross_left(self, other, cross, memory=False, **options): """ Produces the cross product between two datasets during the map stage. This is an incredibly expensive operation when done against two large datasets. When it is known that the left dataset is small, setting `memory` to `True` will cache the right dataset in memory, speeding up the computation significantly. Two items, Xi and Yi, are joined with the cross function. >>> left = Dampr.memory([1,2,3,4,5]) >>> right = Dampr.memory(['foo', 'bar']) >>> left.cross_left(right, lambda x, y: (x, y)).read() [(1, 'foo'), (2, 'foo'), (3, 'foo'), (4, 'foo'), (5, 'foo'), (1, 'bar'), (2, 'bar'), (3, 'bar'), (4, 'bar'), (5, 'bar')] """ def _cross(k1, v1, k2, v2): yield k1, cross(v2, v1) me = self.checkpoint() other = other.checkpoint() pmer = Dampr(me.pmer.graph.union(other.pmer.graph)) name = 'Stage {}: Cross' source, pmer = pmer._add_mapper([other.source, me.source], MapCrossJoin(_cross, cache=memory), combiner=None, name=name, options=options) return PMap(source, pmer) def topk(self, k, value=None): """ Computes the top k elements in the set. The 'value' function is responsible for returning a comparable object. >>> Dampr.memory([1,3,2,4,2.2]).topk(2).read() [3, 4] >>> Dampr.memory([1,3,2,4,2.2]).topk(2, lambda x: -x).read() [1, 2] """ if value is None: value = lambda x: x import heapq def map_topk(it): heap = [] for x in it: heapq.heappush(heap, (value(x), x)) if len(heap) > k: heapq.heappop(heap) return ((1, x) for x in heap) def reduce_topk(it): counts = (v for k, vit in it for v in vit) for count, x in heapq.nlargest(k, counts): yield x, 1 return self \ .partition_map(map_topk) \ .partition_reduce(reduce_topk) \ .map(lambda x: x[0])
Ancestors (in MRO)
- PMap
- dampr.dampr.PBase
- __builtin__.object
Instance variables
var agg
Methods
def __init__(
self, source, pmer, agg=None)
def __init__(self, source, pmer, agg=None): super(PMap, self).__init__(source, pmer) self.agg = [] if agg is None else agg
def a_group_by(
self, key, vf=<function <lambda> at 0x7f14a9c8cb18>)
Groups a collection of X by a key function and optionally mapping X to Y
using vf
. It differs from group_by
by requiring an associative reduction
operator: by forcing this restriction, Dampr is able to perform a partial
reduction of the collection during the mapping sequence which can dramatically
speed up the performace of the reduce stage.
When possible, use a_group_by over the more general group_by.
>>> Dampr.memory([1,2,3,4,5]).a_group_by(lambda x: x % 2).reduce(lambda x, y: x+y).read() [(0, 6), (1, 9)]
def a_group_by(self, key, vf=lambda x: x): """ Groups a collection of X by a key function and optionally mapping X to Y using `vf`. It differs from `group_by` by requiring an associative reduction operator: by forcing this restriction, Dampr is able to perform a partial reduction of the collection during the mapping sequence which can dramatically speed up the performace of the reduce stage. When possible, use a_group_by over the more general group_by. >>> Dampr.memory([1,2,3,4,5]).a_group_by(lambda x: x % 2).reduce(lambda x, y: x+y).read() [(0, 6), (1, 9)] """ def _a_group_by(_key, value): yield key(value), vf(value) # We don't checkpoint here! pm = self._add_map(_a_group_by) return ARReduce(pm)
def cached(
self, **options)
The cached function runs a graph and stores it in memory. This is useful for small datasets to gain extra performance in subsequent computations.
>>> Dampr.memory([1,2,3,4,5,6]).mean(lambda x: x % 2).cached().read() [(0, 4.0), (1, 3.0)]
def cached(self, **options): """ The cached function runs a graph and stores it in memory. This is useful for small datasets to gain extra performance in subsequent computations. >>> Dampr.memory([1,2,3,4,5,6]).mean(lambda x: x % 2).cached().read() [(0, 4.0), (1, 3.0)] """ # Run the pipeline, load it into memory, and create a new graph options['memory'] = True return self.checkpoint(options=options)
def checkpoint(
self, force=False, combiner=None, options=None)
Checkpoint forces Dampr to fuse all cached maps and add it as a MR stage.
This is useful when sharing the results of a computation with multiple other graphs.
Without checkpoint(), Dampr would execute the shared graph multiple times rather than reuse the results of the computation:
>>> evens = Dampr.memory([1,2,3,4,5]).filter(lambda x: x % 2 == 0).checkpoint() >>> summed = evens.group_by(lambda x: 1).sum() >>> multiplied = evens.group_by(lambda x: 1).reduce(lambda x, y: x * y)
def checkpoint(self, force=False, combiner=None, options=None): """ Checkpoint forces Dampr to fuse all cached maps and add it as a MR stage. This is useful when sharing the results of a computation with multiple other graphs. Without checkpoint(), Dampr would execute the shared graph multiple times rather than reuse the results of the computation: >>> evens = Dampr.memory([1,2,3,4,5]).filter(lambda x: x % 2 == 0).checkpoint() >>> summed = evens.group_by(lambda x: 1).sum() >>> multiplied = evens.group_by(lambda x: 1).reduce(lambda x, y: x * y) """ if len(self.agg) > 0 or force: aggs = [Map(_identity)] if len(self.agg) == 0 else self.agg[:] name = ' -> ' .join('{}'.format(str(a)) for a in aggs) name = 'Stage {}: %s' % (name) source, pmer = self.pmer._add_mapper([self.source], fuse(aggs), combiner=combiner, name=name, options=options) return PMap(source, pmer) return self
def count(
self, key=<function <lambda> at 0x7f14a9c8cde8>, **options)
Counts each item X in the collection by its key function.
>>> Dampr.memory([1,2,3,4,5]).count(lambda x: x % 2).read() [(0, 2), (1, 3)]
def count(self, key=lambda x: x, **options): """ Counts each item X in the collection by its key function. >>> Dampr.memory([1,2,3,4,5]).count(lambda x: x % 2).read() [(0, 2), (1, 3)] """ return self.a_group_by(key, lambda v: 1) \ .reduce(operator.add, **options)
def cross_left(
self, other, cross, memory=False, **options)
Produces the cross product between two datasets during the map stage. This is an incredibly expensive operation when done against two large datasets.
When it is known that the left dataset is small, setting memory
to True
will cache the right dataset in memory, speeding up the computation significantly.
Two items, Xi and Yi, are joined with the cross function.
>>> left = Dampr.memory([1,2,3,4,5]) >>> right = Dampr.memory(['foo', 'bar']) >>> left.cross_left(right, lambda x, y: (x, y)).read() [(1, 'foo'), (2, 'foo'), (3, 'foo'), (4, 'foo'), (5, 'foo'), (1, 'bar'), (2, 'bar'), (3, 'bar'), (4, 'bar'), (5, 'bar')]
def cross_left(self, other, cross, memory=False, **options): """ Produces the cross product between two datasets during the map stage. This is an incredibly expensive operation when done against two large datasets. When it is known that the left dataset is small, setting `memory` to `True` will cache the right dataset in memory, speeding up the computation significantly. Two items, Xi and Yi, are joined with the cross function. >>> left = Dampr.memory([1,2,3,4,5]) >>> right = Dampr.memory(['foo', 'bar']) >>> left.cross_left(right, lambda x, y: (x, y)).read() [(1, 'foo'), (2, 'foo'), (3, 'foo'), (4, 'foo'), (5, 'foo'), (1, 'bar'), (2, 'bar'), (3, 'bar'), (4, 'bar'), (5, 'bar')] """ def _cross(k1, v1, k2, v2): yield k1, cross(v2, v1) me = self.checkpoint() other = other.checkpoint() pmer = Dampr(me.pmer.graph.union(other.pmer.graph)) name = 'Stage {}: Cross' source, pmer = pmer._add_mapper([other.source, me.source], MapCrossJoin(_cross, cache=memory), combiner=None, name=name, options=options) return PMap(source, pmer)
def cross_right(
self, other, cross, memory=False)
Produces the cross product between two datasets during the map stage. This is an incredibly expensive operation when done against two large datasets.
When it is known that the right dataset is small, setting memory
to True
will cache the right dataset in memory, speeding up the computation significantly.
Two items, Xi and Yi, are joined with the cross function.
>>> left = Dampr.memory([1,2,3,4,5]) >>> right = Dampr.memory(['foo', 'bar']) >>> left.cross_right(right, lambda x, y: (x, y)).read() [(1, 'foo'), (1, 'bar'), (2, 'foo'), (2, 'bar'), (3, 'foo'), (3, 'bar'), (4, 'foo'), (4, 'bar'), (5, 'foo'), (5, 'bar')]
def cross_right(self, other, cross, memory=False): """ Produces the cross product between two datasets during the map stage. This is an incredibly expensive operation when done against two large datasets. When it is known that the right dataset is small, setting `memory` to `True` will cache the right dataset in memory, speeding up the computation significantly. Two items, Xi and Yi, are joined with the cross function. >>> left = Dampr.memory([1,2,3,4,5]) >>> right = Dampr.memory(['foo', 'bar']) >>> left.cross_right(right, lambda x, y: (x, y)).read() [(1, 'foo'), (1, 'bar'), (2, 'foo'), (2, 'bar'), (3, 'foo'), (3, 'bar'), (4, 'foo'), (4, 'bar'), (5, 'foo'), (5, 'bar')] """ assert isinstance(other, PMap) return other.cross_left(self, lambda xi, yi: cross(yi, xi), memory)
def custom_mapper(
self, mapper, name=None, **options)
Custom Mapper provides a low-level interface to the underlying map function. Users can provide any instance which adhere's to the Mapper interface, allowing for powerful or specific implementations.
This is typically used very rarely and has some caveats such as not fusing with other Mappers. Similarly, the implementer needs to understand more of the nuances associated with keys.
>>> from dampr.base import Map >>> Dampr.memory([1,2,3,4,5]).custom_mapper(Map(lambda k, x: [(k, x+1)])).read() [2, 3, 4, 5, 6]
def custom_mapper(self, mapper, name=None, **options): """ Custom Mapper provides a low-level interface to the underlying map function. Users can provide any instance which adhere's to the Mapper interface, allowing for powerful or specific implementations. This is typically used very rarely and has some caveats such as not fusing with other Mappers. Similarly, the implementer needs to understand more of the nuances associated with keys. >>> from dampr.base import Map >>> Dampr.memory([1,2,3,4,5]).custom_mapper(Map(lambda k, x: [(k, x+1)])).read() [2, 3, 4, 5, 6] """ if isinstance(mapper, Streamable): return self._add_mapper(mapper) assert isinstance(mapper, Mapper) name = name if name is not None else str(mapper) me = self.checkpoint() source, pmer = me.pmer._add_mapper([me.source], mapper, name=name, options=options) return PMap(source, pmer)
def custom_reducer(
self, reducer, name=None, **options)
Allows the user to provide any Reducer which adheres to the Reducer interface. This is a very powerful, low-level interface which should be avoided when possible as it's easy to write bugs.
>>> Dampr.memory([1,2,3,4,5]).custom_reducer(Reduce(lambda k, x: [(k, sum(x))])).read() [[(0, 1)], [(1, 2)], [(2, 3)], [(3, 4)], [(4, 5)]]
def custom_reducer(self, reducer, name=None, **options): """ Allows the user to provide any Reducer which adheres to the Reducer interface. This is a very powerful, low-level interface which should be avoided when possible as it's easy to write bugs. >>> Dampr.memory([1,2,3,4,5]).custom_reducer(Reduce(lambda k, x: [(k, sum(x))])).read() [[(0, 1)], [(1, 2)], [(2, 3)], [(3, 4)], [(4, 5)]] """ assert isinstance(reducer, Reducer) me = self.checkpoint(force=True) name = name if name is not None else str(reducer) new_source, pmer = me.pmer._add_reducer([me.source], reducer, name=name, options=options) return PMap(new_source, pmer)
def filter(
self, f)
Filters items from a collection based on a predicate f. A predicate return True keeps the item.
>>> Dampr.memory([1,2,3,4,5]).filter(lambda x: x % 2 == 1).read() [1, 3, 5]
def filter(self, f): """ Filters items from a collection based on a predicate f. A predicate return True keeps the item. >>> Dampr.memory([1,2,3,4,5]).filter(lambda x: x % 2 == 1).read() [1, 3, 5] """ def _filter(k, v): if f(v): yield k, v return self._add_map(_filter)
def flat_map(
self, f)
Maps elements in the underlying collection using function f, flattening the results
>>> Dampr.memory([1,2,3,4,5]).flat_map(range).read() [0, 0, 1, 0, 1, 2, 0, 1, 2, 3, 0, 1, 2, 3, 4]
def flat_map(self, f): """ Maps elements in the underlying collection using function f, flattening the results >>> Dampr.memory([1,2,3,4,5]).flat_map(range).read() [0, 0, 1, 0, 1, 2, 0, 1, 2, 3, 0, 1, 2, 3, 4] """ def _flat_map(k, v): for vi in f(v): yield k, vi return self._add_map(_flat_map)
def fold_by(
self, key, binop, value=<function <lambda> at 0x7f14a9c8cc08>, **options)
Shortcut for a_group_by(key, value).reduce(binop)
def fold_by(self, key, binop, value=lambda x: x, **options): """ Shortcut for a_group_by(key, value).reduce(binop) """ return self.a_group_by(key, value).reduce(binop, **options)
def group_by(
self, key, vf=<function <lambda> at 0x7f14a9c8ca28>)
Groups a collections of X by a key function, optionally mapping X to Y
using vf
. Returns a Reducer object for different types of aggregations
>>> Dampr.memory([1,2,3,4,5]).group_by(lambda x: x % 2).reduce(lambda k, it: sum(it)).read() [(0, 6), (1, 9)]
def group_by(self, key, vf=lambda x: x): """ Groups a collections of X by a key function, optionally mapping X to Y using `vf`. Returns a Reducer object for different types of aggregations >>> Dampr.memory([1,2,3,4,5]).group_by(lambda x: x % 2).reduce(lambda k, it: sum(it)).read() [(0, 6), (1, 9)] """ def _group_by(_key, value): yield key(value), vf(value) pm = self._add_map(_group_by).checkpoint() return PReduce(pm.source, pm.pmer)
def inspect(
self, prefix='', exit=False)
Inspect is a debug function which prints each item X that flows through it. It's valueable in inspecting intermediate results of a pipeline without changing the values internally.
def inspect(self, prefix="", exit=False): """ Inspect is a debug function which prints each item X that flows through it. It's valueable in inspecting intermediate results of a pipeline without changing the values internally. """ def _inspect(k, v): print("{}: {}".format(prefix, v)) yield k, v ins = self._add_map(_inspect) if exit: ins.run() sys.exit(0) return ins
def join(
self, other)
Joins two independent computations, returning a Joining class.
This is a powerful and expensive operation which can merge two Dampr collections together.
def join(self, other): """ Joins two independent computations, returning a Joining class. This is a powerful and expensive operation which can merge two Dampr collections together. """ assert isinstance(other, PBase) me = self.checkpoint(True) if isinstance(other, PMap): other = other.checkpoint(True) pmer = Dampr(me.pmer.graph.union(other.pmer.graph)) return PJoin(me.source, pmer, other.source)
def len(
self)
Counts the number of items in the collection.
>>> Dampr.memory([1,2,3,4,5]).len().read() [5]
def len(self): """ Counts the number of items in the lection. >>> Dampr.memory([1,2,3,4,5]).len().read() [5] """ def _map_count(items): count = 0 for _ in items: count += 1 yield 1, count def _reduce_count(groups): count = 0 not_empty = False for _, counts in groups: not_empty = True for c in counts: count += c if not_empty: yield 1, count return self \ .partition_map(_map_count) \ .partition_reduce(_reduce_count) \ .map(lambda x: x[1])
def map(
self, f)
Maps elements in the underlying collection using function
f
:
>>> Dampr.memory([1,2,3,4,5]).map(lambda x: x + 1).read() [2, 3, 4, 5, 6]
def map(self, f): """ Maps elements in the underlying collection using function `f`: >>> Dampr.memory([1,2,3,4,5]).map(lambda x: x + 1).read() [2, 3, 4, 5, 6] """ def _map(k, v): yield k, f(v) return self._add_map(_map)
def mean(
self, key=<function <lambda> at 0x7f14a9c8ced8>, value=<function <lambda> at 0x7f14a9c8cf50>, **options)
Finds the mean of X, grouped by its key function and optionally mapped by a value function:
>>> ages = [("Andrew", 33), ("Alice", 42), ("Andrew", 12), ("Bob", 51)] >>> Dampr.memory(ages).mean(lambda x: x[0], lambda v: v[1]).read() [('Alice', 42.0), ('Andrew', 22.5), ('Bob', 51.0)]
def mean(self, key=lambda x: 1, value=lambda x: x, **options): """ Finds the mean of X, grouped by its key function and optionally mapped by a value function: >>> ages = [("Andrew", 33), ("Alice", 42), ("Andrew", 12), ("Bob", 51)] >>> Dampr.memory(ages).mean(lambda x: x[0], lambda v: v[1]).read() [('Alice', 42.0), ('Andrew', 22.5), ('Bob', 51.0)] """ def _mean_binop(x, y): return x[0] + y[0], x[1] + y[1] def _average(x): return (x[0], x[1][0] / float(x[1][1])) return self.a_group_by(key, lambda v: (value(v), 1)) \ .reduce(_mean_binop, **options) \ .map(_average)
def partition_map(
self, f, **options)
Provides a medium-level interface for writing custom functionality for mapping a partition. This can be used for creating custom logic in mappers for the sake of performance or additional functionality.
Note: partition_map will still execute even if the partition is empty! Make sure the logic handles empty sets!
f
is a function that takes in an iterator of items in the partition and yields
out group keys and new values.
>>> def plus_one(items): ... for num in items: ... yield num, num + 1 ... >>> Dampr.memory([1,2,3,4,5]).partition_map(plus_one).read() [2, 3, 4, 5, 6]
def partition_map(self, f, **options): """ Provides a medium-level interface for writing custom functionality for mapping a partition. This can be used for creating custom logic in mappers for the sake of performance or additional functionality. Note: partition_map will still execute even if the partition is empty! Make sure the logic handles empty sets! `f` is a function that takes in an iterator of items in the partition and yields out group keys and new values. >>> def plus_one(items): ... for num in items: ... yield num, num + 1 ... >>> Dampr.memory([1,2,3,4,5]).partition_map(plus_one).read() [2, 3, 4, 5, 6] """ return self.custom_mapper(StreamMapper(f), **options)
def partition_reduce(
self, f)
partition_reduce
is a medium-level function that allows for more complex logic
during reductions. It can be useful in certain cases where reductions over sets
of reduced values is convenient, such as returning only the top K items in a dataset
Note: partition_reduce
will still execute on empty partitions! Make sure the
logic handles cases where a partition is empty!
>>> def largest_number(it): ... largest = float('-inf') ... for group_key, its in it: ... for value in its: ... largest = max(largest, value) ... yield "Largest", largest ... >>> Dampr.memory([1,2,3,4,5]).partition_reduce(largest_number).read(n_partitions=1) [('Largest', 5)]
def partition_reduce(self, f): """ `partition_reduce` is a medium-level function that allows for more complex logic during reductions. It can be useful in certain cases where reductions over sets of reduced values is convenient, such as returning only the top K items in a dataset Note: `partition_reduce` will still execute on empty partitions! Make sure the logic handles cases where a partition is empty! >>> def largest_number(it): ... largest = float('-inf') ... for group_key, its in it: ... for value in its: ... largest = max(largest, value) ... yield "Largest", largest ... >>> Dampr.memory([1,2,3,4,5]).partition_reduce(largest_number).read(n_partitions=1) [('Largest', 5)] """ return self.custom_reducer(StreamReducer(f))
def read(
self, k=None, **kwargs)
Shorthand for run() followed by a read()
def read(self, k=None, **kwargs): """ Shorthand for run() followed by a read() """ return self.run(**kwargs).read(k)
def run(
self, name=None, **kwargs)
Run the defined graph.
def run(self, name=None, **kwargs): """ Run the defined graph. """ if len(self.agg) > 0: return self.checkpoint().run(name, **kwargs) else: return super(PMap, self).run(name, **kwargs)
def sample(
self, prob)
Samples data with a given probability. For example:
graph.sample(0.1)
will uniformly sample 10% of the data in the collection.
def sample(self, prob): """ Samples data with a given probability. For example: `graph.sample(0.1)` will uniformly sample 10% of the data in the collection. """ assert 0 <= prob <= 1.0 def _sample(k, v): if get_rand().random() < prob: yield k, v return self._add_map(_sample)
def sink(
self, path)
Since writes each X in a collection to a given path. The path will create a directory and write each map or reduce partition into the directory as partitions. Sink assumes each X in the collection is already a unicode string.
>>> Dampr.memory(["foo", "bar", "baz"]).sink("/tmp/foo").run() >>> open("/tmp/foo/0").read() >>> open("/tmp/foo/1").read() >>> open("/tmp/foo/2").read()
def sink(self, path): """ Since writes each X in a collection to a given path. The path will create a directory and write each map or reduce partition into the directory as partitions. Sink assumes each X in the collection is already a unicode string. >>> Dampr.memory(["foo", "bar", "baz"]).sink("/tmp/foo").run() >>> open("/tmp/foo/0").read() >>> open("/tmp/foo/1").read() >>> open("/tmp/foo/2").read() """ aggs = [Map(_identity)] if len(self.agg) == 0 else self.agg[:] name = ' -> ' .join('{}'.format(a.mapper.__name__) for a in aggs) name = 'Stage {}: %s' % (name) source, pmer = self.pmer._add_sink([self.source], fuse(aggs), path=path, name=name, options=None) return PMap(source, pmer)
def sink_json(
self, path)
A convenience function which takes a simple python object and serializes it to a line-delimited json to the given path:
>>> Dampr.memory([{"name": "Hank Aaron", "home runs": 755}]).sink_json("/tmp/foo").run() >>> open("/tmp/foo/0").read()
def sink_json(self, path): """ A convenience function which takes a simple python object and serializes it to a line-delimited json to the given path: >>> Dampr.memory([{"name": "Hank Aaron", "home runs": 755}]).sink_json("/tmp/foo").run() >>> open("/tmp/foo/0").read() """ return self.map(json.dumps).sink(path)
def sink_tsv(
self, path)
A convenience function which takes a tuple or list, creates a simple tab-delimited output, and sinks it to the provided path:
>>> Dampr.memory([("Hank Aaron", 755)]).sink_tsv("/tmp/foo").run() >>> open("/tmp/foo/0").read()
def sink_tsv(self, path): """ A convenience function which takes a tuple or list, creates a simple tab-delimited output, and sinks it to the provided path: >>> Dampr.memory([("Hank Aaron", 755)]).sink_tsv("/tmp/foo").run() >>> open("/tmp/foo/0").read() """ return self.map(lambda x: u'\t'.join(unicode(p) for p in x)).sink(path)
def sort_by(
self, key, **options)
Sorts the results by a given key function.
Dampr.memory([1,2,3,4,5]).filter(lambda x: x % 2 == 1).sort_by(lambda x: -x).read() [5, 3, 1]
def sort_by(self, key, **options): """ Sorts the results by a given key function. >>> Dampr.memory([1,2,3,4,5]).filter(lambda x: x % 2 == 1).sort_by(lambda x: -x).read() [5, 3, 1] """ def _sort_by(_key, value): yield key(value), value return self._add_map(_sort_by).checkpoint(options=options)
def topk(
self, k, value=None)
Computes the top k elements in the set. The 'value' function is responsible for returning a comparable object.
>>> Dampr.memory([1,3,2,4,2.2]).topk(2).read() [3, 4] >>> Dampr.memory([1,3,2,4,2.2]).topk(2, lambda x: -x).read() [1, 2]
def topk(self, k, value=None): """ Computes the top k elements in the set. The 'value' function is responsible for returning a comparable object. >>> Dampr.memory([1,3,2,4,2.2]).topk(2).read() [3, 4] >>> Dampr.memory([1,3,2,4,2.2]).topk(2, lambda x: -x).read() [1, 2] """ if value is None: value = lambda x: x import heapq def map_topk(it): heap = [] for x in it: heapq.heappush(heap, (value(x), x)) if len(heap) > k: heapq.heappop(heap) return ((1, x) for x in heap) def reduce_topk(it): counts = (v for k, vit in it for v in vit) for count, x in heapq.nlargest(k, counts): yield x, 1 return self \ .partition_map(map_topk) \ .partition_reduce(reduce_topk) \ .map(lambda x: x[0])
class PReduce
A more general reduce class.
class PReduce(PBase): """ A more general reduce class. """ def reduce(self, f): """ Reduces a grouped set of items by f, which takes two arguments: the group key and an iterator lazily yield all items in the group. >>> Dampr.memory([1,2,3,4,5]).group_by(lambda x: x % 2).reduce(lambda k, it: sum(it)).read() [(0, 6), (1, 9)] """ new_source, pmer = self.pmer._add_reducer([self.source], KeyedReduce(f)) return PMap(new_source, pmer) def unique(self, key=lambda x: x): """ Returns the unique set of items in the grouping. >>> names = [("Andrew", 1), ("Andrew", 1), ("Andrew", 2), ("Becky", 13)] >>> Dampr.memory(names).group_by(lambda x: x[0], lambda x: x[1]).unique().read() [('Andrew', [1, 2]), ('Becky', [13])] """ def _uniq(k, it): seen = set() agg = [] for v in it: fv = key(v) if fv not in seen: seen.add(fv) agg.append(v) return agg return self.reduce(_uniq) def join(self, other): """ Performs a join between two datasets on a given key, returning a PJoin object. """ assert isinstance(other, PBase) if isinstance(other, PMap): other = other.checkpoint(True) pmer = Dampr(self.pmer.graph.union(other.pmer.graph)) return PJoin(self.source, pmer, other.source) def partition_reduce(self, f): """ Provides medium-level functionality for partition reductions. See PMap.partition_reduce for more details. """ reducer = StreamReducer(f) name = str(reducer) new_source, pmer = me.pmer._add_reducer([self.pm.source], reducer, name=name) return PMap(new_source, pmer)
Ancestors (in MRO)
- PReduce
- dampr.dampr.PBase
- __builtin__.object
Methods
def __init__(
self, source, pmer)
def __init__(self, source, pmer): assert isinstance(source, Source) self.source = source self.pmer = pmer
def join(
self, other)
Performs a join between two datasets on a given key, returning a PJoin object.
def join(self, other): """ Performs a join between two datasets on a given key, returning a PJoin object. """ assert isinstance(other, PBase) if isinstance(other, PMap): other = other.checkpoint(True) pmer = Dampr(self.pmer.graph.union(other.pmer.graph)) return PJoin(self.source, pmer, other.source)
def partition_reduce(
self, f)
Provides medium-level functionality for partition reductions. See PMap.partition_reduce for more details.
def partition_reduce(self, f): """ Provides medium-level functionality for partition reductions. See PMap.partition_reduce for more details. """ reducer = StreamReducer(f) name = str(reducer) new_source, pmer = me.pmer._add_reducer([self.pm.source], reducer, name=name) return PMap(new_source, pmer)
def read(
self, k=None, **kwargs)
Shorthand for run() followed by a read()
def read(self, k=None, **kwargs): """ Shorthand for run() followed by a read() """ return self.run(**kwargs).read(k)
def reduce(
self, f)
Reduces a grouped set of items by f, which takes two arguments: the group key and an iterator lazily yield all items in the group.
>>> Dampr.memory([1,2,3,4,5]).group_by(lambda x: x % 2).reduce(lambda k, it: sum(it)).read() [(0, 6), (1, 9)]
def reduce(self, f): """ Reduces a grouped set of items by f, which takes two arguments: the group key and an iterator lazily yield all items in the group. >>> Dampr.memory([1,2,3,4,5]).group_by(lambda x: x % 2).reduce(lambda k, it: sum(it)).read() [(0, 6), (1, 9)] """ new_source, pmer = self.pmer._add_reducer([self.source], KeyedReduce(f)) return PMap(new_source, pmer)
def run(
self, name=None, **kwargs)
Evaluates the composed Dampr graph with the provided name and subsequent options. By default, uses /tmp as temporary storage.
Returns a ValueEmitter useful for shell access.
def run(self, name=None, **kwargs): """ Evaluates the composed Dampr graph with the provided name and subsequent options. By default, uses /tmp as temporary storage. Returns a ValueEmitter useful for shell access. """ if name is None: name = 'dampr/{}'.format(random.random()) logging.debug("run source: %s", self.source) ds = self.pmer.runner(name, self.pmer.graph, **kwargs).run([self.source]) return ValueEmitter(ds[0])
def unique(
self, key=<function <lambda> at 0x7f14a9c906e0>)
Returns the unique set of items in the grouping.
names = [("Andrew", 1), ("Andrew", 1), ("Andrew", 2), ("Becky", 13)] Dampr.memory(names).group_by(lambda x: x[0], lambda x: x[1]).unique().read() [('Andrew', [1, 2]), ('Becky', [13])]
def unique(self, key=lambda x: x): """ Returns the unique set of items in the grouping. >>> names = [("Andrew", 1), ("Andrew", 1), ("Andrew", 2), ("Becky", 13)] >>> Dampr.memory(names).group_by(lambda x: x[0], lambda x: x[1]).unique().read() [('Andrew', [1, 2]), ('Becky', [13])] """ def _uniq(k, it): seen = set() agg = [] for v in it: fv = key(v) if fv not in seen: seen.add(fv) agg.append(v) return agg return self.reduce(_uniq)


Musb device object used by the MUSB driver. More...
#include <usb_musb_dcd.h>
Data Fields | |
| uint32_t | baseAddr |
| uint32_t | uiSubBaseAddr |
| uint32_t | phyBaseAddr |
| char | name [10] |
| uint32_t | controllerId |
| uint32_t | deviceNewAddrValidFlag |
| uint32_t | devAddr |
| usbDcd_t * | pDcdCore |
| usbEndptRequest_t * | inEpReq |
| usbEndptRequest_t * | outEpReq |
Musb device object used by the MUSB driver.
| uint32_t usbMusbDcdDevice::baseAddr |
Base address of the USB core
| uint32_t usbMusbDcdDevice::controllerId |
ID of the controllerr
| uint32_t usbMusbDcdDevice::devAddr |
Device address
| uint32_t usbMusbDcdDevice::deviceNewAddrValidFlag |
Check if New device address has been assigned
| usbEndptRequest_t* usbMusbDcdDevice::inEpReq |
Current endpoint request for in transfer
| char usbMusbDcdDevice::name[10] |
name of the device controller
| usbEndptRequest_t* usbMusbDcdDevice::outEpReq |
Current endpoint request for out transfer
| usbDcd_t* usbMusbDcdDevice::pDcdCore |
Pointer to the DCD code, used for callbacks to the core
| uint32_t usbMusbDcdDevice::phyBaseAddr |
Base address of USB PHY control registers
| uint32_t usbMusbDcdDevice::uiSubBaseAddr |
Base address of the USB wrapper/controller


Autumn in New York (2000) 1 star out of 4. Starring Richard Gere, Winona Ryder, Anthony LaPaglia, Elaine Stritch, Vera Farmiga, Sherry Stringfield, Jill Hennessy and J.K. Simmons. Written by Allison Burnett. Directed by Joan Chen. Rated PG-13.
I must confess that underneath this gruff, cynical movie critic's exterior beats the heart of a sentimental romantic.
I enjoy love stories - good ones, that is. One of my favorites is a forgotten little gem called "The Barretts of Wimpole Street," which tells of the love affair of Victorian-era poets Robert Browning and Elizabeth Barrett. And, of course, there's "Casablanca," the greatest love story of all. When Rick sends Ilsa away at the end, telling her, `We'll always have Paris,' I get a bit choked up.
In my mind, the saddest love story is "King Kong." Boy finds girl, boy loses girl, boy finds girl again in New York, boy gets riddled with machine gun bullets and falls from the top of the Empire State Building. It brings a tear to my eye just thinking about it.
Bringing tears to my eyes, but for a different reason, was the new Richard Gere-Winona Ryder weeper, "Autumn in New York." This May-December romance is so artificial, so contrived that no matter how hard you try liking it, you can't bring yourself to embrace it fully.
He is Will Keane, 48, a famed chef and restaurant owner and a notorious womanizer. He's one of those love ‘em and leave ‘em guys. He is portrayed by Gere, who remains one of the handsomest men on the planet. In fact, age has added a bit of character and depth to Gere's features.
But the way in which some of the other supporting players talk about Gere's Keane and his remarkable good looks for a guy his age, you'd think he had a portrait of himself hidden somewhere in his apartment doing the aging for him. (For those who don't know what the heck I'm talking about, I suggest you read Oscar Wilde's "Picture of Dorian Gray," or at least rent the movie.)
She is Charlotte Fielding, a 22-year-old free spirit in the bloom of young adulthood. She also has a terrible secret. Well, it's not really a secret. She has a rare, congenital heart condition, and Charlotte knows she has very little time left. And if anyone thinks, I'm spoiling the movie, forget it. The film's trailer spills the beans, so don't go blaming me.
Anyway, our Charlotte figures, what the heck, why not spend what time she has left rolling around in bed with a guy twice her age.
Call me old fashioned, but if I knew my time was limited, I think I'd find other worthwhile things to do - visit ballparks I haven't seen, fly to countries I've always wanted to visit. Beg Sophia Loren to allow me to take her to dinner. That sort of thing.
Surprising, Ryder is very good as Charlotte. Her character is pragmatic and confident, and in some instances much more mature than her much-older lover. Because Charlotte seems so self-assured, so together, it is difficult to grasp her attraction to the shallow Will.
"Autumn in New York" contains some of the most lame dialogue I've heard in years. You don't even hear lines like these on daytime soaps.
Most of these clunkers are mouthed by Gere: `Man, you don't dance, you float' and `You're a kid and I'm a creep,' are two examples.
And those are some of the better lines. Gere and Ryder, under the direction of former actress Joan Chen, spend a lot of time staring into each other's eyes, crying, arguing then making up.
After a while, it is all very tiresome.
"Autumn in New York" is a pretty movie, with an appropriately sweet score by Gabriel Yared.
It is difficult to understand what attracted Chen, who made a very impressive directorial debut with the acclaimed "Xiu Xiu: The Sent-Down Girl," to this project. I hope it was only to get her foot in the door of the Directors Guild of America, and that now she will go on to better and more worthwhile projects as befits her talent.
For that is all "Autumn in New York" is good for - being a stepping stone.
Bob Bloom is the film critic at the Journal and Courier. in Lafayette, IN He can be reached by e-mail at<EMAIL_ADDRESS>or at<EMAIL_ADDRESS>Reviews by Bloom can be found on the Internet Movie Database at: http://www.imdb.com/M/reviews_by?Bob+Bloom
The review above was posted to the
rec.arts.movies.reviews newsgroup (de.rec.film.kritiken for German reviews).
The Internet Movie Database accepts no responsibility for the contents of the
review and has no editorial control. Unless stated otherwise, the copyright
belongs to the author.
Please direct comments/criticisms of the review to relevant newsgroups.
Broken URLs inthe reviews are the responsibility of the author.
The formatting of the review is likely to differ from the original due
to ASCII to HTML conversion.
Related links: index of all rec.arts.movies.reviews reviews


lab07 : Advanced arrays and linked lists
| num | ready? | description | assigned | due |
|---|---|---|---|---|
| lab07 | false | Advanced arrays and linked lists | Tue 05/22 10:00AM | Tue 05/29 11:59PM |
Goals of this lab
The goal of this lab is get more practice with iterating through arrays and linked lists, and applying recursion to solving problems. Continue to practice code tracing to reason about your code. We request that you DO NOT ask the staff to debug your code. They have been specifically instructed not to debug for you, rather to guide in the process.
Step by Step Instructions: PLEASE READ CAREFULLY!
Step 1: Getting Started
-
Decide if you are working alone, or working in a pair. Pair programming is OPTIONAL for this lab.
-
If you are working as a pair, go to submit.cs, navigate to this lab page and create a team for you and your pair partner. Choose who will be the first driver and who will start as navigator, and then remember to switch (at least once) during the lab.
-
Go to github and create a git repo for this lab following the naming convention specified in previous labs (this step carries style points, see our feedback on previous labs to understand what we are looking for). If you are working with a partner only one of you needs to create the repo.
-
If you are working with a partner and you are the one who created the github repo, add your partner as a collborator on the repo
-
Decide on initial navigator and driver.
-
Driver, log on to your CSIL account.
-
Open a terminal window and log into the correct machine.
-
Change into your CS 16 directory
Note: Remember to push your work to github at the end of EVERY work session. That way, both partners always have access to the latest version of the code even if the code is being developed on one partner’s CoE account.
Step 2: Obtaining the starter code
This step is similar to lab02, first open terminal and go to the directory where you cloned the starter code in lab02 and pull the latest version of the starter code.
cd ~/cs16/cs16-w18-starter-code
git pull
Clone your github repo in the ~/cs16/ directory. Then cd into your repo directory.
cd ../lab07_gaucho_ally
Copy the code from your starter code directory to your local lab07 repo using the following command.
cp ~/cs16/cs16-w18-starter-code/lab07/* ./
Typing the list (ls) command should show you the following files in your current directory
[dimirza@csil-03 lab07-startercode]$ ls
addIntToEndOfListTest.cpp mafTest4.cpp mllfTest7.cpp
arrayFuncs.h Makefile moreArrayFuncs.cpp
linkedListFuncs.cpp mllfTest1.cpp moreArrayFuncs.h
linkedListFuncs.h mllfTest2.cpp moreLinkedListFuncs.cpp
linkedList.h mllfTest3.cpp moreLinkedListFuncs.h
mafTest1.cpp mllfTest4.cpp README.md
mafTest2.cpp mllfTest5.cpp tddFuncs.cpp
mafTest3.cpp mllfTest6.cpp tddFuncs.h
[dimirza@csil-03 lab07-startercode]$
Step 3: Reviewing the files and what your tasks are
Here is a list of your tasks for this lab:
Step 3a: Familiarize yourself with the big picture
Type “make tests” and you will see some tests pass, but some fail.
You are finished when all the tests pass.
There are only two files that you need to edit this week, and both are based on functions that you worked with in previous labs.
-
moreArrayFuncs.cpp
contains more functions that deal with arrays. These are in addition to the arrayFuncs.cpp that you worked with before in lab04. -
moreLinkedListFuncs.cpp
contains more functions that deal with linked lists. These are in addition to the linkedListFuncs.cpp that you worked with before in lab06.
Step 3b: Work on the array functions first
Even if you only get these working, this will be an important step in terms of partial credit for this lab. So concentrate on these first.
There are four files that run test cases for the functions in moreArrayFuncs.cpp. These are shown in the following table. In each case, you should work on one file at a time. For example, your first step is:
- type
make mafTest1
to compile and link mafTest1.cpp - type
./mafTest1
to run the program that testsindexOfMax
andindexOfMin
from moreArrayFuncs.cpp - edit the code in the file moreArrayFuncs.cpp so that
indexOfMax
andindexOfMin
are correct. - repeat the steps above until the tests in mafTest1.cpp pass.
- Make sure to push your code to github as you make progress on each part.
Then do the same for the other lines in this table, i.e.. tests mafTest2.cpp through mafTest4.cpp
| file | command to compile | command to run | what it tests |
|---|---|---|---|
| mafTest1.cpp | make mafTest1 | ./mafTest1 | indexOfMax and indexOfMin from moreArrayFuncs.cpp |
| mafTest2.cpp | make mafTest2 | ./mafTest2 | largestValue and smallestValue from moreArrayFuncs.cpp |
| mafTest3.cpp | make mafTest3 | ./mafTest3 | sum from moreArrayFuncs.cpp |
| mafTest4.cpp | make mafTest4 | ./mafTest4 | copyElements , copyOddOnly (*see tip box below), and multiplyPairwise from moreArrayFuncs.cpp |
Then, submit your code to submit.cs and see if the tests for the array functions are at least passing on submit.cs. The command for that is listed in step 4 of the lab.
When your array tests pass, you can move on to the linked list part. Or, you can go ahead and move on, and wait to ask for help from a TA.
Step 3c: COPY A FILE FROM YOUR COMPLETED lab06
The next step is crucial. You need a function from your completed lab06 in order for your lab07 to work.
That function is in the file linkedListFuncs.cpp.
So, you need to use a Unix command to copy the file linkedListFuncs.cpp from your lab06 github directory to your lab07 github directory.
I am NOT going to tell you what that command is. By now, you should know. And if you don’t, you should be able to look it up. Also I may ask you about this on the final exam.
Once you’ve copied that over, push your code to github. Now you are ready for the next step.
Step 3d: Work on the linked list functions next
Working on the linked list functions below is one of the most important things you can do to prepare for the final exam.
There are seven files that run tests cases for the functions in moreLinkedListFuncs.cpp.
Proceed as you did for the four mafTest1.cpp through mafTest4.cpp files.
| file | command to compile | command to run | what it tests |
|---|---|---|---|
| mllfTest1.cpp | make mllfTest1 | ./mllfTest1 | pointerToMax from moreLinkedListFuncs.cpp |
| mllfTest2.cpp | make mllfTest2 | ./mllfTest2 | pointerToMin from moreLinkedListFuncs.cpp |
| mllfTest3.cpp | make mllfTest3 | ./mllfTest3 | largestValue from moreLinkedListFuncs.cpp |
| mllfTest4.cpp | make mllfTest4 | ./mllfTest4 | smallestValue from moreLinkedListFuncs.cpp |
| mllfTest5.cpp | make mllfTest5 | ./mllfTest5 | sum from moreLinkedListFuncs.cpp |
| mllfTest6.cpp | make mllfTest6 | ./mllfTest6 | deleteNodeIteratively; deleteNodeRecursively from moreLinkedListFuncs.cpp |
| mllfTest7.cpp | make mllfTest7 | ./mllfTest7 | insertNodeToSortedList from moreLinkedListFuncs.cpp |
Note: current output of mllfTest6 and mllfTest7 are INCORRECT. You should first write your own test code and implement the functions (deleteNodeIteratively, deleteNodeRecursivelyHelper; insertNodeToSortedList) in moreLinkedListFuncs.cpp to make them work correctly! See next section for instructions
Step 3e:
Add the following function declarations to moreLinkedListFuncs.h
. Then add test code in mllfTesr6.cpp to test the first two functions. Add test code in mllfTest7.cpp to test insertNodeToSortedList
. Finally implement all three functions in moreLinkedListFuncs.cpp as described below:
// list: ptr to a linked list of Node (each with int data, and Node * next)
// value: int value of the node to delete
// Delete all instances of Nodes with the given value in a recursive way.
// Use a helper function that performs the recursion
void deleteNodeRecursively(LinkedList * list, int value);
// list: ptr to a linked list of Node (each with int data, and Node * next)
// value: int value of the node to delete
// Delete the instances of all Node with the given value using loops.
void deleteNodeIteratively(LinkedList * list, int value);
// list: ptr to a linked list of Node (each with int data, and Node * next)
// Nodes in the linked list are expected to be in sorted ascending order
// value: int value of the node to insert
// Insert a new node in the linked list with the given value at the right spot
// The final linked list should have all its nodes in sorted ascending order
// Your implmentation should allow duplicate values in the list
void insertNodeToSortedList(LinkedList * list, int value);
We will test your code for these functions our own test cases, so don’t try to hard code it! Note that these functions are more challenging than the other functions in this lab, so think them through carefully before you start coding.
Step 4: Checking your work before submitting
When you are finished, you should be able to type make clean
and then make tests
and see the following output:
-bash-4.2$ make clean
/bin/rm -f mafTest1 mafTest2 mafTest3 mafTest4 addIntToEndOfListTest mllfTest1 mllfTest2 mllfTest3 mllfTest4 mllfTest5 mllfTest6 mllfTest7 *.o
-bash-4.2$ make tests
clang++ -Wall -Wno-uninitialized -c -o mafTest1.o mafTest1.cpp
clang++ -Wall -Wno-uninitialized -c -o linkedListFuncs.o linkedListFuncs.cpp
clang++ -Wall -Wno-uninitialized -c -o moreArrayFuncs.o moreArrayFuncs.cpp
clang++ -Wall -Wno-uninitialized -c -o moreLinkedListFuncs.o moreLinkedListFuncs.cpp
clang++ -Wall -Wno-uninitialized -c -o tddFuncs.o tddFuncs.cpp
clang++ -Wall -Wno-uninitialized mafTest1.o linkedListFuncs.o moreArrayFuncs.o moreLinkedListFuncs.o tddFuncs.o -o mafTest1
clang++ -Wall -Wno-uninitialized -c -o mafTest2.o mafTest2.cpp
clang++ -Wall -Wno-uninitialized mafTest2.o linkedListFuncs.o moreArrayFuncs.o moreLinkedListFuncs.o tddFuncs.o -o mafTest2
clang++ -Wall -Wno-uninitialized -c -o mafTest3.o mafTest3.cpp
clang++ -Wall -Wno-uninitialized mafTest3.o linkedListFuncs.o moreArrayFuncs.o moreLinkedListFuncs.o tddFuncs.o -o mafTest3
clang++ -Wall -Wno-uninitialized -c -o mafTest4.o mafTest4.cpp
clang++ -Wall -Wno-uninitialized mafTest4.o linkedListFuncs.o moreArrayFuncs.o moreLinkedListFuncs.o tddFuncs.o -o mafTest4
clang++ -Wall -Wno-uninitialized -c -o addIntToEndOfListTest.o addIntToEndOfListTest.cpp
clang++ -Wall -Wno-uninitialized addIntToEndOfListTest.o linkedListFuncs.o moreArrayFuncs.o moreLinkedListFuncs.o tddFuncs.o -o addIntToEndOfListTest
clang++ -Wall -Wno-uninitialized -c -o mllfTest1.o mllfTest1.cpp
clang++ -Wall -Wno-uninitialized mllfTest1.o linkedListFuncs.o moreArrayFuncs.o moreLinkedListFuncs.o tddFuncs.o -o mllfTest1
clang++ -Wall -Wno-uninitialized -c -o mllfTest2.o mllfTest2.cpp
clang++ -Wall -Wno-uninitialized mllfTest2.o linkedListFuncs.o moreArrayFuncs.o moreLinkedListFuncs.o tddFuncs.o -o mllfTest2
clang++ -Wall -Wno-uninitialized -c -o mllfTest3.o mllfTest3.cpp
clang++ -Wall -Wno-uninitialized mllfTest3.o linkedListFuncs.o moreArrayFuncs.o moreLinkedListFuncs.o tddFuncs.o -o mllfTest3
clang++ -Wall -Wno-uninitialized -c -o mllfTest4.o mllfTest4.cpp
clang++ -Wall -Wno-uninitialized mllfTest4.o linkedListFuncs.o moreArrayFuncs.o moreLinkedListFuncs.o tddFuncs.o -o mllfTest4
clang++ -Wall -Wno-uninitialized -c -o mllfTest5.o mllfTest5.cpp
clang++ -Wall -Wno-uninitialized mllfTest5.o linkedListFuncs.o moreArrayFuncs.o moreLinkedListFuncs.o tddFuncs.o -o mllfTest5
clang++ -Wall -Wno-uninitialized -c -o mllfTest6.o mllfTest6.cpp
clang++ -Wall -Wno-uninitialized mllfTest6.o linkedListFuncs.o moreArrayFuncs.o moreLinkedListFuncs.o tddFuncs.o -o mllfTest6
clang++ -Wall -Wno-uninitialized -c -o mllfTest7.o mllfTest7.cpp
clang++ -Wall -Wno-uninitialized mllfTest7.o linkedListFuncs.o moreArrayFuncs.o moreLinkedListFuncs.o tddFuncs.o -o mllfTest7
./mafTest1
PASSED: arrayToString(fiveThrees,5)
PASSED: arrayToString(zeros,3)
PASSED: arrayToString(empty,0)
PASSED: arrayToString(primes,5)
PASSED: arrayToString(primes,10)
PASSED: arrayToString(meaning,1)
PASSED: arrayToString(mix1,10)
PASSED: arrayToString(mix2,10)
PASSED: arrayToString(descending,5)
PASSED: indexOfMax(fiveThrees,5)
PASSED: indexOfMax(zeros,3)
PASSED: indexOfMax(primes,1)
PASSED: indexOfMax(primes,5)
PASSED: indexOfMax(primes,10)
PASSED: indexOfMax(meaning,1)
PASSED: indexOfMax(mix1,10)
PASSED: indexOfMax(mix2,10)
PASSED: indexOfMax(mix1,3)
PASSED: indexOfMax(mix2,3)
PASSED: indexOfMax(mix2,1)
PASSED: indexOfMax(mix2,5)
PASSED: indexOfMin(fiveThrees,5)
PASSED: indexOfMin(zeros,3)
PASSED: indexOfMin(primes,5)
PASSED: indexOfMin(primes,10)
PASSED: indexOfMin(meaning,1)
PASSED: indexOfMin(mix1,10)
PASSED: indexOfMin(mix2,10)
PASSED: indexOfMin(mix1,3)
PASSED: indexOfMin(mix2,3)
PASSED: indexOfMin(descending,5)
PASSED: indexOfMin(descending,1)
./mafTest2
PASSED: largestValue(fiveThrees,5)
PASSED: largestValue(zeros,3)
PASSED: largestValue(primes,5)
PASSED: largestValue(primes,10)
PASSED: largestValue(meaning,1)
PASSED: largestValue(mix1,10)
PASSED: largestValue(mix2,10)
PASSED: largestValue(mix1,3)
PASSED: largestValue(mix2,3)
PASSED: largestValue(descending,5)
PASSED: largestValue(descending,1)
PASSED: smallestValue(fiveThrees,5)
PASSED: smallestValue(zeros,3)
PASSED: smallestValue(primes,5)
PASSED: smallestValue(primes,10)
PASSED: smallestValue(meaning,1)
PASSED: smallestValue(mix1,10)
PASSED: smallestValue(mix2,10)
PASSED: smallestValue(mix1,3)
PASSED: smallestValue(mix2,3)
PASSED: smallestValue(descending,5)
PASSED: smallestValue(descending,1)
./mafTest3
PASSED: arrayToString(fiveThrees,5)
PASSED: arrayToString(zeros,3)
PASSED: arrayToString(empty,0)
PASSED: arrayToString(primes,5)
PASSED: arrayToString(primes,10)
PASSED: arrayToString(meaning,1)
PASSED: arrayToString(mix1,10)
PASSED: arrayToString(mix2,10)
PASSED: arrayToString(descending,5)
PASSED: sum(fiveThrees,5)
PASSED: sum(zeros,3)
PASSED: sum(primes,5)
PASSED: sum(primes,10)
PASSED: sum(meaning,1)
PASSED: sum(mix1,10)
PASSED: sum(mix2,10)
PASSED: sum(mix1,3)
PASSED: sum(mix2,3)
PASSED: sum(descending,5)
PASSED: sum(descending,1)
./mafTest4
PASSED: arrayToString(fiveThrees,5)
PASSED: arrayToString(zeros,3)
PASSED: arrayToString(empty,0)
PASSED: arrayToString(primes,5)
PASSED: arrayToString(primes,10)
PASSED: arrayToString(meaning,1)
PASSED: arrayToString(mix1,10)
PASSED: arrayToString(mix2,10)
PASSED: arrayToString(descending,5)
PASSED: arrayToString(primes,10)
PASSED: arrayToString(mix1,10)
PASSED: arrayToString(mix1,10)
PASSED: arrayToString(mix2,10)
PASSED: arrayToString(mix1,10)
PASSED: arrayToString(mix2,10)
PASSED: arrayToString(descending,5)
PASSED: copyOddOnly(a,descending,5)
PASSED: arrayToString(descending,5)
PASSED: arrayToString(a,3)
PASSED: copyOddOnly(a,mix2,10)
PASSED: arrayToString(mix2,10)
PASSED: arrayToString(a,5)
PASSED: arrayToString(fiveThrees,5)
PASSED: arrayToString(zeros,3)
PASSED: arrayToString(primes,5)
PASSED: arrayToString(descending,5)
PASSED: arrayToString(fiveThrees,5)
PASSED: arrayToString(descending,5)
PASSED: arrayToString(a,5)
PASSED: arrayToString(primes,5)
PASSED: arrayToString(descending,5)
PASSED: arrayToString(a,4)
PASSED: arrayToString(primes,7)
PASSED: arrayToString(a,7)
./addIntToEndOfListTest
PASSED: linkedListToString(list)
PASSED: list->head->data == 42
PASSED: list->tail->data == 25
PASSED: list after adding 25
PASSED: list->head->data == 42
PASSED: list->tail->data == 31
PASSED: list after adding 31
PASSED: list->head->data == NULL
PASSED: list->tail->data == NULL)
PASSED: linkedListToString(emptyList)
PASSED:
PASSED: list->head->data == 7
PASSED:
PASSED: list->tail->data == 7)
PASSED: list after adding 7
PASSED: list != NULL
PASSED: list->head == list->tail
PASSED: list after adding -6
PASSED:
PASSED: list->head->data == 7
PASSED:
PASSED: list->tail->data == -6)
./mllfTest1
PASSED: linkedListToString(list)
Testing pointerToMax
PASSED: p!=NULL
PASSED: p->data==61
PASSED: p->next==NULL
PASSED: p==list->tail
PASSED: list->head->data == 42
PASSED: list->tail->data == 25
PASSED: p!=NULL
PASSED: p->data==61
PASSED: p->next->data==25
PASSED: p->next==list->tail
PASSED: list after adding 25
PASSED: list->head->data == 42
PASSED: list->tail->data == 99
PASSED: list after adding 99
PASSED: p!=NULL
PASSED: p->data==99
PASSED: p->next==NULL
PASSED: p==list->tail
./mllfTest2
PASSED: linkedListToString(list)
Testing pointerToMin
PASSED: p!=NULL
PASSED: p->data==42
PASSED: p==list->head
PASSED: list->head->data == 42
PASSED: list->tail->data == 25
PASSED: p!=NULL
PASSED: p->data==25
PASSED: p->next==NULL
PASSED: p==list->tail
PASSED: list after adding 25
PASSED: list->head->data == 42
PASSED: list->tail->data == 99
PASSED: list after adding 99
PASSED: p!=NULL
PASSED: p->data==25
PASSED: p->next==list->tail
./mllfTest3
PASSED: linkedListToString(list)
Testing pointerToMax
PASSED: p!=NULL
PASSED: largestValue(list)
PASSED: p!=NULL
PASSED: largestValue(list)
PASSED: list after adding 25
PASSED: list after adding 99
PASSED: p!=NULL
PASSED: largestValue(list)
./mllfTest4
PASSED: linkedListToString(list)
Testing pointerToMin
PASSED: p!=NULL
PASSED: p->data==42
PASSED: p==list->head
PASSED: list->head->data == 42
PASSED: list->tail->data == 25
PASSED: p!=NULL
PASSED: p->data==25
PASSED: p->next==NULL
PASSED: p==list->tail
PASSED: list after adding 25
PASSED: list->head->data == 42
PASSED: list->tail->data == 99
PASSED: list after adding 99
PASSED: p!=NULL
PASSED: p->data==25
PASSED: p->next==list->tail
./mllfTest5
PASSED: sum(&emptyList)
PASSED: linkedListToString(list)
PASSED: sum(list)
PASSED: list after adding 25
PASSED: sum(list)
./mllfTest6
./mllfTest7
-bash-4.2$
Note: You must write your own test code in mllfTest6.cpp and mllfTest7.cpp and make sure you pass those tests
At that point, you are ready to try submitting on the submit.cs system.
Step 5: Submitting via submit.cs
Here is the command to submit this week’s labs:
~submit/submit -p 927 *.cpp *.h
Grading Rubric
Some of the points will be awarded based on submit.cs automatic grading. Other points will be assigned after visual code inspection by TAs.
Submit.cs system automatic points
| Test Group | Test Name | Value |
|---|---|---|
| mafTest1 | mafTest1 | (20 pts) |
| mafTest2 | ./mafTest2 | (20 pts) |
| mafTest3 | ./mafTest3 | (20 pts) |
| mafTest4 | ./mafTest4 | (20 pts) |
| mllfTest1 | ./mllfTest1 | (20 pts) |
| mllfTest2 | ./mllfTest2 | (20 pts) |
| mllfTest3 | ./mllfTest3 | (20 pts) |
| mllfTest4 | ./mllfTest4 | (20 pts) |
| mllfTest5 | ./mllfTest5 | (20 pts) |
| mllfTest6 | ./mllfTest6 | (20 pts) |
| mllfTest7 | ./mllfTest7 | (20 pts) |
Code inspection human-assigned points
- (30 pts) Submitting on time, per instructions
- (90 pts) Code style, including but not limited to:
- Code can be easily understood by humans familiar with C++ (including both the author(s) of the code, and non-authors of the code.)
- Code is neatly indented and formatted, following standard code indentation practices for C++ as illustrated in either the textbook, or example code given in lectures and labs
- Variable names choices are reasonable
- Code is reasonably “DRY” (as in “don’t repeat yourself”)—where appropriate, common code is factored out into functions
- Code is not unnecessarily or unreasonably complex when a simpler solution is available
An important word about academic honesty and the submit.cs system
We will test your code against other data files too—not just these. So while you might be able to pass the tests on submit.cs now by just doing a hard-coded “cout” of the expected output, that will NOT receive credit.
To be very clear, code like this will pass on submit.cs, BUT REPRESENTS A FORM OF ACADEMIC DISHONESTY since it is an attempt to just “game the system”, i.e. to get the tests to pass without really solving the problem.
I would hope this would be obvious, but I have to say it so that there is no ambiguity: hard coding your output is a form of cheating, i.e. a form of “academic dishonesty”. Submitting a program of this kind would be subject not only to a reduced grade, but to possible disciplinary penalties. If there is any doubt about this fact, please ask your TA and/or your instructor for clarification.


Going to home community solo + child this evening. No naps in afternoon == Mom needs some downtime.
1/28/2009, 6:22:04 PM
Favs: 0
Retweets: 0
link--- layout: tweet ---
Going to home community solo + child this evening. No naps in afternoon == Mom needs some downtime.
1/28/2009, 6:22:04 PM
Favs: 0
Retweets: 0
link


Implements methods described by the paper Robins and Tsiatis (1991) <doi:10.1080/03610929108830654>. These use g-estimation to estimate the causal effect of a treatment in a two-armed randomised control trial where non-compliance exists and is measured, under an assumption of an accelerated failure time model and no unmeasured confounders.
| Version: | 1.2.3 |
| Depends: | R (≥ 2.10) |
| Imports: | survival, rootSolve, ggplot2, stats |
| Suggests: | testthat, knitr, rmarkdown, tableone |
| Published: | 2018-08-14 |
| Author: | Simon Bond [aut, cre] (primary author of code, secondary author of vignette), Annabel Allison [aut] (primary author of vignette, secondary author of code) |
| Maintainer: | Simon Bond <simon.bond at addenbrookes.nhs.uk> |
| License: | GPL-2 |
| NeedsCompilation: | no |
| Materials: | README NEWS |
| CRAN checks: | rpsftm results |
| Reference manual: | rpsftm.pdf |
| Vignettes: |
rpsftm: rank-preserving structural failure time models for survival data |
| Package source: | rpsftm_1.2.3.tar.gz |
| Windows binaries: | r-devel: rpsftm_1.2.3.zip, r-release: rpsftm_1.2.3.zip, r-oldrel: rpsftm_1.2.3.zip |
| OS X binaries: | r-release: rpsftm_1.2.3.tgz, r-oldrel: rpsftm_1.2.3.tgz |
| Old sources: | rpsftm archive |
Please use the canonical form https://CRAN.R-project.org/package=rpsftm to link to this page.


Preparation time
overnight
Cooking time
10 to 30 mins
Serves
Serves 10-12
Dietary
Take a masterclass in impressive desserts with this chocolate and cream-filled charlotte topped with fancy spun sugar.
For the stock syrup, mix the sugar and water together in a small pan, then bring the mixture slowly to the boil until the sugar dissolves. Reduce the heat until the mixture is simmering and simmer for 4-5 minutes. Remove from heat and set aside to cool, then chill in the fridge overnight.
For the charlotte, place the dried sour cherries and the dried cranberries in a small bowl, pour over six tablespoons of the orange liqueur, cover the bowl with cling film and leave the fruit to soak for 12 hours, or at least overnight.
When the fruit has soaked, grease and line a deep 22cm/9in loose-bottomed cake tin (or charlotte mould) with greaseproof paper.
Pour the remaining orange liqueur and the stock syrup into a wide, shallow bowl.
Dip the non-sugared sides of the sponge fingers into the syrup mixture one at a time and stand them up vertically next to each other to line the sides of the cake tin. Don’t leave them to soak for too long as they will fall apart.
Bring a small amount of water to a gentle simmer in a saucepan. Suspend a heatproof bowl over the pan (do not allow the base of the bowl to touch the surface of the water). Add the chocolate to the bowl and stir until melted. Set aside.
Using a hand-held electric whisk, cream the butter and sugar together in a bowl until pale and fluffy.
Fold the chestnut purée into the creamed butter and sugar mixture until just combined, using a metal spoon. Pour in the melted chocolate, then add the soaked fruit and ground almonds. Mix until well combined.
Pour the cream into a separate bowl and whisk until it is thick enough to leave a trail behind when the whisk is removed. Fold the whipped cream into the chocolate and fruit mixture until it gives a marbled effect.
Spoon the mixture into the cake tin, then cover the tin with cling film and chill in the fridge overnight.
For the spun sugar, tip the sugar into a heavy-based saucepan. Add the water and stir until the sugar has dissolved. Gradually bring the mixture to a simmer over a gentle heat.
Using a sugar thermometer, heat the mixture to a temperature of 155C/311F (this is known as the hard crack stage). Do not be tempted to stir the sugar mixture as this will cause the syrup to crystallise. (Caution: hot sugar syrup can burn. Do not leave unattended and do not taste or touch the syrup.)
Fill an empty sink with cold water. As soon as the sugar mixture reaches the desired temperature, plunge the base of the pan into the water to stop the cooking process.
Now work quickly: place the saucepan onto a heatproof surface. Turn two forks back to back and hold them in one hand. Hold a rolling pin in the other. Dip the forks into the sugar syrup and flick the caramel over the rolling pin to create long strands of sugar. Using your hands, gather the strands up into a large ball and mould together to form shapes. Continue to create sugar shapes until all of the sugar syrup has been used.
When ready to serve the sweet chestnut and chocolate charlotte, turn the charlotte out onto a serving plate or cake stand. Garnish with the spun sugar shapes.
You need JavaScript to view this clip.
Melting chocolate


Chelsea Handler, in full Chelsea Joy Handler, (born February 25, 1975, Livingston, New Jersey, U.S.), American comedian and author known for her earthy, bawdy style and her late-night talk show, Chelsea Lately (2007–14).
 Do I really need to say it? Do I need to say it again, yeah? You better stop the sweet talk and keep your pretty mouth shut.  Ay When I'm kissing you, gripping your thigh I realize you were destined and meant to be mine But who I am to conflict with you living your life?  I'm sorry by the way I'm never coming back down Can't you see I could but wouldn't stay? I wouldn't put it like that What do you mean? I'm sorry by the way I'm never coming around It'd be so sweet if Things just stayed the same La-da-da-da-da All the lights couldn't put out the dark Running through my heart Lights up and they know who you are Know who you are Do you know who you are? (Shine) Step into the light (Shine) So bright sometimes (Shine) I'm not ever going back


|<EMAIL_ADDRESS>| xnum@GitHub |
| Taichung, Taiwan |
Jia-Jun is a professional software engineer specializing in C++ and Go. He first worked for Taipei Exchange, where he had experience in classical financial system. After that, he joined a blockchain startup as a senior developer and became team leader in charge of cryptocurrency exchange. Now in order to build competitive product, he is bringing cutting-edge technology to FinTech startup.
Stranity Technology Co,. Ltd. Chief Technology Officer
spin-off
Sky Mirror Technology Co., Ltd. VP of Engineering
Futures trading system development using C++17.
COBINHOOD, Ltd. Lead Backend developer
Cryptocurrency exchange development.
Taipei Exchange Software Engineer
Development of an emerging stock trading system.