Spark AI Summit 2020

kdf = ks.from_pandas(pdf)

pdf - kdf.to_pandas()

sdf = kdf.to_spark(index_col='index').show() # if you want to add an 1:N index column

pdf = sdf.to_koalas(index_col=['x', 'y'])

kdf.mean()

kdf.describe()

kdf.groupby('A').sum

kdf['col'].cumsum().plot()

kdf = ks.DataFrame({'a': [1, 2,3 ], 'b':[4, 5, 6]})

def pandas_plus(pser):

    #return pser + 1

    #return pser[pser % 2 == 1]

#kdf.transform(pandas_plus)

#kdf.apply(pandas_plus)

kdf.spark.schema().simpleString()

kdf.spark.print_schema()

kdf.spark.apply(lambda sdf: sdf.select(sdf['A'] * sdf['B'])).head()

kdf.spark.apply(lambda sdf: sdf.select(sdf['index'], sdf['A'] * sdf['B']), index_col='index').head()

kdf.A.spark.transform(lambda scol: scol.cast('int')).head()

(kdf + 1).spark.explain()

naive_solution = featurestore_sample_100k.crossJoin(functions.broadcast(model_data))

# Have at least 1 inclusion and no exclusion hashes

def arrays_intersect(co1, co2):

    return functions.size(functions.array_intersect(col1, col2)) != 0

