def shuffle_store_dask_partitions(ddf: dd.DataFrame, table: str, secondary_indices: List[str], metadata_version: int, partition_on: List[str], store_factory: StoreFactory, df_serializer: Optional[DataFrameSerializer], dataset_uuid: str, num_buckets: int, sort_partitions_by: List[str], bucket_by: Sequence[str]) -> da.Array:
    '\n    Perform a dataset update with dask reshuffling to control partitioning.\n\n    The shuffle operation will perform the following steps\n\n    1. Pack payload data\n\n       Payload data is serialized and compressed into a single byte value using\n       ``distributed.protocol.serialize_bytes``, see also ``pack_payload``.\n\n    2. Apply bucketing\n\n       Hash the column subset ``bucket_by`` and distribute the hashes in\n       ``num_buckets`` bins/buckets. Internally every bucket is identified by an\n       integer and we will create one physical file for every bucket ID. The\n       bucket ID is not exposed to the user and is dropped after the shuffle,\n       before the store. This is done since we do not want to guarantee at the\n       moment, that the hash function remains stable.\n\n    3. Perform shuffle (dask.DataFrame.groupby.apply)\n\n        The groupby key will be the combination of ``partition_on`` fields and the\n        hash bucket ID. This will create a physical file for every unique tuple\n        in ``partition_on + bucket_ID``. The function which is applied to the\n        dataframe will perform all necessary subtask for storage of the dataset\n        (partition_on, index calc, etc.).\n\n    4. Unpack data (within the apply-function)\n\n        After the shuffle, the first step is to unpack the payload data since\n        the follow up tasks will require the full dataframe.\n\n    5. Pre storage processing and parquet serialization\n\n        We apply important pre storage processing like sorting data, applying\n        final partitioning (at this time there should be only one group in the\n        payload data but using the ``MetaPartition.partition_on`` guarantees the\n        appropriate data structures kartothek expects are created.).\n        After the preprocessing is done, the data is serialized and stored as\n        parquet. The applied function will return an (empty) MetaPartition with\n        indices and metadata which will then be used to commit the dataset.\n\n    Returns\n    -------\n\n    A dask.Array holding relevant MetaPartition objects as values\n\n    '
    if (ddf.npartitions == 0):
        return ddf
    group_cols = partition_on.copy()
    if (num_buckets is None):
        raise ValueError('``num_buckets`` must not be None when shuffling data.')
    meta = ddf._meta
    meta[_KTK_HASH_BUCKET] = np.uint64(0)
    ddf = ddf.map_partitions(_hash_bucket, bucket_by, num_buckets, meta=meta)
    group_cols.append(_KTK_HASH_BUCKET)
    unpacked_meta = ddf._meta
    ddf = pack_payload(ddf, group_key=group_cols)
    ddf_grouped = ddf.groupby(by=group_cols)
    unpack = partial(_unpack_store_partition, secondary_indices=secondary_indices, sort_partitions_by=sort_partitions_by, table=table, dataset_uuid=dataset_uuid, partition_on=partition_on, store_factory=store_factory, df_serializer=df_serializer, metadata_version=metadata_version, unpacked_meta=unpacked_meta)
    return cast(da.Array, ddf_grouped.apply(unpack, meta=('MetaPartition', 'object')))
Perform a dataset update with dask reshuffling to control partitioning.

The shuffle operation will perform the following steps

1. Pack payload data

   Payload data is serialized and compressed into a single byte value using
   ``distributed.protocol.serialize_bytes``, see also ``pack_payload``.

2. Apply bucketing

   Hash the column subset ``bucket_by`` and distribute the hashes in
   ``num_buckets`` bins/buckets. Internally every bucket is identified by an
   integer and we will create one physical file for every bucket ID. The
   bucket ID is not exposed to the user and is dropped after the shuffle,
   before the store. This is done since we do not want to guarantee at the
   moment, that the hash function remains stable.

3. Perform shuffle (dask.DataFrame.groupby.apply)

    The groupby key will be the combination of ``partition_on`` fields and the
    hash bucket ID. This will create a physical file for every unique tuple
    in ``partition_on + bucket_ID``. The function which is applied to the
    dataframe will perform all necessary subtask for storage of the dataset
    (partition_on, index calc, etc.).

