B ٻd C@sdZddlmZddlmZddlmZddlmZddlmZddlm Z ddlm Z dd l m Z dd lmZdd lmZed d dZeddeddddZeddeddddZGdddejZedGdddZdS) z!Grouping dataset transformations.) dataset_ops)structured_function)nest) structure)dtypes)ops) tensor_spec)gen_experimental_dataset_ops) deprecation) tf_exportz"data.experimental.group_by_reducercsfdd}|S)aA transformation that groups elements and performs a reduction. This transformation maps element of a dataset to a key using `key_func` and groups the elements by key. The `reducer` is used to process each group; its `init_func` is used to initialize state for each group when it is created, the `reduce_func` is used to update the state every time an element is mapped to the matching group, and the `finalize_func` is used to map the final state to an output value. Args: key_func: A function mapping a nested structure of tensors (having shapes and types defined by `self.output_shapes` and `self.output_types`) to a scalar `tf.int64` tensor. reducer: An instance of `Reducer`, which captures the reduction logic using the `init_func`, `reduce_func`, and `finalize_func` functions. Returns: A `Dataset` transformation function, which can be passed to `tf.data.Dataset.apply`. cs t|S)zEFunction from `Dataset` to `Dataset` that applies the transformation.)_GroupByReducerDataset)dataset)key_funcreducerb/var/www/html/venv/lib/python3.7/site-packages/tensorflow/python/data/experimental/ops/grouping.py _apply_fn3sz#group_by_reducer.._apply_fnr)rrrr)rrrgroup_by_reducersrNz+Use `tf.data.Dataset.group_by_window(...)`.z!data.experimental.group_by_windowcsfdd}|S)aYA transformation that groups windows of elements by key and reduces them. This transformation maps each consecutive element in a dataset to a key using `key_func` and groups the elements by key. It then applies `reduce_func` to at most `window_size_func(key)` elements matching the same key. All except the final window for each key will contain `window_size_func(key)` elements; the final window may be smaller. You may provide either a constant `window_size` or a window size determined by the key through `window_size_func`. Args: key_func: A function mapping a nested structure of tensors (having shapes and types defined by `self.output_shapes` and `self.output_types`) to a scalar `tf.int64` tensor. reduce_func: A function mapping a key and a dataset of up to `window_size` consecutive elements matching that key to another dataset. window_size: A `tf.int64` scalar `tf.Tensor`, representing the number of consecutive elements matching the same key to combine in a single batch, which will be passed to `reduce_func`. Mutually exclusive with `window_size_func`. window_size_func: A function mapping a key to a `tf.int64` scalar `tf.Tensor`, representing the number of consecutive elements matching the same key to combine in a single batch, which will be passed to `reduce_func`. Mutually exclusive with `window_size`. Returns: A `Dataset` transformation function, which can be passed to `tf.data.Dataset.apply`. Raises: ValueError: if neither or both of {`window_size`, `window_size_func`} are passed. cs|jdS)zEFunction from `Dataset` to `Dataset` that applies the transformation.)r reduce_func window_sizewindow_size_func)group_by_window)r )rrrrrrrcs z"group_by_window.._apply_fnr)rrrrrr)rrrrrr:s)rz5Use `tf.data.Dataset.bucket_by_sequence_length(...)`.z+data.experimental.bucket_by_sequence_lengthFc sfdd}|S)aKA transformation that buckets elements in a `Dataset` by length. Elements of the `Dataset` are grouped together by length and then are padded and batched. This is useful for sequence tasks in which the elements have variable length. Grouping together elements that have similar lengths reduces the total fraction of padding in a batch which increases training step efficiency. Below is an example to bucketize the input data to the 3 buckets "[0, 3), [3, 5), [5, inf)" based on sequence length, with batch size 2. >>> elements = [ ... [0], [1, 2, 3, 4], [5, 6, 7], ... [7, 8, 9, 10, 11], [13, 14, 15, 16, 19, 20], [21, 22]] >>> dataset = tf.data.Dataset.from_generator( ... lambda: elements, tf.int64, output_shapes=[None]) >>> dataset = dataset.apply( ... tf.data.experimental.bucket_by_sequence_length( ... element_length_func=lambda elem: tf.shape(elem)[0], ... bucket_boundaries=[3, 5], ... bucket_batch_sizes=[2, 2, 2])) >>> for elem in dataset.as_numpy_iterator(): ... print(elem) [[1 2 3 4] [5 6 7 0]] [[ 7 8 9 10 11 0] [13 14 15 16 19 20]] [[ 0 0] [21 22]] There is also a possibility to pad the dataset till the bucket boundary. You can also provide which value to be used while padding the data. Below example uses `-1` as padding and it also shows the input data being bucketizied to two buckets "[0,3], [4,6]". >>> elements = [ ... [0], [1, 2, 3, 4], [5, 6, 7], ... [7, 8, 9, 10, 11], [13, 14, 15, 16, 19, 20], [21, 22]] >>> dataset = tf.data.Dataset.from_generator( ... lambda: elements, tf.int32, output_shapes=[None]) >>> dataset = dataset.apply( ... tf.data.experimental.bucket_by_sequence_length( ... element_length_func=lambda elem: tf.shape(elem)[0], ... bucket_boundaries=[4, 7], ... bucket_batch_sizes=[2, 2, 2], ... pad_to_bucket_boundary=True, ... padding_values=-1)) >>> for elem in dataset.as_numpy_iterator(): ... print(elem) [[ 0 -1 -1] [ 5 6 7]] [[ 1 2 3 4 -1 -1] [ 7 8 9 10 11 -1]] [[21 22 -1]] [[13 14 15 16 19 20]] When using `pad_to_bucket_boundary` option, it can be seen that it is not always possible to maintain the bucket batch size. You can drop the batches that do not maintain the bucket batch size by using the option `drop_remainder`. Using the same input data as in the above example you get the following result. >>> elements = [ ... [0], [1, 2, 3, 4], [5, 6, 7], ... [7, 8, 9, 10, 11], [13, 14, 15, 16, 19, 20], [21, 22]] >>> dataset = tf.data.Dataset.from_generator( ... lambda: elements, tf.int32, output_shapes=[None]) >>> dataset = dataset.apply( ... tf.data.experimental.bucket_by_sequence_length( ... element_length_func=lambda elem: tf.shape(elem)[0], ... bucket_boundaries=[4, 7], ... bucket_batch_sizes=[2, 2, 2], ... pad_to_bucket_boundary=True, ... padding_values=-1, ... drop_remainder=True)) >>> for elem in dataset.as_numpy_iterator(): ... print(elem) [[ 0 -1 -1] [ 5 6 7]] [[ 1 2 3 4 -1 -1] [ 7 8 9 10 11 -1]] Args: element_length_func: function from element in `Dataset` to `tf.int32`, determines the length of the element, which will determine the bucket it goes into. bucket_boundaries: `list`, upper length boundaries of the buckets. bucket_batch_sizes: `list`, batch size per bucket. Length should be `len(bucket_boundaries) + 1`. padded_shapes: Nested structure of `tf.TensorShape` to pass to `tf.data.Dataset.padded_batch`. If not provided, will use `dataset.output_shapes`, which will result in variable length dimensions being padded out to the maximum length in each batch. padding_values: Values to pad with, passed to `tf.data.Dataset.padded_batch`. Defaults to padding with 0. pad_to_bucket_boundary: bool, if `False`, will pad dimensions with unknown size to maximum length in batch. If `True`, will pad dimensions with unknown size to bucket boundary minus 1 (i.e., the maximum length in each bucket), and caller must ensure that the source `Dataset` does not contain any elements with length longer than `max(bucket_boundaries)`. no_padding: `bool`, indicates whether to pad the batch features (features need to be either of type `tf.sparse.SparseTensor` or of same shape). drop_remainder: (Optional.) A `tf.bool` scalar `tf.Tensor`, representing whether the last batch should be dropped in the case it has fewer than `batch_size` elements; the default behavior is not to drop the smaller batch. Returns: A `Dataset` transformation function, which can be passed to `tf.data.Dataset.apply`. Raises: ValueError: if `len(bucket_batch_sizes) != len(bucket_boundaries) + 1`. c s|jdS)N)element_length_funcbucket_boundariesbucket_batch_sizes padded_shapespadding_valuespad_to_bucket_boundary no_paddingdrop_remainder)bucket_by_sequence_length)r )rrrrrrrrrrrsz,bucket_by_sequence_length.._apply_fnr) rrrrrrrrrr)rrrrrrrrrr ns  r cs\eZdZdZfddZddZddZdd Zd d Ze d d Z ddZ ddZ Z S)r z;A `Dataset` that groups its input and performs a reduction.cs||_|||||j||j|||jtj |jj |j j j |jj j |jj j |jj j f|j j |jj |jj |jj d|j}tt|||dS)z%See `group_by_reducer()` for details.)r init_funcr finalize_funcN)Z_input_dataset_make_key_func_make_init_funcr!_make_reduce_funcr_make_finalize_funcr"ged_opsZ%experimental_group_by_reducer_datasetZ_variant_tensor _key_funcfunctionZcaptured_inputs _init_func _reduce_func_finalize_funcZ_flat_structuresuperr __init__)self input_datasetrrZvariant_tensor) __class__rrr.s"      z_GroupByReducerDataset.__init__cCsPtj|||d|_|jjtgtj sLt d|jj d|jj ddS)z!Make wrapping defun for key_func.)r ztInvalid `key_func`. Expected `key_func` to return a scalar tf.int64 tensor, but instead `key_func` has output types=z and shapes=.N) rStructuredFunctionWrapper_transformation_namer(output_structureZis_compatible_withr TensorSpecrint64 ValueError output_types output_shapes)r/rr0rrrr#s z%_GroupByReducerDataset._make_key_funccCs$tj||tgtjd|_dS)z"Make wrapping defun for init_func.)input_structureN)rr3r4rr6rr7r*)r/r!rrrr$(sz&_GroupByReducerDataset._make_init_funccCs|jj|_|jj}|jj}|jj}d}xX|rtj|||j|j fdd}xHt t |jt |D],\}} t || shtd|jd|jdqhWxHt t |jt |D],\} } | | krtd|jd|jjdqWt |} t |j} d d t | | D}d}xHt | |D]:\}}|jd k r|jd ksN||krd}PqW|r*t |jj|}t||||_q*W||_|jjtd S) z$Make wrapping defun for reduce_func.TF)r; add_to_graphzAInvalid `reducer`. The output class of the `reducer.reduce_func` z/, does not match the class of the reduce state r2z7Invalid `reducer`. The element types for the new state z1 do not match the element types of the old state cSsg|]\}}||qSr)Zmost_specific_compatible_shape).0originalnewrrr Zsz<_GroupByReducerDataset._make_reduce_func..N)r*r5_state_structurer9r:Zoutput_classesrr3r4 element_specziprflatten issubclass TypeErrorZ_state_classesZndimsas_listZpack_sequence_asrZconvert_legacy_structurer+r)r<rZget_default_graph)r/rr0Z state_typesZ state_shapesZ state_classesZ need_to_rerunZ wrapped_funcZnew_state_classZ state_classZnew_state_typeZ state_typeZflat_state_shapesZflat_new_state_shapesZweakened_state_shapesZoriginal_shapeZweakened_shaperrrr%/sR          z(_GroupByReducerDataset._make_reduce_funccCstj|||jd|_dS)z&Make wrapping defun for finalize_func.)r;N)rr3r4rAr,)r/r"rrrr&psz*_GroupByReducerDataset._make_finalize_funccCs|jjS)N)r,r5)r/rrrrBwsz#_GroupByReducerDataset.element_speccCs|j|j|j|jgS)N)r(r*r+r,)r/rrr _functions{sz!_GroupByReducerDataset._functionscCsdS)Nz'tf.data.experimental.group_by_reducer()r)r/rrrr4sz+_GroupByReducerDataset._transformation_name)__name__ __module__ __qualname____doc__r.r#r$r%r&propertyrBrHr4 __classcell__rr)r1rr s  A r zdata.experimental.Reducerc@s<eZdZdZddZeddZeddZedd Zd S) ReducerauA reducer is used for reducing a set of elements. A reducer is represented as a tuple of the three functions: - init_func - to define initial value: key => initial state - reducer_func - operation to perform on values with same key: (old state, input) => new state - finalize_func - value to return in the end: state => result For example, ``` def init_func(_): return (0.0, 0.0) def reduce_func(state, value): return (state[0] + value['features'], state[1] + 1) def finalize_func(s, n): return s / n reducer = tf.data.experimental.Reducer(init_func, reduce_func, finalize_func) ``` cCs||_||_||_dS)N)r*r+r,)r/r!rr"rrrr.szReducer.__init__cCs|jS)N)r*)r/rrrr!szReducer.init_funccCs|jS)N)r+)r/rrrrszReducer.reduce_funccCs|jS)N)r,)r/rrrr"szReducer.finalize_funcN) rIrJrKrLr.rMr!rr"rrrrrOs   rO)NN)NNFFF)rLZtensorflow.python.data.opsrrZtensorflow.python.data.utilrrZtensorflow.python.frameworkrrrZtensorflow.python.opsr r'Ztensorflow.python.utilr Z tensorflow.python.util.tf_exportr r deprecatedrr Z UnaryDatasetr rOrrrrs6           /