### Install and Use gist for Sharing Task Stream

Source: https://docs.dask.org/en/stable/diagnostics-distributed.html

This example shows how to install the `gist` utility and use it to upload the generated `task-stream.html` file, providing a sharable link.

```shell
$ python -m pip install gist
$ gist task-stream.html
https://gist.github.com/8a5b3c74b10b413f612bb5e250856ceb
```

--------------------------------

### Full Example: Setup Scheduler, Workers, and Client

Source: https://docs.dask.org/en/stable/_sources/deploying-python-advanced.rst

This comprehensive example demonstrates setting up a Scheduler, two Workers, and a Client within the same event loop, executing a simple computation, and then cleaning up all resources.

```python
import asyncio
from dask.distributed import Scheduler, Worker, Client

async def f():
    async with Scheduler() as s:
        async with Worker(s.address) as w1, Worker(s.address) as w2:
            async with Client(s.address, asynchronous=True) as client:
                future = client.submit(lambda x: x + 1, 10)
                result = await future
                print(result)

asyncio.get_event_loop().run_until_complete(f())
```

--------------------------------

### Start Worker and Wait for Completion

Source: https://docs.dask.org/en/stable/_sources/deploying-python-advanced.rst

This example illustrates how to start a Worker, which requires the scheduler's address. It then waits for the worker to finish, similar to the Scheduler example.

```python
import asyncio
from dask.distributed import Scheduler, Worker

async def f(scheduler_address):
    w = await Worker(scheduler_address)
    await w.finished()

asyncio.get_event_loop().run_until_complete(f("tcp://127.0.0.1:8786"))
```

--------------------------------

### Create a Dask Client

Source: https://docs.dask.org/en/stable/_sources/deploying-python.rst

Import and create a Client with no arguments to set up a local Dask scheduler. This is useful for getting started with Dask on your local machine.

```python
from dask.distributed import Client
client = Client()
```

--------------------------------

### Start Nanny Unsafely

Source: https://docs.dask.org/en/stable/_modules/distributed/nanny.html

Initiates the nanny, starts a local process, and begins watching. This method handles network setup, scheduler registration, and worker instantiation.

```python
async def start_unsafe(self):
        """Start nanny, start local process, start watching"""

        await super().start_unsafe()

        ports = parse_ports(self._start_port)
        for port in ports:
            start_address = address_from_user_args(
                host=self._start_host,
                port=port,
                interface=self._interface,
                protocol=self._protocol,
                security=self.security,
            )
            try:
                await self.listen(
                    start_address, **self.security.get_listen_args("worker")
                )
            except OSError as e:
                if len(ports) > 1 and e.errno == errno.EADDRINUSE:
                    continue
                else:
                    raise
            else:
                self._start_address = start_address
                break
        else:
            raise ValueError(
                f"Could not start Nanny on host {self._start_host} "
                f"with port {self._start_port}"
            )

        self.ip = get_address_host(self.address)

        await self.preloads.start()
        saddr = self.scheduler.addr
        comm = await self.rpc.connect(saddr)
        comm.name = "Nanny->Scheduler (registration)"

        try:
            await comm.write({"op": "register_nanny", "address": self.address})
            msg = await comm.read()
            try:
                for name, plugin in msg["nanny-plugins"].items():
                    await self.plugin_add(plugin=plugin, name=name)

                logger.info("        Start Nanny at: %r", self.address)
                response = await self.instantiate()

                if response != Status.running:
                    raise RuntimeError("Nanny failed to start worker process")
            except Exception:
                try:
                    await comm.write({"status": "error"})

                # If self.instantiate() failed, the comm will already be closed.
                except CommClosedError:
                    pass
                await self.close(reason="nanny-start-failed")
                raise
            else:
                await comm.write({"status": "ok"})
        finally:
            await comm.close()

        assert self.worker_address

        self.start_periodic_callbacks()

        return self

```

--------------------------------

### Start Scheduler and Wait for Completion

Source: https://docs.dask.org/en/stable/_sources/deploying-python-advanced.rst

This example shows how to create a Scheduler object, await its startup, and then wait for it to finish. The scheduler remains active until explicitly closed or an external process stops it.

```python
import asyncio
from dask.distributed import Scheduler, Worker

async def f():
    s = Scheduler()        # scheduler created, but not yet running
    s = await s            # the scheduler is running
    await s.finished()     # wait until the scheduler closes

asyncio.get_event_loop().run_until_complete(f())
```

--------------------------------

### Nanny.start_unsafe

Source: https://docs.dask.org/en/stable/_modules/distributed/nanny.html

Starts the nanny, including its local process and begins watching its status. This method handles network setup and scheduler registration.

```APIDOC
## Nanny.start_unsafe

### Description
Starts nanny, starts local process, starts watching.

### Method
async def start_unsafe(self)

### Returns
- The Nanny instance itself upon successful startup.
```

--------------------------------

### Use Context Managers for Scheduler and Worker

Source: https://docs.dask.org/en/stable/_sources/deploying-python-advanced.rst

