<jupyter_start><jupyter_text>Exploratory Data Science (EDS) at scale with Dask<jupyter_code>!pip install --upgrade "dask-cloudprovider[azure]" dask-lightgbm lightgbm
from azureml.core import Workspace

ws = Workspace.from_config()
ws
import git
from pathlib import Path

# get root of git repo
prefix = Path(git.Repo(".", search_parent_directories=True).working_tree_dir)
prefix
from azureml.core import Environment
from dask.distributed import Client
from dask_cloudprovider import AzureMLCluster

env = Environment.from_conda_specification(
    "dask-tutorial", prefix.joinpath("environments", "dask.yml")
)
cluster = AzureMLCluster(
    ws,
    vm_size="STANDARD_DS5_V2",
    environment_definition=env,
    initial_node_count=60,
    scheduler_idle_timeout=1200,
)

c = Client(cluster)
c
container_name = "isdweatherdatacontainer"

storage_options = {"account_name": "azureopendatastorage"}
from adlfs import AzureBlobFileSystem

fs = AzureBlobFileSystem(**storage_options)
fs
%%time
files = fs.glob(f"{container_name}/ISDWeather/year=*/month=*/*.parquet")
files = [f"az://{file}" for file in files]
len(files)
files[-5:]
%%time
import dask.dataframe as dd

npartitions = 256
engine = "pyarrow"
blocksize = "1GB"

ddf = dd.read_parquet(
    files, storage_options=storage_options, engine=engine, blocksize=blocksize
).repartition(npartitions=npartitions)
ddf = ddf.set_index(
    dd.to_datetime(ddf.datetime).dt.floor("d"), sorted=False
).persist()
ddf
%%time
len(ddf)
%%time
len(ddf)
%%time
gbs = round(ddf.memory_usage(index=True, deep=True).sum().compute() / 1e9, 2)
print(f"ddf is {gbs} GBs")<jupyter_output><empty_output><jupyter_text>EDA<jupyter_code>%%time
ddf.describe().compute()
%%time
places = (
    ddf.groupby(ddf.index)[["longitude", "latitude", "year"]].mean().compute()
)
import matplotlib.pyplot as plt

plt.figure(figsize=(16, 16))
plt.scatter(places.longitude, places.latitude, c=places.year)
plt.title("Lat/long")
plt.xlabel("Longitude")
plt.ylabel("Latitude")
plt.grid()
plt.colorbar()
plt.figure(figsize=(16, 16))
plt.scatter(places.longitude, places.latitude, c=places.year)
plt.title("Lat/long")
plt.xlabel("Longitude")
plt.ylabel("Latitude")
plt.xlim([-50, -30])  # zoom in
plt.ylim([35, 40])  # zoom in
plt.grid()
plt.colorbar()
%%time
means = ddf.groupby(ddf.index).mean().compute()
means.head()
from datetime import datetime

for col in list(means.columns):
    fig = plt.figure(figsize=(16, 8))
    # plt.style.use('dark_background')
    means[col].plot(color="b")
    plt.title("Average of {}".format(col))
    plt.xlim([datetime(2008, 1, 1), datetime(2021, 1, 1)])
    plt.grid()<jupyter_output><empty_output><jupyter_text>Process and persist<jupyter_code>ddf["temperature"] = ddf["temperature"] * (9 / 5) + 32
ds = ws.get_default_datastore()
container_name = ds.container_name

storage_options = {
    "account_name": ds.account_name,
    "account_key": ds.account_key,
}
%%time
#  ddf.to_csv("az://{container_name}/data/dask/isd", storage_options=storage_options)<jupyter_output><empty_output><jupyter_text>Prepare data and train LightGBM model<jupyter_code>ddf = ddf.fillna(0)
cols = list(ddf.columns)
cols = [
    col
    for col in cols
    if ddf.dtypes[col] != "object"
    and col not in ["version", "datetime", "temperature"]
]
cols
X = ddf[cols].persist()
y = ddf.temperature.persist()
%%time
from dask_lightgbm import LGBMRegressor

params = {
    "n_estimators": 64,
    "num_iterations": 100,
    "boosting_type": "gbdt",
    "num_leaves": 31,
    "max_depth": -1,
    "learning_rate": 0.1,
    "tree_learner": "data",
}

lgbm = LGBMRegressor(**params)
lgbm.fit(X, y)
%%time
y_pred = lgbm.predict(X)
%%time
rmse = ((((y.to_dask_array() - y_pred) ** 2).mean()) ** 0.5).compute()
print(f"Training RMSE: {round(rmse, 3)}")<jupyter_output><empty_output><jupyter_text>Close Cluster and Client<jupyter_code>cluster.close()
c.close()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.15855532786885246, 'fraction_numerical': 0.016905737704918034, 'mean_word_length': 5.838879159369527, 'pattern_counts': {'":': 10, '<': 21, '<?xml version=': 0, '>': 21, 'https://': 0, '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': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '37856', 'license': 'MIT', 'n_tokens_mistral': 1564, 'n_tokens_neox': 1470, 'n_words': 276, 'path': 'tutorials/using-dask/2.eds-at-scale.ipynb', 'repo_name': 'datashinobi/azureml-examples', 'size': '3904'}