<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 def _do_extraction(df, column_id, column_value, column_kind, default_fc_parameters, kind_to_fc_parameters, n_jobs, chunk_size, disable_progressbar, distributor):
 """ Wrapper around the _do_extraction_on_chunk, which calls it on all chunks in the data frame. A chunk is a subset of the data, with a given kind and id - so a single time series. The data is separated out into those single time series and the _do_extraction_on_chunk is called on each of them. The results are then combined into a single pandas DataFrame. The call is either happening in parallel or not and is showing a progress bar or not depending on the given flags. :param df: The dataframe in the normalized format which is used for extraction. :type df: pd.DataFrame :param default_fc_parameters: mapping from feature calculator names to parameters. Only those names which are keys in this dict will be calculated. See the class:`ComprehensiveFCParameters` for more information. :type default_fc_parameters: dict :param kind_to_fc_parameters: mapping from kind names to objects of the same type as the ones for default_fc_parameters. If you put a kind as a key here, the fc_parameters object (which is the value), will be used instead of the default_fc_parameters. :type kind_to_fc_parameters: dict :param column_id: The name of the id column to group by. :type column_id: str :param column_kind: The name of the column keeping record on the kind of the value. :type column_kind: str :param column_value: The name for the column keeping the value itself. :type column_value: str :param chunk_size: The size of one chunk for the parallelization :type chunk_size: None or int :param n_jobs: The number of processes to use for parallelization. If zero, no parallelization is used. :type n_jobs: int :param disable_progressbar: Do not show a progressbar while doing the calculation. :type disable_progressbar: bool :param distributor: Advanced parameter: See the utilities/distribution.py for more information. Leave to None, if you want TSFresh to choose the best distributor. :type distributor: DistributorBaseClass :return: the extracted features :rtype: pd.DataFrame """


    data_in_chunks = generate_data_chunk_format(df, column_id, column_kind, column_value)

    if distributor is None:
        if n_jobs == 0:
            distributor = MapDistributor(disable_progressbar=disable_progressbar,
                                         progressbar_title="Feature Extraction")
        else:
            distributor = MultiprocessingDistributor(n_workers=n_jobs,
                                                     disable_progressbar=disable_progressbar,
                                                     progressbar_title="Feature Extraction")

    if not isinstance(distributor, DistributorBaseClass):
        raise ValueError("the passed distributor is not an DistributorBaseClass object")

    kwargs = dict(default_fc_parameters=default_fc_parameters,
                  kind_to_fc_parameters=kind_to_fc_parameters)

    result = distributor.map_reduce(_do_extraction_on_chunk, data=data_in_chunks,
                                    chunk_size=chunk_size,
                                    function_kwargs=kwargs)
    distributor.close()

    # Return a dataframe in the typical form (id as index and feature names as columns)
    result = pd.DataFrame(result)
    if result.columns.contains("value"):
        result["value"] = result["value"].astype(float)

    if len(result) != 0:
        result = result.pivot("id", "variable", "value")
        result.index = result.index.astype(df[column_id].dtype)

    return result