This example demonstrates the use of `async with` context managers for automatically handling the setup and cleanup of a Scheduler and a Worker. It ensures resources are properly released.

```python
import asyncio
from dask.distributed import Scheduler, Worker

async def f():
    async with Scheduler() as s:
        async with Worker(s.address) as w:
            await w.finished()
            await s.finished()

asyncio.get_event_loop().run_until_complete(f())
```

--------------------------------

### Create DataFrame for Rounding Examples

Source: https://docs.dask.org/en/stable/generated/dask.dataframe.DataFrame.round.html

Initializes a pandas DataFrame with sample data for demonstrating the `round` method. This setup is common for testing rounding functionalities.

```python
import pandas as pd
df = pd.DataFrame(
    [(0.21, 0.32), (0.01, 0.67), (0.66, 0.03), (0.21, 0.18)],
    columns=["dogs", "cats"],
)
df
```

--------------------------------

### Start ProgressBar

Source: https://docs.dask.org/en/stable/_modules/dask/diagnostics/progress.html

Starts the progress bar by initializing state, recording the start time, and launching a background thread for updates.

```python
def _start(self, dsk):
    self._state = None
    self._start_time = default_timer()
    # Start background thread
    self._running = True
    self._timer = threading.Thread(target=self._timer_func)
    self._timer.daemon = True
    self._timer.start()
```

--------------------------------

### start_unsafe

Source: https://docs.dask.org/en/stable/deploying-python-advanced.html

Start nanny, start local process, start watching.

```APIDOC
## *async* start_unsafe()

### Description
Start nanny, start local process, start watching.
```

--------------------------------

### Install Dask from Source

Source: https://docs.dask.org/en/stable/install.html

Installs Dask after cloning the repository. This command installs Dask into your current Python environment.

```bash
cd dask
python -m pip install .
```

--------------------------------

### Compute first non-null entry with Resampler.first()

Source: https://docs.dask.org/en/stable/generated/dask.dataframe.tseries.resample.Resampler.first.html

Use Resampler.first() to get the first non-null value for each resampled period. This example demonstrates resampling a Series by month start frequency.

```python
import pandas as pd
s = pd.Series(
    [1, 2, 3, 4],
    index=pd.DatetimeIndex(
        ["2023-01-01", "2023-01-15", "2023-02-01", "2023-02-15"]
    ),
)
s.resample("MS").first()
```

--------------------------------

### Install Single Dask Cluster with Helm

Source: https://docs.dask.org/en/stable/_sources/deploying-kubernetes.rst

Deploy a single Dask cluster, optionally with Jupyter, on Kubernetes using the Dask Helm chart. This is a simple way to get started with Dask on Kubernetes.

```bash
helm install --repo https://helm.dask.org my-dask dask
```

--------------------------------

### Create DataFrame for Examples

Source: https://docs.dask.org/en/stable/generated/dask.dataframe.api.GroupBy.mean.html

Initializes a pandas DataFrame with sample data for demonstrating groupby operations.

```python
import pandas as pd
import numpy as np

df = pd.DataFrame(
    {"A": [1, 1, 2, 1, 2], "B": [np.nan, 2, 3, 4, 5], "C": [1, 2, 1, 1, 2]},
    columns=["A", "B", "C"],
)
```

--------------------------------

### Securely Install Package with PipInstall and Environment Variables

Source: https://docs.dask.org/en/stable/_sources/changelog.rst

Demonstrates how to use the distributed.PipInstall plugin with environment variables for secure installation from private repositories. Ensure the PipInstall plugin is configured and the necessary environment variables are set.

```python
from distributed.deploy import PipInstall

plugin = PipInstall(target="/path/to/install/packages")

# Install a package from a private repository using a TOKEN environment variable
plugin.install("my-private-package==1.0.0", envvars=["TOKEN"])
```

--------------------------------

### Client.start()

Source: https://docs.dask.org/en/stable/_modules/distributed/client.html

Starts the client and its associated scheduler and workers.

```APIDOC
## Client.start(**kwargs)

### Description
Starts the client and its associated scheduler and workers.

### Parameters
#### Keyword Arguments
- **kwargs**: Additional keyword arguments to pass to the scheduler and worker startup.
```

--------------------------------

### Get the 3 largest elements

Source: https://docs.dask.org/en/stable/generated/dask.dataframe.Index.nlargest.html

This example demonstrates how to specify n=3 to get the three largest elements. The default 'keep' value is 'first', so Malta is kept.

```python
s.nlargest(3)
```

--------------------------------

### Describe a DataFrame

Source: https://docs.dask.org/en/stable/generated/dask.dataframe.Index.describe.html

This example shows how to get descriptive statistics for all columns in a DataFrame.

```APIDOC
## describe()

### Description
Compute descriptive statistics for a DataFrame.

This includes count, mean, standard deviation, min, max, and quartiles for numeric columns, and count, unique, top, and frequency for object or categorical columns.

### Parameters
#### Query Parameters
- **include** (list, optional) - A list of data types to include in the description. Defaults to None.
- **exclude** (list, optional) - A list of data types to exclude from the description. Defaults to None.

### Request Example
```python
df.describe(include=[np.number])
df.describe(include=[object])
df.describe(include=["category"])
df.describe(exclude=[np.number])
df.describe(exclude=[object])
```

### Response
#### Success Response (200)
Returns a DataFrame containing descriptive statistics.
```

