<jupyter_start><jupyter_text>Introduction to DaskIn this notebook, we'll learn how to use [Dask](https://dask.org) for reading and processing data from Azure.  Install required packages<jupyter_code>!pip install --upgrade dask==2.30.0 distributed==2.30.0 dask-sql bokeh adlfs fsspec fastparquet pyarrow python-snappy lz4<jupyter_output><empty_output><jupyter_text>Get AML WorkspaceYou can use the AML workspace to retrieve datastores and keyvaults for accessing data credentials securely.<jupyter_code>from azureml.core import Workspace

ws = Workspace.from_config()
ws<jupyter_output><empty_output><jupyter_text>Create a distributed clientThe [client](https://distributed.dask.org/en/latest/client.html) is the primary entrypoint for parallel processing with Dask. Calling it without inputs will create a local distributed scheduler, utilizing all the CPUs and cores on your machine. This can be useful for faster processing of larger in memory dataframes, or even computations on out of memory (OOM) data. When your local machine isn't powerful enough, you can provision a larger VM in Azure - the M series has 100+ CPUs and TBs of RAM. If this still isn't powerful enough, you can create a distributed Dask cluster on most hardware - see [the Dask setup guide](https://docs.dask.org/en/latest/setup.html) for details.If you still need acceleration, [RAPIDSAI](https://github.com/rapidsai) further extends the PyData APIs on GPUs.**Make sure you check out the dashboard!**<jupyter_code>from distributed import Client

# initialize local client
c = Client()

# print Python objects
print(c)
print(c.dashboard_link)

# print notebook widget widget
c<jupyter_output><empty_output><jupyter_text>Reading cloud dataReading data from the cloud is as easy! Python implements various cloud protocols, including ``az`` for Blob and ADLSv2 and ``adl`` for ADLSv1. Public DataPublic data can simply be read via ``https``.<jupyter_code>account_name = "azuremlexamples"
container_name = "datasets"

storage_options = {"account_name": account_name}
data_uri = f"https://{account_name}.blob.core.windows.net/{container_name}/iris.csv"
data_uri
import pandas as pd

df = pd.read_csv(data_uri)
df<jupyter_output><empty_output><jupyter_text>alternatively, we can use the ``az`` protocol and pass in ``storage_options``:<jupyter_code>data_uri = f"az://{container_name}/iris.csv"
data_uri
import pandas as pd

df = pd.read_csv(data_uri, storage_options=storage_options)
df<jupyter_output><empty_output><jupyter_text>Private Data Passing in storage options allows for reading private data. For instance, you can easily retrieve the information from an Azure ML Datastore:```pythonfrom azureml.core import Workspacews = Workspace.from_config()ds = ws.get_default_datastore()  ws.datastores["my-datastore-name"]container_name = ds.container_namestorage_options = {    "account_name": ds.account_name,    "account_key": ds.account_key,}```<jupyter_code>from adlfs import AzureBlobFileSystem

ds = ws.get_default_datastore()

container_name = ds.container_name
storage_options = {"account_name": ds.account_name, "account_key": ds.account_key}

fs = AzureBlobFileSystem(**storage_options)
fs
fs.ls(f"{container_name}")<jupyter_output><empty_output><jupyter_text>Pythonic FilesystemIn the previous section, we used [ADLFS](https://github.com/dask/adlfs) to initialize a Pythonic filesystem and perform operations.The below cell demonstrate some basic operations to raed and manipulate data in Python.<jupyter_code>color = "green"
container_name = "nyctlc"
storage_options = {"account_name": "azureopendatastorage"}
fs = AzureBlobFileSystem(**storage_options)
fs
fs.ls(f"{container_name}")
fs.ls(f"{container_name}/{color}")
fs.ls(f"{container_name}/{color}/puYear=2016/")
files = fs.glob(f"{container_name}/{color}/puYear=2016/puMonth=12/*.parquet")
files = [f"az://{file}" for file in files]
files[-5:]
import dask.dataframe as dd

ddf = (
    dd.read_parquet(files, storage_options=storage_options)
    .repartition(npartitions=8)
    .persist()
)
ddf
%%time
len(ddf)
%%time
len(ddf)
ddf.info()
import matplotlib.pyplot as plt

plt.style.use("dark_background")

ddf["tipAmount"].compute().hist(
    figsize=(16, 8),
    bins=256,
    range=(0.1, 20),
)
df = ddf.compute()
df.info()
%%time
df.describe()
%%time
gbs = round(df.memory_usage(index=True, deep=True).sum() / 1e9, 2)
print(f"df is {gbs} GBs")
%%time
gbs = round(ddf.memory_usage(index=True, deep=True).sum().compute() / 1e9, 2)
print(f"ddf is {gbs} GBs")<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11324928113249282, 'fraction_numerical': 0.0092899800928998, 'mean_word_length': 6.352845528455284, 'pattern_counts': {'":': 6, '<': 29, '<?xml version=': 0, '>': 29, 'https://': 6, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '119636', 'license': 'MIT', 'n_tokens_mistral': 1520, 'n_tokens_neox': 1466, 'n_words': 412, 'path': 'tutorials/using-dask/1.intro-to-dask.ipynb', 'repo_name': 'Nanthini10/azureml-examples', 'size': '4521'}