native_solution = native_solution.filter(arrays_intersect('featureIds', 'geoInclusionHashes') & (~arrays_intersect('featureIds', 'geoEnclusionHashes'))

native_solution  = native_solution .select('modelId', 'id').distinct()

native_solution  = native_solution .groupby('modelId').count()

def simple_group_by(feature_store_id_data):

    id_set = []

    for model_id, inclusion_ids, exclusion_ids in model_data_df[['modelId', 'geoInclusionSet', 'geoExclusionSet']].values:

        for id_row in feature_store_id_data['featureIds'].values:

            if (not inclusion_ids.isdisjoint(id_row) and exclusion_ids.isdisjoin(id_row)):

                id_set.append(model_id)

               break

    return pd.DataFrame([[cur_id] for cur_id in id_set], columns=['modelId'])

simple_group_by_udf = functions.pandas_udf(

    simple_group_by,

    StructType([StructField('modelId', IntergetType())]),

    PandasUDFType.GROUPED_MAP)

key_aggregated_solution = (featurestore_sample_100k

  .groupby('id')

  .apply(key_aggregated_group_by_udf)

  .groupby('filterId')

  .count()

  .join(functions.broadcast(filter_model_id_pairs), on='filterId'))

batch_solution = (featurestore_sample_1M

  .withColumn('batch', functions.abs(functions.hash('id') % 100))

  .groupby('batch').apply(batch_group_by_udf)

  .groupby('filterId')

  .agg(functions

  .sum('batchCount')

  .alias('count'))

  .join(functions.broadcast(filter_model_id_pairs), on='filterId'))

df = spark.read.parquet(path)

featureNames = Sep("f1", "f2", "f3")

vectorAssemble = VectorAssembler().setInputCols(featureNames.toArray).setOutputCol("features")

xgbInput = vectorAssemble .transform(df).select("features", labelColName)

xgbClassifier = XGBoostClassifier(params).setLabelCol(labelColName).setTreeMethod("hist").setFeaturesCol("features")

model = xgbClassifier.fit(xgbInput)

gpuDF = GpuDataReader(spark).parquet(path)

featureNames = Sep("f1", "f2", "f3")

xgbClassifier = XGBoostClassifier(params).setLabelCol(labelColName).setTreeMethod("gpu_hist").setFeaturesCol("featuresNames")

model = xgbClassifier.fit(gpuDF)

import pandas as pd

df = pd.read_csv('my_data.csv')

df.columns = ['x', 'y', 'z1']

df['x2'] = df.x * df.x

df = spark.read.option("inferSchume", "true").csv("my_data.csv")

df = df.toDF('x', 'y', 'z1')

df = df.withColumn('x2', df.x * df.x)

import databricks.koalas as ks

df = ks.read_csv('my_data.csv')

df.columns = ['x', 'y', 'z1']

df['x2'] = df.x * df.x

kdf = ks.read_delta("dbfs:/home/niall/koalas/sais_2020")

kdf.head()

kdf.shape

genus_test = "Phascolarctos"

ks.sql("""

SELECT COUNT(*) 

FROM {kdf}

WHERE genus = {genus_test}

""")

kdf.columns

kdf.describe()

kdf.isna().sum()

kdf = kdf[kdf["individualCount"].notnull() & kdf["month"].notnull() & kdf["year"].notnull()]

kdf["kingdom"].value_counts(normalize=True)

kdf["kingdom"].value_counts(normalize=True).plot.bar(rot=25, figsize=(12,8), title="Bar plot of kingdom column using Koalas DataFrame")

len(kdf["class"].unique())

# Group by the class column, sum up the individual counts for each group, and get the count of instances in each group.

# Note that using as_index=False resets the index

class_ks_df = kdf.groupby("class", as_index=False).agg({"individualCount": "sum", "gbifID": "count"})

# Renaming our columns to reflect the grouping aggregates

class_ks_df.columns = ["class", "total_count", "number_distinct_recordings"]

# Create a column which calculates the average number of animal/plant counts per each distinct recording for each class

class_ks_df["avg_count_per_recording"] = class_ks_df["total_count"] / class_ks_df["number_distinct_recordings"]

class_ks_df.sort_values(by="number_distinct_recordings", ascending=False).head(10)

class_ks_df["avg_count_per_recording"].hist(bins=100, figsize=(12,8), title="Histogram of average individualCount per observation (Aggregated to class level)")

class_ks_df.sort_values(by="avg_count_per_recording", ascending=False).head().style.background_gradient(cmap="Greens")

koalas_df = kdf[kdf.genus == "Phascolarctos"]

koalas_df.head()

koalas_df.shape[0]

koalas_df["eventDate"].min(), koalas_df["eventDate"].max()

koalas_df["stateProvince"].value_counts()

koalas_df["individualCount"].hist(bins=100, figsize=(12,8), title="Histogram of individualCount per recording for Koalas")

koalas_df = koalas_df[koalas_df["individualCount"] < 100]

koalas_df["individualCount"].hist(bins=100, figsize=(12,8), title="Histogram of individualCount per recording for Koalas")

# Get the sum of all koalas recorded in Australia for each month

koalas_sum_ks_df = koalas_df.groupby(["year", "month"], as_index=False)["individualCount"].sum()

# Get the count of all koalas occurrences in the Australia for each month

koalas_count_ks_df = koalas_df.groupby(["year", "month"], as_index=False)["individualCount"].count()

# Join the above Koalas DataFrames and divide the total_count column by num_recordings column

koalas_avg_cnt_ks_df = koalas_sum_ks_df.merge(koalas_count_ks_df, on=["year", "month"])

koalas_avg_cnt_ks_df.columns = ["year", "month", "total_count", "num_recordings"]

# Get number of koalas per recording nationally

koalas_avg_cnt_ks_df["avg_cnt"] = koalas_avg_cnt_ks_df["total_count"] / koalas_avg_cnt_ks_df["num_recordings"]

koalas_avg_cnt_ks_df.sort_values(by=["year", "month"], ascending=False).head(12)

# Concatenating the year and month columns to create a year_month column

koalas_avg_cnt_ks_df = ks.sql("""

SELECT year, month, num_recordings, total_count, avg_cnt, CONCAT(year, "-", month) AS year_month

FROM {koalas_avg_cnt_ks_df}

""")

koalas_avg_cnt_ks_df.head()

# Convert the year_month column to datetime (returns in format %Y-%m-%d)

koalas_avg_cnt_ks_df["year_month"] = ks.to_datetime(koalas_avg_cnt_ks_df["year_month"])

# Need to sort by year_month before plotting

koalas_avg_cnt_ks_df.sort_values(by=['year_month'], inplace=True)

# Creating numpy arrays from year_month, total_count and num_recordings

x_axis_vals = koalas_avg_cnt_ks_df["year_month"].to_numpy()

y1_vals = koalas_avg_cnt_ks_df["total_count"].to_numpy()

y2_vals = koalas_avg_cnt_ks_df["num_recordings"].to_numpy()

import matplotlib.pyplot as plt

import seaborn as sns; sns.set_style("white")

plt.rcParams.update({'font.size': 9})

fig, ax1 = plt.subplots(figsize=(12,8))

color = 'tab:red'

ax1.set_xlabel('year_month')

ax1.set_ylabel('total_count', color=color)

ax1.plot(x_axis_vals, y1_vals, color=color)

ax1.tick_params(axis='y', labelcolor=color)

ax2 = ax1.twinx() 

color = 'tab:blue'

ax2.set_ylabel('num_recordings', color=color) 

ax2.plot(x_axis_vals, y2_vals, color=color)

ax2.tick_params(axis='y', labelcolor=color)

fig.tight_layout()  

ax1.set_title("Monthly total count and number of recordings for Koalas in Australia from 2011 - 2018", size=9)

plt.show()

sns.set_style("whitegrid")

# Set year_month to be the index and keep only avg_cnt

koalas_avg_count_ks_series = koalas_avg_cnt_ks_df.set_index("year_month")["avg_cnt"]

plt.title("Monthly average koala count in Australia from 2010 - 2020", size=9)

plt.xlabel("date")

plt.ylabel("avg_num_koalas_per_occurence")

koalas_avg_count_ks_series.plot(figsize=(12, 8))

# First filter to the classes we would like to forecast for

filter_1 = (kdf["genus"] == "Phascolarctos") # Koala

filter_2 = (kdf["genus"] == "Macropus") # Kangaroo

filter_3 = (kdf["genus"] == "Cacatua") # Cockatoo

filter_4 = (kdf["genus"] == "Dromaius") # Emu

subset_kdf = kdf[filter_1 | filter_2 | filter_3 | filter_4]

# Get the total number of counts and total number of observations for each class in every month

grouped_kdf = subset_kdf.groupby(["genus", "year", "month"], as_index=False) \

                          .agg({"individualCount": "sum", "occurrenceID": "count"}) \

                          .sort_values(by=["year", "month"])

grouped_kdf.columns = ["genus", "year", "month", "total_count", "num_recordings"]

# Concatenating the year and month columns to create a year_month column

grouped_kdf = ks.sql("""

SELECT genus, year, month, num_recordings, total_count, CONCAT(year, "-", month) AS year_month

FROM {grouped_kdf}

ORDER BY year, month

""")

# Convert the year_month column to datetime (returns in format %Y-%m-%d)

grouped_kdf["year_month"] = ks.to_datetime(grouped_kdf["year_month"], format="%Y-%m")

grouped_kdf.head()

# MLflow set up

import mlflow

from  mlflow.tracking import MlflowClient

# MLflow experiment (specify path for experiment to be set to)

# NOTE: you will need to change the following experiment path 

# dependent on where you would like to save the model to

mlflow_exp = f"/Users/niall.turbitt@databricks.com/Koalas/Koalas SAIS 2020/koalas_sais20_exp"

mlflow.set_experiment(mlflow_exp)

# Get the experiment ID to track to

exp_id = MlflowClient().get_experiment_by_name(mlflow_exp).experiment_id

from fbprophet import Prophet

import numpy as np

import matplotlib.pyplot as plt

def fcst_func(pdf) -> ks.DataFrame[np.datetime64, np.float64, np.float64, np.int64, str]:

  """

  Function to take grouped genus data, and provide 3-year monthly 

  forecast of total count. Log model plot as artifacts in MLflow 

  pdf: grouped genus data

  """ 

  genus_name = pdf["genus"].iloc[0]

  pdf["ds"] = pdf["year_month"]

  pdf["y"] = pdf["total_count"]

  df = pdf[["ds", "y"]].copy()

  # log the run with MLflow

  with mlflow.start_run(experiment_id=exp_id,

                        run_name=f"{genus_name} - Prophet Forecast") as run:

    prophet_daily_seasonality = False

    prophet_weekly_seasonality = False

    prophet_yearly_seasonality = True

    mlflow.log_params({"genus": genus_name,

                       "prophet_param_daily_seasonality": prophet_daily_seasonality,

                       "prophet_param_weekly_seasonality": prophet_weekly_seasonality,

                       "prophet_param_yearly_seasonality": prophet_yearly_seasonality})

    m = Prophet(daily_seasonality=prophet_daily_seasonality, 

                weekly_seasonality=prophet_weekly_seasonality, 

                yearly_seasonality=prophet_yearly_seasonality, 

                seasonality_mode="additive")

    m.fit(df)

    future = m.make_future_dataframe(periods=36, freq="M")

    fcst = m.predict(future)

    fig_1 = m.plot(fcst)

    plt.xlabel(f"Date")

    plt.ylabel(f"{genus_name} Numbers (Monthly)")

    plt.title(f"{genus_name} - Prophet Forecast")

    fcst_path = f"/tmp/{genus_name}_forecast.png"

    fig_1.savefig(fcst_path)

    mlflow.log_artifact(fcst_path)

    fig_2 = m.plot_components(fcst)

    components_path = f"/tmp/{genus_name}_components.png"

    fig_2.savefig(components_path)

    mlflow.log_artifact(components_path)

    plt.clf()

  ret_pdf = fcst[["ds", "yhat", "yhat_lower", "yhat_upper"]]

  ret_pdf["genus"] = [genus_name] * len(ret_pdf)

  return ret_pdf

fcst_df = grouped_kdf.groupby("genus", as_index=False).apply(fcst_func)

len(fcst_df)

test_genus = "Cacatua"

# The following returns a pandas DataFrame of runs

mlflow_res_pdf = mlflow.search_runs(filter_string=f"params.genus='{test_genus}'").sort_values("start_time", ascending=False)

mlflow_res_pdf

artifact_uri = mlflow_res_pdf.iloc[0]["artifact_uri"]

local_img_path = "/" + artifact_uri.replace(":", "") + f"/{test_genus}_forecast.png"

plt.figure(figsize = (12,8))

img = plt.imread(local_img_path)

plt.imshow(img)

plt.show()

import torch

class Net(torch.nn.Module):

    def __initi__(self):

        self.fc1 = torch.nn.Linear(8, 64)

        self.fc2 = torch.nn.Linear(64, 1)

    def forward(self, x):

        x = torch.relu(self.fc1.forward(x))

        x = torch.dropout(x, p=0.5)

        x = torch.sigmoid(self.fc2.forward(x))

        return x

net = Net()

data_loader = torch.utils.data.DataLoader(torchvision.datasets.MNIST('./data'))

optimizer = torch.optim.SGD(net.parameters())

for epoch in range(1, 11):

    for data, target in data_loader:

        optimizer.zero_grad()

        prediction = net.forward(data)

        loss = F.nll_loss(prediction, target) # neg log-loss

        loss.backward()

        optimizer.step()

    if epoch % 2 == 0:

        torch.save(net, "net.pt")

@pandas_udf(df.schema, PandasUDFType.GROUPED_MAP)

def fill_pdf(pdf):

    for col in col_list:

        pdf.loc[:, col] = pdf[[col]].ffill().bfill()

    return pdf

new_df = df.groupBy('FIELD').apply(fill_pdf)