--------------------------------

### Configure setup.cfg for dask.sizeof entrypoint

Source: https://docs.dask.org/en/stable/_sources/how-to/extend-sizeof.rst

Configure your project's `setup.cfg` to expose a `dask.sizeof` entrypoint. This allows third-party libraries to provide sizeof implementations without direct code integration.

```ini
[options.entry_points]
dask.sizeof =
   numpy = numpy_sizeof_dask:sizeof_plugin
```

--------------------------------

### Create a Dask DataFrame for examples

Source: https://docs.dask.org/en/stable/generated/dask.dataframe.DataFrame.fillna.html

This code sets up a sample DataFrame using pandas and numpy, which is then used in subsequent examples to demonstrate the `fillna` functionality.

```python
import pandas as pd
import numpy as np

df = pd.DataFrame(
    [
        [np.nan, 2, np.nan, 0],
        [3, 4, np.nan, 1],
        [np.nan, np.nan, np.nan, np.nan],
        [np.nan, 3, np.nan, 4],
    ],
    columns=list("ABCD"),
)

```

--------------------------------

### SSH Worker Start Method

Source: https://docs.dask.org/en/stable/_modules/distributed/deploy/ssh.html

Starts an SSH worker by establishing an asyncSSH connection and executing the Dask worker command remotely. It handles environment variable setup for configuration inheritance.

```python
    async def start(self):
        try:
            import asyncssh  # import now to avoid adding to module startup time
        except ImportError:
            raise ImportError(
                "Dask's SSHCluster requires the `asyncssh` package to be installed. "
                "Please install it using pip or conda."
            )

        self.connection = await asyncssh.connect(self.address, **self.connect_options)

        result = await self.connection.run("uname")
        if result.exit_status == 0:
            set_env = f'env DASK_INTERNAL_INHERIT_CONFIG="{dask.config.serialize(dask.config.global_config)}"'
        else:
            result = await self.connection.run("cmd /c ver")
            if result.exit_status == 0:
                set_env = f"set DASK_INTERNAL_INHERIT_CONFIG={dask.config.serialize(dask.config.global_config)} &&"
            else:
                raise Exception(
                    "Worker failed to set DASK_INTERNAL_INHERIT_CONFIG variable "
                )

        if not self.remote_python:
            self.remote_python = sys.executable

        cmd = " ".join(
            [
                set_env,
                self.remote_python,
                "-m",
                "distributed.cli.dask_spec",
                self.scheduler,
                "--spec",
                "'%s'"
                % dumps(
                    {
                        i: {

```

--------------------------------

### Get Diagonals of a 3D Array

Source: https://docs.dask.org/en/stable/generated/dask.array.diagonal.html

Extracts diagonals from a 3D array by specifying axes. This example shows how to get the main diagonals of sub-arrays formed by fixing the last two axes.

```python
a = np.arange(8).reshape(2,2,2); a
a.diagonal(0,  # Main diagonals of two arrays created by skipping
           0,  # across the outer(left)-most axis last and
           1)  # the "middle" (row) axis first.
```

--------------------------------

### Install Dask from Source

Source: https://docs.dask.org/en/stable/_sources/install.rst

Clones the Dask repository and installs it using pip. Use ".[complete]" to include all optional dependencies.

```bash
git clone https://github.com/dask/dask.git
cd dask
python -m pip install .
```

```bash
python -m pip install ".[complete]"
```

--------------------------------

### Convert Configuration to Environment Variables

Source: https://docs.dask.org/en/stable/configuration.html

This example shows how YAML configuration can be represented as environment variables for distribution.

```bash
export DASK_ARRAY__CHUNK_SIZE="128 MiB"
export DASK_DISTRIBUTED__WORKERS__MEMORY__SPILL=0.85
export DASK_DISTRIBUTED__WORKERS__MEMORY__TARGET=0.75
export DASK_DISTRIBUTED__WORKERS__MEMORY__TERMINATE=0.98
```

--------------------------------

### Build Documentation with Make

Source: https://docs.dask.org/en/stable/_sources/develop.rst

Build the HTML documentation files. The output will be located in the 'build/html' directory. Run this command again after making changes to rst files.

```bash
make html
```

--------------------------------

### Start Dask Client

Source: https://docs.dask.org/en/stable/_sources/futures.rst

Initialize a Dask Client to manage tasks on a cluster. Use `processes=False` to start workers as threads instead of processes. A diagnostic dashboard is available at http://localhost:8787 if Bokeh is installed.

```python
from dask.distributed import Client

client = Client()  # start local workers as processes
# or
client = Client(processes=False)  # start local workers as threads
```

--------------------------------

### Add Conda Installation Instructions for Development

Source: https://docs.dask.org/en/stable/_sources/changelog

Includes instructions on how to use conda for installing Dask for development purposes. This simplifies the setup process for contributors.