4. Unpack data (within the apply-function)

    After the shuffle, the first step is to unpack the payload data since
    the follow up tasks will require the full dataframe.

5. Pre storage processing and parquet serialization

    We apply important pre storage processing like sorting data, applying
    final partitioning (at this time there should be only one group in the
    payload data but using the ``MetaPartition.partition_on`` guarantees the
    appropriate data structures kartothek expects are created.).
    After the preprocessing is done, the data is serialized and stored as
    parquet. The applied function will return an (empty) MetaPartition with
    indices and metadata which will then be used to commit the dataset.

Returns
-------

A dask.Array holding relevant MetaPartition objects as values
def shuffle_store_dask_partitions(ddf: dd.DataFrame, table: str, secondary_indices: List[str], metadata_version: int, partition_on: List[str], store_factory: StoreFactory, df_serializer: Optional[DataFrameSerializer], dataset_uuid: str, num_buckets: int, sort_partitions_by: List[str], bucket_by: Sequence[str]) -> da.Array:
    '\n    Perform a dataset update with dask reshuffling to control partitioning.\n\n    The shuffle operation will perform the following steps\n\n    1. Pack payload data\n\n       Payload data is serialized and compressed into a single byte value using\n       ``distributed.protocol.serialize_bytes``, see also ``pack_payload``.\n\n    2. Apply bucketing\n\n       Hash the column subset ``bucket_by`` and distribute the hashes in\n       ``num_buckets`` bins/buckets. Internally every bucket is identified by an\n       integer and we will create one physical file for every bucket ID. The\n       bucket ID is not exposed to the user and is dropped after the shuffle,\n       before the store. This is done since we do not want to guarantee at the\n       moment, that the hash function remains stable.\n\n    3. Perform shuffle (dask.DataFrame.groupby.apply)\n\n        The groupby key will be the combination of ``partition_on`` fields and the\n        hash bucket ID. This will create a physical file for every unique tuple\n        in ``partition_on + bucket_ID``. The function which is applied to the\n        dataframe will perform all necessary subtask for storage of the dataset\n        (partition_on, index calc, etc.).\n\n    4. Unpack data (within the apply-function)\n\n        After the shuffle, the first step is to unpack the payload data since\n        the follow up tasks will require the full dataframe.\n\n    5. Pre storage processing and parquet serialization\n\n        We apply important pre storage processing like sorting data, applying\n        final partitioning (at this time there should be only one group in the\n        payload data but using the ``MetaPartition.partition_on`` guarantees the\n        appropriate data structures kartothek expects are created.).\n        After the preprocessing is done, the data is serialized and stored as\n        parquet. The applied function will return an (empty) MetaPartition with\n        indices and metadata which will then be used to commit the dataset.\n\n    Returns\n    -------\n\n    A dask.Array holding relevant MetaPartition objects as values\n\n    '
    if (ddf.npartitions == 0):
        return ddf
    group_cols = partition_on.copy()
    if (num_buckets is None):
        raise ValueError('``num_buckets`` must not be None when shuffling data.')
    meta = ddf._meta
    meta[_KTK_HASH_BUCKET] = np.uint64(0)
    ddf = ddf.map_partitions(_hash_bucket, bucket_by, num_buckets, meta=meta)
    group_cols.append(_KTK_HASH_BUCKET)
    unpacked_meta = ddf._meta
    ddf = pack_payload(ddf, group_key=group_cols)
    ddf_grouped = ddf.groupby(by=group_cols)
    unpack = partial(_unpack_store_partition, secondary_indices=secondary_indices, sort_partitions_by=sort_partitions_by, table=table, dataset_uuid=dataset_uuid, partition_on=partition_on, store_factory=store_factory, df_serializer=df_serializer, metadata_version=metadata_version, unpacked_meta=unpacked_meta)
    return cast(da.Array, ddf_grouped.apply(unpack, meta=('MetaPartition', 'object')))