<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Rechunking
Rechunking lets us re-distribute how datasets are split between variables and chunks across a Beam PCollection.
To get started we'll recreate our dummy data from the data model tutorial
Step1:  Choosing chunks
Chunking can be essential for some operations. Some operations are very hard or impossible to perform with certain chunking schemes. For example, to make a plot all the data needs to come toether on a single machine. Other calculations such as calculating a median are possible to perform on distributed data, but require tricky algorithms and/or approximation.
More broadly, chunking can have critical performance implications, similar to those for Xarray and Dask. As a rule of thumb, chunk sizes of 10-100 MB work well. The optimal chunk size is a balance among a number of considerations, adapted here from Dask docs
Step2:  Adjusting chunks
You can also adjust chunks in a dataset to distribute arrays of different sizes. Here you have two choices of API
Step3:  Note that because these transformations only split or consolidate, they cannot necessary fully rechunk a dataset in a single step if the new chunk sizes are not multiples of old chunks (with consolidate) or do not even divide the old chunks (with split), e.g.,
Step4:  For such uneven cases, you'll need to use split followed by consolidate
Step5:  High level rechunking
Alternatively, the high-level Rechunk() method applies multiple split and consolidate steps based on the Rechunker algorithm
<ASSISTANT_TASK:>
Python Code:
import apache_beam as beam
import numpy as np
import xarray_beam as xbeam
import xarray
def create_records():
    for offset in [0, 4]:
        key = xbeam.Key({'x': offset, 'y': 0})
        data = 2 * offset + np.arange(8).reshape(4, 2)
        chunk = xarray.Dataset({
            'foo': (('x', 'y'), data),
            'bar': (('x', 'y'), 100 + data),
        })
        yield key, chunk
        
inputs = list(create_records())
Explanation: Rechunking
Rechunking lets us re-distribute how datasets are split between variables and chunks across a Beam PCollection.
To get started we'll recreate our dummy data from the data model tutorial:
End of explanation
inputs | xbeam.SplitVariables()
Explanation: Choosing chunks
Chunking can be essential for some operations. Some operations are very hard or impossible to perform with certain chunking schemes. For example, to make a plot all the data needs to come toether on a single machine. Other calculations such as calculating a median are possible to perform on distributed data, but require tricky algorithms and/or approximation.
More broadly, chunking can have critical performance implications, similar to those for Xarray and Dask. As a rule of thumb, chunk sizes of 10-100 MB work well. The optimal chunk size is a balance among a number of considerations, adapted here from Dask docs:
Chunks should be small enough to fit comfortably into memory on a single machine. As an upper limit, chunks over roughly 2 GB in size will not fit into the protocol buffers Beam uses to pass data between workers. 
There should be enough chunks for Beam runners (like Cloud Dataflow) to elastically shard work over many workers.
Chunks should be large enough to amortize the overhead of networking and the Python interpreter, which starts to become noticeable for arrays with fewer than 1 million elements.
The nbytes attribute on both NumPy arrays and xarray.Dataset objects is a good easy way to figure out how larger chunks are.
Adjusting variables
The simplest transformation is splitting (or consoldating) different variables in a Dataset with SplitVariables() and ConsolidateVariables(), e.g.,
End of explanation
inputs | xbeam.ConsolidateChunks({'x': -1})
Explanation: Adjusting chunks
You can also adjust chunks in a dataset to distribute arrays of different sizes. Here you have two choices of API:
The lower level {py:class}~xarray_beam.SplitChunks and {py:class}~xarray_beam.ConsolidateChunks. These transformations apply a single splitting (with indexing) or consolidation (with {py:function}xarray.concat) function to array elements.
The high level {py:class}~xarray_beam.Rechunk, which uses a pipeline of multiple split/consolidate steps (as needed) to efficiently rechunk a dataset.
Low level rechunking
For minor adjustments (e.g., mostly along a single dimension), the more explicit SplitChunks() and ConsolidateChunks() are good options. They take a dict of desired chunk sizes as a parameter, which can also be -1 to indicate "no chunking" along a dimension:
End of explanation
inputs | xbeam.SplitChunks({'x': 5})  # notice that the first two chunks are still separate!
Explanation: Note that because these transformations only split or consolidate, they cannot necessary fully rechunk a dataset in a single step if the new chunk sizes are not multiples of old chunks (with consolidate) or do not even divide the old chunks (with split), e.g.,
End of explanation
inputs | xbeam.SplitChunks({'x': 5}) | xbeam.ConsolidateChunks({'x': 5})
Explanation: For such uneven cases, you'll need to use split followed by consolidate:
End of explanation
inputs | xbeam.Rechunk(dim_sizes={'x': 6}, source_chunks={'x': 3}, target_chunks={'x': 5}, itemsize=8)
Explanation: High level rechunking
Alternatively, the high-level Rechunk() method applies multiple split and consolidate steps based on the Rechunker algorithm:
End of explanation
<END_TASK>