```python
Add instructions for using conda when installing code for development (:pr:`6399`)
```

--------------------------------

### Get Day of the Week using Series.dt.weekday

Source: https://docs.dask.org/en/stable/generated/dask.dataframe.Series.dt.weekday.html

Use the `.dt.weekday` accessor on a Series of datetime objects to get the day of the week, where Monday is 0 and Sunday is 6. This example demonstrates its usage with a date range.

```python
>>> s = pd.date_range('2016-12-31', '2017-01-08', freq='D').to_series()
>>> s.dt.weekday
2016-12-31    5
2017-01-01    6
2017-01-02    0
2017-01-03    1
2017-01-04    2
2017-01-05    3
2017-01-06    4
2017-01-07    5
2017-01-08    6
Freq: D, dtype: int32
```

--------------------------------

### Dask Distributed: Local Client Setup

Source: https://docs.dask.org/en/stable/_sources/10-minutes-to-dask.rst

Illustrates setting up a Dask distributed client on a local machine. This client manages a cluster using your computer's resources.

```python
from dask.distributed import Client

client = Client()
client
```

--------------------------------

### Example Usage of apply_gufunc

Source: https://docs.dask.org/en/stable/_modules/dask/array/gufunc.html

This example demonstrates how to use `dask.array.apply_gufunc` with a simple Python function. It shows the basic setup for applying a function that operates on core dimensions, including specifying the signature and output core dimensions.

```python
import dask.array as da
import numpy as np

def my_func(x, y):
    return x + y

x = da.from_array(np.arange(12).reshape(3, 4), chunks=(2, 2))
y = da.from_array(np.arange(12).reshape(3, 4), chunks=(2, 2))

# Apply the function with a signature and output core dimensions
z = da.apply_gufunc(my_func, signature='(i),(i)->(i)',
                    x, y, output_core_dims=[['i']],
                    allow_rechunk=True, vectorize=True)
print(z.compute())

# Example with a function that returns multiple outputs
def my_multi_output_func(x, y):
    return x + y, x - y

z = da.apply_gufunc(my_multi_output_func, signature='(i),(i)->(i),(i)',
                    x, y, output_core_dims=[['i'], ['i']],
                    allow_rechunk=True, vectorize=True)
print(z[0].compute())
print(z[1].compute())
```

--------------------------------

### Scheduler Setup Script with Plugin

Source: https://docs.dask.org/en/stable/_sources/customize-initialization.rst

Use this script with `dask-scheduler --preload` to register a custom scheduler plugin. The `dask_setup` function can accept command-line arguments defined using Click.

```python
# scheduler-setup.py
import click

from distributed.diagnostics.plugin import SchedulerPlugin

class MyPlugin(SchedulerPlugin):
    def __init__(self, print_count):
      self.print_count = print_count
      super().__init__()

    def add_worker(self, scheduler=None, worker=None, **kwargs):
        print("Added a new worker at:", worker)
        if self.print_count and scheduler is not None:
            print("Total workers:", len(scheduler.workers))

@click.command()
@click.option("--print-count/--no-print-count", default=False)
def dask_setup(scheduler, print_count):
    plugin = MyPlugin(print_count)
    scheduler.add_plugin(plugin)

```

--------------------------------

### Get Dask Installation Information

Source: https://docs.dask.org/en/stable/cli.html

Use the `dask info versions` command to display versions of Dask-related projects.

```shell
dask info versions
```

--------------------------------

### Get Non-Empty Metadata Example

Source: https://docs.dask.org/en/stable/_sources/dataframe-design.rst

Retrieve a non-empty sample DataFrame from _meta_nonempty, useful for understanding inferred metadata.

```python
>>> ddf._meta_nonempty
   a    b
0  1  foo
1  1  foo
```

--------------------------------

### Install Documentation Dependencies

Source: https://docs.dask.org/en/stable/_sources/develop.rst

Install the required Python packages for documentation building using pip.

```bash
python -m pip install -r requirements-docs.txt
```

--------------------------------

### Dask Documentation: Add docker image customization example

Source: https://docs.dask.org/en/stable/changelog

Includes an example demonstrating how to customize Docker images for use with Dask, facilitating deployment and environment setup.

```markdown
Add docker image customization example
```

--------------------------------

### Install Dask on GPU CI

Source: https://docs.dask.org/en/stable/_sources/changelog

This snippet details the command used to install Dask on GPU CI builds, ensuring proper environment setup for GPU-accelerated computations.

```bash
pip install dask
```

--------------------------------

### Set up Local Dask Client

Source: https://docs.dask.org/en/stable/10-minutes-to-dask.html

Instantiate a Dask client to manage a local cluster using your computer's resources.

```python
>>> from dask.distributed import Client
...
... client = Client()
...
... client
<Client: 'tcp://127.0.0.1:41703' processes=4 threads=12, memory=31.08 GiB>
```

--------------------------------

### Accessing the year from a datetime Series

Source: https://docs.dask.org/en/stable/generated/dask.dataframe.Series.dt.year.html

This example demonstrates how to use the `.dt.year` accessor to get the year from a Dask Series of datetimes.

```APIDOC
## Series.dt.year

### Description

The year of the datetime.

This docstring was copied from pandas.core.indexes.accessors.CombinedDatetimelikeProperties.year.

Some inconsistencies with the Dask version may exist.

### Method

Attribute Access

### Endpoint

`Series.dt.year`

### Parameters

This attribute does not take any parameters.

### Request Example

```python
import dask.dataframe as dd
import pandas as pd

datetime_series = pd.Series(
    pd.date_range("2000-01-01", periods=3, freq="YE")
)
dask_series = dd.from_pandas(datetime_series, npartitions=1)

dask_series.dt.year.compute()
```

### Response

#### Success Response

Returns a Dask Series containing the year for each datetime element.

#### Response Example

```
0    2000
1    2001
2    2002
dtype: int32
```
```

--------------------------------

### Create a sample DataFrame

Source: https://docs.dask.org/en/stable/generated/dask.dataframe.api.GroupBy.idxmin.html

This code sets up a sample pandas DataFrame for demonstration purposes. It includes columns for 'consumption' and 'co2_emissions' with a custom index.

```python
import pandas as pd
df = pd.DataFrame({
    "consumption": [10.51, 103.11, 55.48],
    "co2_emissions": [37.2, 19.66, 1712],
},
index=["Pork", "Wheat Products", "Beef"],
)
```

--------------------------------

### Describe a Series (Column)

Source: https://docs.dask.org/en/stable/generated/dask.dataframe.Index.describe.html

This example demonstrates how to get descriptive statistics for a single column (Series) of a DataFrame by accessing it as an attribute.

```APIDOC
## describe()

### Description
Compute descriptive statistics for a Series.

This includes count, mean, standard deviation, min, max, and quartiles for numeric data.

### Parameters
This method does not accept any parameters when called on a Series.

### Request Example
```python
df.numeric.describe()
```

### Response
#### Success Response (200)
Returns a Series containing descriptive statistics for the column.
```

--------------------------------

### Install All Dependencies from Source

Source: https://docs.dask.org/en/stable/install.html

Installs Dask from source along with all optional dependencies. This is useful for testing the full Dask suite.

```bash
python -m pip install ".[complete]"
```

--------------------------------

### Start Local Dask Scheduler and Client

Source: https://docs.dask.org/en/stable/_sources/dashboard.rst

Instantiate a Client to automatically start a local Dask scheduler and connect to it. This will also launch the dashboard.

```python
from dask.distributed import Client
client = Client()  # start distributed scheduler locally.
```

--------------------------------

### Extracting Time from a Series

Source: https://docs.dask.org/en/stable/generated/dask.dataframe.Series.dt.time.html

This example demonstrates how to use the `.dt.time` accessor on a Dask Series to get the time component of each datetime entry.

```APIDOC
## Accessing Time Component

### Description
Extracts the time part from a Series of datetime objects.

### Method
Attribute Access

### Endpoint
`Series.dt.time`

### Parameters
None

### Request Example
```python
import dask.dataframe as dd
import pandas as pd

s = pd.Series(["1/1/2020 10:00:00+00:00", "2/1/2020 11:00:00+00:00"])
s = pd.to_datetime(s)
dask_series = dd.from_pandas(s, npartitions=1)

time_part = dask_series.dt.time
print(time_part.compute())
```

### Response
#### Success Response
Returns a Dask Series (or a NumPy array if computed) containing `datetime.time` objects representing the time component of the original Series.

#### Response Example
```
0    10:00:00
1    11:00:00
dtype: object
```
```

--------------------------------

### Create a sample DataFrame

Source: https://docs.dask.org/en/stable/generated/dask.dataframe.DataFrame.drop_duplicates.html

This snippet demonstrates how to create a sample pandas DataFrame for use in subsequent examples.

```python
import pandas as pd
df = pd.DataFrame(
    {
        "brand": ["Yum Yum", "Yum Yum", "Indomie", "Indomie", "Indomie"],
        "style": ["cup", "cup", "cup", "pack", "pack"],
        "rating": [4, 4, 3.5, 15, 5],
    }
)
df
```

--------------------------------

### Get Dashboard URL

Source: https://docs.dask.org/en/stable/_sources/deploying-python.rst

Retrieve the URL for the Dask diagnostic dashboard from a cluster object. Ensure Bokeh is installed to view the dashboard.

```python
>>> cluster.dashboard_link
'http://127.0.0.1:8787/status'
```

--------------------------------

### Create a table from scratch with 4 rows

Source: https://docs.dask.org/en/stable/generated/dask.dataframe.to_sql.html

This example demonstrates creating a Pandas DataFrame, converting it to a Dask DataFrame, and then storing it into a new SQL table named 'test' using a temporary SQLite database. It verifies the data by reading it back from the table.

```python
import pandas as pd
import dask.dataframe as dd
df = pd.DataFrame([ {'i':i, 's':str(i)*2 } for i in range(4) ])
ddf = dd.from_pandas(df, npartitions=2)
ddf
```

```python
from dask.utils import tmpfile
from sqlalchemy import create_engine, text
with tmpfile() as f:
    db = 'sqlite:///%s' %f
    ddf.to_sql('test', db)
    engine = create_engine(db, echo=False)
    with engine.connect() as conn:
        result = conn.execute(text("SELECT * FROM test")).fetchall()
result
```

--------------------------------

### Dask Documentation: Use pip install . instead of setup.py

Source: https://docs.dask.org/en/stable/changelog

Recommends using `pip install .` for installing Dask locally instead of directly calling `setup.py`, aligning with modern Python packaging practices.

```bash
Use pip install . instead of calling setup.py
```

--------------------------------

### Resample and get the last entry

Source: https://docs.dask.org/en/stable/generated/dask.dataframe.tseries.resample.Resampler.last.html

Demonstrates how to resample a pandas Series with a DatetimeIndex to monthly start frequency and then compute the last non-null value for each period.

```python
s = pd.Series(
    [1, 2, 3, 4],
    index=pd.DatetimeIndex(
        ["2023-01-01", "2023-01-15", "2023-02-01", "2023-02-15"]
    ),
)
print(s.resample("MS").last())
```

--------------------------------

### Create sample DataFrames

Source: https://docs.dask.org/en/stable/generated/dask.dataframe.DataFrame.join.html

Initializes two pandas DataFrames for demonstration purposes.

```python
import pandas as pd
df = pd.DataFrame({
    "key": ["K0", "K1", "K2", "K3", "K4", "K5"],
    "A": ["A0", "A1", "A2", "A3", "A4", "A5"],
})

other = pd.DataFrame({"key": ["K0", "K1", "K2"], "B": ["B0", "B1", "B2"]})
```

--------------------------------

### Execute Function on Dask Scheduler

Source: https://docs.dask.org/en/stable/futures.html

Executes a defined function on the Dask scheduler and returns the result. This example shows how to get the total number of tasks.

```python
>>> client.run_on_scheduler(get_number_of_tasks)
100
```

--------------------------------

### Get Worker Monitor Info

Source: https://docs.dask.org/en/stable/_modules/distributed/scheduler.html

Asynchronously gather monitor information from all workers. Can fetch recent data or data since a specific start time.

```python
async def get_worker_monitor_info(self, recent: bool = False, starts: dict | None = None) -> dict:
        if starts is None:
            starts = {}
        results = await asyncio.gather(
            *(self.rpc(w).get_monitor_info(recent=recent, start=starts.get(w, 0)) for w in self.workers)
        )
        return dict(zip(self.workers, results))
```

--------------------------------

### Partial by Order Example

Source: https://docs.dask.org/en/stable/_modules/dask/utils.html

Demonstrates how to use `partial_by_order` to create a function with pre-filled arguments, useful for partial application.

```python
from operator import add
partial_by_order(5, function=add, other=[(1, 10)])
```

--------------------------------

### Nanny Initialization and Configuration

Source: https://docs.dask.org/en/stable/_modules/distributed/nanny.html

This code snippet illustrates the initialization of a Nanny instance, including handling various configuration parameters like security, preloaded modules, scheduler address, and worker-specific settings. It sets up logging, establishes connection arguments, and prepares environment variables for subprocesses.

```python
if loop is not None:
            warnings.warn(
                "the `loop` kwarg to `Nanny` is ignored, and will be removed in a future release. "
                "The Nanny always binds to the current loop.",
                DeprecationWarning,
                stacklevel=2,
            )

        self.__exit_stack = stack = contextlib.ExitStack()
        self.process = None
        self._setup_logging(logger)
        self.loop = self.io_loop = IOLoop.current()

        if isinstance(security, dict):
            security = Security(**security)
        self.security = security or Security()
        assert isinstance(self.security, Security)
        self.connection_args = self.security.get_connection_args("worker")

        self.preload = preload
        if self.preload is None:
            self.preload = dask.config.get("distributed.worker.preload")
        self.preload_argv = preload_argv
        if self.preload_argv is None:
            self.preload_argv = dask.config.get("distributed.worker.preload-argv")

        if preload_nanny is None:
            preload_nanny = dask.config.get("distributed.nanny.preload")
        if preload_nanny_argv is None:
            preload_nanny_argv = dask.config.get("distributed.nanny.preload-argv")

        handlers = {
            "instantiate": self.instantiate,
            "kill": self.kill,
            "restart": self.restart,
            "get_logs": self.get_logs,
            # cannot call it 'close' on the rpc side for naming conflict
            "terminate": self.close,
            "close_gracefully": self.close_gracefully,
            "run": self.run,
            "plugin_add": self.plugin_add,
            "plugin_remove": self.plugin_remove,
        }
        super().__init__(
            handlers=handlers,
            connection_args=self.connection_args,
            local_directory=local_directory,
            needs_workdir=False,
        )

        self.preloads = preloading.process_preloads(
            self, preload_nanny, preload_nanny_argv, file_dir=self.local_directory
        )

        self.death_timeout = parse_timedelta(death_timeout)
        if scheduler_file:
            cfg = json_load_robust(scheduler_file, timeout=self.death_timeout)
            self.scheduler_addr = cfg["address"]
        elif scheduler_ip is None and dask.config.get("scheduler-address"):
            self.scheduler_addr = dask.config.get("scheduler-address")
        elif scheduler_port is None:
            self.scheduler_addr = coerce_to_address(scheduler_ip)
        else:
            self.scheduler_addr = coerce_to_address((scheduler_ip, scheduler_port))

        if protocol is None:
            protocol_address = self.scheduler_addr.split("://")
            if len(protocol_address) == 2:
                protocol = protocol_address[0]

        self._given_worker_port = worker_port
        self.nthreads: int = nthreads or CPU_COUNT
        self.reconnect = reconnect
        self.validate = validate
        self.resources = resources

        self.Worker = Worker if worker_class is None else worker_class

        self.pre_spawn_env = _get_env_variables("distributed.nanny.pre-spawn-environ")
        # To get consistent hashing on subprocesses, we need to set a consistent seed for
        # the Python hash algorithm; xref https://github.com/dask/distributed/pull/8400
        if self.pre_spawn_env.get("PYTHONHASHSEED") in (None, "0"):
            # This number is arbitrary; it was chosen to commemorate
            # https://github.com/dask/dask/issues/6640.
            self.pre_spawn_env.update({"PYTHONHASHSEED": "6640"})

        self.env = merge(
            self.pre_spawn_env,
            _get_env_variables("distributed.nanny.environ"),
            {k: str(v) for k, v in env.items()} if env else {},
        )

        self.config = merge(dask.config.config, config or {})
        worker_kwargs.update(
            {
                "port": worker_port,
                "interface": interface,
                "protocol": protocol,
                "host": host,
            }
        )
        self.worker_kwargs = worker_kwargs

        self.contact_address = contact_address

        self.services = services
        self.name = name
        self.quiet = quiet

        if silence_logs:
            stack.enter_context(silence_logging_cmgr(level=silence_logs))
        self.silence_logs = silence_logs

        self.plugins: dict[str, NannyPlugin] = {}
        self.scheduler = self.rpc(self.scheduler_addr)
        self.memory_manager = NannyMemoryManager(self, memory_limit=memory_limit)

        if (
            not host
            and not interface
            and not self.scheduler_addr.startswith("inproc://")
        ):
            host = unparse_host_port(get_ip(get_address_host(self.scheduler.address)))

        self._start_port = port
        self._start_host = host
        self._interface = interface
        self._protocol = protocol
```

--------------------------------

### Perform normaltest on Dask array

Source: https://docs.dask.org/en/stable/generated/dask.array.stats.normaltest.html

This example demonstrates how to use `stats.normaltest` with Dask arrays. Ensure you have Dask and NumPy installed. The function returns a statistic and a p-value.

```python
import dask.array as da
import numpy as np
from scipy import stats

# Create a Dask array (e.g., from a NumPy array)
size = 10000
data = np.random.normal(0, 1, size=size)
dask_array = da.from_array(data, chunks=(size//4))

# Perform the normaltest
statistic, pvalue = stats.normaltest(dask_array)

print(f"Statistic: {statistic}")
print(f"P-value: {pvalue}")
```

--------------------------------

### Starting Adaptive Scaling

Source: https://docs.dask.org/en/stable/_modules/distributed/deploy/adaptive.html

Initiates the adaptive scaling process by starting the periodic callback and setting the state to 'running'.

```python
assert self.periodic_callback is not None
self.periodic_callback.start()
self.state = "running"
logger.info(
    "Adaptive scaling started: minimum=%s maximum=%s",
    self.minimum,
    self.maximum,
)
```

--------------------------------

### Create a Dask Series with MultiIndex

Source: https://docs.dask.org/en/stable/generated/dask.dataframe.Series.sum.html

This example demonstrates how to create a Dask Series with a pandas MultiIndex. This setup is often used for more complex data indexing and manipulation.

```python
import pandas as pd
import numpy as np

idx = pd.MultiIndex.from_arrays(
    [["warm", "warm", "cold", "cold"], ["dog", "falcon", "fish", "spider"]],
    names=["blooded", "animal"],
)
s = pd.Series([4, 2, 0, 8], name="legs", index=idx)
s
```

--------------------------------

### Basic String Matching with dask.dataframe.Series.str.match

Source: https://docs.dask.org/en/stable/generated/dask.dataframe.Series.str.match.html

Use `str.match()` to check if each string in a Dask Series starts with a given character or pattern. This example demonstrates a simple prefix check.

```python
import pandas as pd

ser = pd.Series(["horse", "eagle", "donkey"])
ser.str.match("e")
```

--------------------------------

### Register Dask CLI entry point in setup.cfg

Source: https://docs.dask.org/en/stable/_sources/cli.rst

Register a custom command as a Dask CLI entry point using the [options.entry_points] section in setup.cfg. This method is suitable for existing projects using setuptools with setup.cfg.

```ini
[options.entry_points]
dask_cli =
    mycommand = mypackage.cli:main

```

--------------------------------

### Initialize Xarray DataArray with Dask array

Source: https://docs.dask.org/en/stable/changelog.html

Example of initializing an Xarray DataArray with a Dask array, specifying dimensions and chunking. This is a common setup for using Dask with Xarray.

```python
arr = xr.DataArray(
    da.random.random((100, 100, 100), chunks=(5, 5, 50)),
    dims=['a', "b", "c"],
)
```

--------------------------------

### Install Dask with all dependencies

Source: https://docs.dask.org/en/stable/_sources/index.rst

Installs Dask and all optional dependencies using pip.

```bash
pip install "dask[complete]"
```

--------------------------------

### client.get_task_stream

Source: https://docs.dask.org/en/stable/_modules/distributed/client.html

Get task stream data from the scheduler. This collects data from the diagnostic 'Task Stream' plot, including start, stop, transfer, and deserialization times for tasks.

```APIDOC
## client.get_task_stream

### Description
Get task stream data from the scheduler. This collects the data present in the diagnostic "Task Stream" plot on the dashboard. It includes the start, stop, transfer, and deserialization time of every task for a particular duration.

Note that the task stream diagnostic does not run by default. You may wish to call this function once before you start work to ensure that things start recording, and then again after you have completed.

### Parameters
- start (Number or string, optional) - When you want to start recording. If a number it should be the result of calling time(). If a string then it should be a time difference before now, like '60s' or '500 ms'.
- stop (Number or string, optional) - When you want to stop recording.
- count (int, optional) - The number of desired records, ignored if both start and stop are specified.
- plot (boolean or str, optional) - If true then also return a Bokeh figure. If plot == 'save' then save the figure to a file.
- filename (str, optional) - The filename to save to if you set ``plot='save'``.
- bokeh_resources (bokeh.resources.Resources, optional) - Specifies if the resource component is INLINE or CDN.

### Request Example
```python
client.get_task_stream()  # prime plugin if not already connected
x.compute()  # do some work
client.get_task_stream()
```

Pass the ``plot=True`` or ``plot='save'`` keywords to get back a Bokeh figure

```python
data, figure = client.get_task_stream(plot='save', filename='myfile.html')
```

Alternatively consider the context manager

```python
from dask.distributed import get_task_stream
with get_task_stream() as ts:
    x.compute()
ts.data
```

### Returns
- L: List[Dict] - A list of dictionaries, where each dictionary represents a task's timing information.

### See Also
- get_task_stream : a context manager version of this method.
```

--------------------------------

### Create Dask Array from HDF5 Dataset

Source: https://docs.dask.org/en/stable/generated/dask.array.from_array.html

This example shows how to create a Dask array from an HDF5 dataset, specifying custom chunk sizes. Ensure you have the h5py library installed.

```python
>>> x = h5py.File("data.h5")["/x"]
>>> a = da.from_array(x, chunks=500)
```

--------------------------------

### Cluster Widget Setup

Source: https://docs.dask.org/en/stable/_modules/distributed/deploy/cluster.html

Sets up an interactive widget for monitoring and controlling cluster scaling. It includes buttons for adaptive scaling and manual scaling, along with status updates.

```python
            adapt.on_click(adapt_cb)

            @log_errors
            def scale_cb(b):
                n = request.value
                with suppress(AttributeError):
                    self._adaptive.stop()
                self.scale(n)
                update()

            scale.on_click(scale_cb)
        else:  # pragma: no cover
            accordion = HTML("")

        scale_status = HTML(self._scaling_status())

        tab = Tab()
        tab.children = [status, VBox([scale_status, accordion])]
        tab.set_title(0, "Status")
        tab.set_title(1, "Scaling")

        self._cached_widget = tab

        def update():
            status.value = self._repr_html_()
            scale_status.value = self._scaling_status()

        cluster_repr_interval = parse_timedelta(
            dask.config.get("distributed.deploy.cluster-repr-interval", default="ms")
        )

        def install():
            pc = PeriodicCallback(update, cluster_repr_interval * 1000)
            self.periodic_callbacks["cluster-repr"] = pc
            pc.start()

        self.loop.add_callback(install)
        return tab
```

--------------------------------

### Create and plot band-limited image

Source: https://docs.dask.org/en/stable/generated/dask.array.fft.ifftn.html

This example demonstrates creating an image with band-limited frequency content by applying an inverse FFT to a Dask array. Ensure matplotlib is installed for plotting.

```python
import matplotlib.pyplot as plt
n = np.zeros((200,200), dtype=complex)
n[60:80, 20:40] = np.exp(1j*np.random.uniform(0, 2*np.pi, (20, 20)))
im = np.fft.ifftn(n).real
plt.imshow(im)
plt.show()
```

--------------------------------

### Example: Restarting Workers by Address

Source: https://docs.dask.org/en/stable/_modules/distributed/client.html

An example demonstrating how to restart specific workers using their network addresses. This requires obtaining worker information from the scheduler first.

```python
client.restart_workers(workers=['tcp://address:port', ...])
```