California Housing Dataset

Mix.install([
  {:nx, "~> 0.2"},
  {:exla, "~> 0.5"},
  {:scholar, "~> 0.1"},
  {:req, "~> 0.3"},
  {:explorer, "~> 0.5.0"},
  {:vega_lite, "~> 0.1.6"},
  {:kino_vega_lite, "~> 0.1.7"},
  {:kino_explorer, "~> 0.1.7"}
])

# Use XLA backend for faster computation
Nx.global_default_backend(EXLA.Backend)
Nx.Defn.global_default_options(compiler: EXLA, client: :host)

Introducción

El objetivo de este livebook es presentar algunas de las librerias que provee Elixir para trabajar en ciencia de datos y aprendizaje automático. El ejemplo que veremos está fuertemente basado en el libro Hands on Machine Learning w/Scikit-Learn, Keras, and TensorFlow, específicamente en el Jupyter Notebook del capítulo 2: End-to-end ML Project. A su vez, vamos a demostrar como procesar un conjunto de datos para luego entrenar y evaluar un modelo de ML básico.

Calofornia Housing es un conocido dataset que consiste en información obtenida de un censo en California en el año 1990, y contiene información relacionada a las casas de los diferentes Census Block Groups (CBG) del estado. Cada dato del dataset contiene la siguiente información acerca de un CBG:

• longitude
• latitude
• housing_median_age
• total_rooms
• total_bedrooms
• population
• households
• median_income
• median_house_value
• ocean_proximity

Supongamos que estamos trabajando para una empresa que nos asignó la tarea utilizar este dataset para construir un modelo de predicción del valor medio de una casa en un CBG. Es decir, nuestro modelo deberá ser capaz de aprender de nuestros datos para poder realizar una predicción del median_house_value, dada una nueva instancia con todos los otros datos.

Un problema con este conjunto de datos es que no está limpio: tiene algunas filas con datos sin completar. Por ello vamos a necesitar preprocesar los datos antes de entrenar el modelo.

Obtención de datos

Lo primero que debemos hacer es descargar los datos y convertirlos a un formato tabular. Para eso vamos a utilizar la libreria Explorer, la cual provee funciones convenientes para manipular datos en forma de tabla.

require Explorer.DataFrame, as: DF

url = "https://raw.githubusercontent.com/ageron/data/main/housing.tgz"

[{'housing/housing.csv', csv_data}] = Req.get!(url).body

housing_df = DF.load_csv!(csv_data)

Vistazo a la estructura de datos

Una vez obtenido el dataset podemos comenzar a analizar los datos. Usando algunas funciones de Explorer.DataFrame podemos dar un vistazo inicial a la estructura.

housing_df
|> DF.describe()

count, null_count, min y max son bastante autoexplicatorios. std representa la desviación estándar, que cuantifica la dispersión de los datos. Las filas 25%, 50%, y 75% son los llamados percentiles, e indican el valor del atributo por debajo del cual se encuentra un porcentaje de muestras.

Todos las características del dataset son numéricas, salvo por ocean_proximity que es de tipo string. Al analizar un poco más de que se trata esta columna, nos damos cuenta que es un atributo categórico del dataset.

alias Explorer.Series, as: S

housing_df["ocean_proximity"]
|> S.frequencies()

Otra forma de analizar los datos es creando diferentes tipos de gráficas para nuestros datos. Por ejemplo, podemos graficar un histograma para cada dato numérico.

alias VegaLite, as: VL

VL.new(
  title: [
    text: "Univariate Histograms of all features"
  ],
  width: 500,
  height: 500,
  columns: 3,
  config: [
    axis: [
      grid: true,
      grid_color: "#dedede"
    ]
  ]
)
|> VL.data_from_values(housing_df)
|> VL.concat(
  for name <- List.delete(housing_df.names, "ocean_proximity") do
    VL.new()
    |> VL.mark(:bar)
    |> VL.encode_field(:x, name, bin: [bin: true, maxbins: 50], axis: [ticks: true])
    |> VL.encode_field(:y, "value count", aggregate: :count)
  end
)

Algunas apreciaciones destacables:

• los atributos tienen escalas muy distintas entre si. Más adelante veremos una técnica de preprocesado de datos llamada feature scaling que nos permitirá normalizar la escala de todas las características.
• muchas de las gráficas están sesgadas hacia la derecha (right-skewed). Esto puede ser perjudicial para nuestro modelo para detectar patrones entre las características. Transformaremos estos atributos para que tengan una distribución gaussiana (forma de campana y simétrica).

Explorando los datos en profundidad

Dadas las columnas de latitud y longitud, podemos crear un represantación geográfica de los datos. En la siguiente representación podemos apreciar las zonas de alta densidad de CBG en California.

california =
  VL.new(
    title: [
      text: "Latitude-longitude scatterplot"
    ],
    width: 750,
    height: 750,
    config: [
      axis: [
        grid: true,
        grid_color: "#dedede"
      ]
    ]
  )
  |> VL.data_from_values(housing_df)
  |> VL.mark(:point, opacity: 1)
  |> VL.encode_field(
    :x,
    "longitude",
    type: :quantitative,
    scale: [
      zero: false
    ]
  )
  |> VL.encode_field(
    :y,
    "latitude",
    type: :quantitative,
    scale: [
      zero: false
    ]
  )

Si además codificamos otros datos de la tabla en nuestra gráfica, podemos obtener una visualización más útil de los datos. En la siguiente gráfica codificamos dos atributos más: median_house_value con un esquema de colores predefinido; y population, para el cual aumentaremos el tamaño del punto proporcinalmente.

california
|> VL.encode(:color,
  field: "median_house_value",
  type: :quantitative,
  scale: [
    scheme: "turbo"
  ]
)
|> VL.encode(:size,
  type: :quantitative,
  field: "population"
)

Esta imágen nos dice que median_house_value esta fuertemente relacionada con la locación (cercanía al océano) y con la densidad de población. Un algoritmo de clustering puede ser útil para detectar los clusters geográficos principales y para agregar nuevos features que midan la cercanía con los clusters principales. Más adelante veremos una técnica de clustering denominada KMeans. El atributo ocean_proximity también puede ser útil, aunque en la costa norte de California los precios no parecen tan altos como en la costa sur.

Experimentando con combinaciones

Entre las características del dataset podemos encontrar algunas que no parecen tan útiles por sí solas. Por ejemplo, la cantidad de salas o cantidad de dormitorios por CBG (total_rooms y total_bedrooms) no parecen tener mucha significación. Sin embargo, podemos obtener algo más útil a partir de la combinación de estos datos, por ejemplo las tasas de dormitorio por sala, salas por casa, y personas por casa. Esto se puede lograr simplemente utilizando la macro Explorer.DataFrame.mutate/2.

DF.mutate(
  housing_df,
  rooms_per_house: total_rooms / households,
  bedrooms_ratio: total_bedrooms / total_rooms,
  people_per_house: population / households
)
|> DF.select(["rooms_per_house", "bedrooms_ratio", "people_per_house"])

Preprocesamiento de datos

El preprocesamiento de datos es fundamental antes de entrenar nuestro modelo. Nos va a permitir tanto mejorar la calidad y representatividad de nuestros datos, como descartar y transformar características irrelevantes. Además de las utilidades que nos brinda Explorer, vamos a utilizar Scholar, una librería que provee herramientas de ML tradicionales entre las cuales encontraremos utilidades para preprocesado.

Limpieza

Ya vimos que Explorer nos provee de la función Explorer.Series.fill_missing/2 que nos permite completar las celdas vacías de una columna siguiendo una estrategía preferida. Sin embargo, Scholar también nos provee de un módulo específico para esto. En cualquier caso, no podremos convertir un dataframe a tensor si el dataframe contiene celdas vacías. Por lo tanto, debemos primero sustituírlas por :nan. SimpleImputer.fit/1 por defecto tomará :nan como un valor faltante y lo completará siguiendo la estrategia preferida.

alias Scholar.Impute.SimpleImputer

num_housing_df = DF.discard(housing_df, "ocean_proximity")

clean_cols =
  for name <- DF.names(num_housing_df), into: [] do
    {name, S.fill_missing(housing_df[name], :nan)}
  end

x = DF.new(clean_cols) |> Nx.stack(axis: 1)

imputer = SimpleImputer.fit(x, strategy: :median)

El struct devuelto por la función tiene dos campos: statistics es un vector con los valores imputados para cada característica (en este caso es la mediana), y missing_values es el valor que tiene una celda vacía y que fue imputada. Finalmente podemos aplicar SimpleImputer.transform/2 y obtener un tensor con los datos limpios.

x_imputed = SimpleImputer.transform(imputer, x)

Podemos construir un nuevo dataframe con nuestros datos limpios.

num_attrs = DF.names(DF.discard(housing_df, "ocean_proximity"))

num_housing_df =
  x_imputed
  |> DF.new()
  |> DF.rename(num_attrs)

Escalado (feature scaling)

El escalado es una de las transformaciones más importantes que vamos a realizar. Los algoritmos de aprendizaje automático no dan buenos resultados si los atributos numéricos difieren mucho en cuanto a sus escalas. Scholar nos provee de algunas funciones de escalado en el módulo Scholar.Preprocessing. Particularmente utilizaremos standard_scale/2.

La estandarización consiste en restar la media y dividir por la desviación estándar. Esta transformación es útil en casos donde los datos tienen una distribución gaussiana. Recordemos que en nuestro caso muchas de las columnas tienen distribución sesgada hacia la derecha.

Una forma de obtener una distribución gaussiana es reemplazando los valores por el logarítmo.

heavy_tail_attrs = ["total_bedrooms", "total_rooms", "population", "households", "median_income"]

x_log =
  num_housing_df
  |> DF.select(heavy_tail_attrs)
  |> Nx.stack(axis: 1)
  |> Nx.log()

Veamos como se ve la distribución de estas características luego de aplicar logarítmo.

log_df =
  x_log
  |> DF.new()
  |> DF.rename(heavy_tail_attrs)

VL.new(
  title: [
    text: "Histograms after applying log"
  ],
  width: 500,
  height: 500,
  columns: 3,
  config: [
    axis: [
      grid: true,
      grid_color: "#dedede"
    ]
  ]
)
|> VL.data_from_values(log_df)
|> VL.concat(
  for name <- heavy_tail_attrs do
    VL.new()
    |> VL.mark(:bar)
    |> VL.encode_field(:x, name, bin: [bin: true, maxbins: 50], axis: [ticks: true])
    |> VL.encode_field(:y, "value count", aggregate: :count)
  end
)

Mucho mejor! Ahora sí estamos en condiciones de aplicar la transformación de Standard Scaler.

alias Scholar.Preprocessing

x_scaled = Preprocessing.standard_scale(x_log, axes: [0])

Una vez más podemos integar nuestros datos transformados en un dataframe.

log_attrs = Enum.map(heavy_tail_attrs, &"log_#{&1}")

log_num_df =
  x_scaled
  |> DF.new()
  |> DF.rename(log_attrs)

Creando nuestros propios transformadores

Aunque las librerías estamos utilizando proveen herramientas para transformar datos, en general no van a ser suficientes y tendremos que crear nuestros propios transformadores. Por ejemplo, en la sección anterior vimos que aplicar logarítmo a distribuciones de cola pesada es una buena idea.

En el siguiente ejemplo creamos un transformador que utilizaremos para determinar los clusters geográficos principales y posteriormente medir la similaridad de un punto a su cluster.

defmodule ClusterSimilarity do
  alias Scholar.Cluster.KMeans
  alias Scholar.Metrics.Clustering

  def fit(x, n_clusters \\ 10, random_state \\ 42) do
    KMeans.fit(x, num_clusters: n_clusters, seed: random_state)
  end

  def transform(x, labels, n_clusters \\ 10) do
    Clustering.silhouette_samples(x, labels, num_clusters: n_clusters)
  end

  def fit_transform(x, n_clusters \\ 10, random_state \\ 42) do
    %KMeans{
      clusters: _cluster_centers,
      labels: labels
    } = fit(x, n_clusters, random_state)

    transform(x, labels, n_clusters)
  end
end

Hagamos un desglose del transformador anterior:

• utilizaremos dos módulos de Scholar: Cluster.KMeans y Metrics.Clustering.
• KMeans es un algoritmo de localización de clusters en los datos, y la cantidad de clusters que debe buscar está dado por el hiperparámetro num_clusters. Luego de entrenar el algoritmo de KMeans, retorna un struct con los clusters y las etiquetas, entre otros datos. Este algoritmo lo ejecutamos en la función fit/3 para obtener los clusters.
• El coeficiente de Silueta es una métrica que evalúa la calidad de los agrupamientos obtenidos a través de un algoritmo de clustering. Mide la similaridad de cada muestra respecto a su cluster asignado en comparación con la similaridad a los otros clusters. Utilizamos esta métrica para transformar nuestros datos de geolocalización.

x =
  housing_df
  |> DF.select(["latitude", "longitude"])
  |> Nx.stack(axis: 1)

kmeans = ClusterSimilarity.fit(x)

Si integramos los resultados de KMeans con nuestro dataframe, podemos visualizar en una gráfica los clusters obtenidos. Además, mediante una gráfica multi-layer podemos marcar los centros de masa de los agrupamientos. Podremos ver claramente que los centros de los clusters fueron posicionados en las zonas costosas y de alta densidad poblacional.

clusters_df =
  Nx.tensor(kmeans.clusters)
  |> DF.new()
  |> DF.rename(x1: "latitude", x2: "longitude")

df_with_kmeans_labels =
  housing_df
  |> DF.put("kmeans_labels", Nx.stack(kmeans.labels, axis: 1))

geo_vl =
  VL.new()
  |> VL.data_from_values(df_with_kmeans_labels)
  |> VL.mark(:point)
  |> VL.encode_field(
    :x,
    "longitude",
    type: :quantitative,
    scale: [
      zero: false
    ]
  )
  |> VL.encode_field(
    :y,
    "latitude",
    type: :quantitative,
    scale: [
      zero: false
    ]
  )

clusters_vl =
  VL.new()
  |> VL.data_from_values(clusters_df)
  |> VL.mark(:point,
    size: 70,
    color: "#0d0154",
    stroke_width: 10,
    opacity: 1
  )
  |> VL.encode_field(
    :x,
    "longitude",
    type: :quantitative,
    scale: [
      zero: false
    ]
  )
  |> VL.encode_field(
    :y,
    "latitude",
    type: :quantitative,
    scale: [
      zero: false
    ]
  )

VL.new(
  title: [
    text: "Geographic clusters found by KMeans algorithm"
  ],
  width: 700,
  height: 700,
  config: [
    axis: [
      grid: true,
      grid_color: "#dedede"
    ]
  ]
)
|> VL.layers([
  geo_vl
  |> VL.encode_field(
    :color,
    "kmeans_labels",
    type: :ordinal,
    scale: [
      scheme: "category10"
    ]
  ),
  clusters_vl
])

Ahora vamos a utilizar nuestro transformador para obtener las coeficientes Silueta de similaridad de cada punto geográfico.

silhouette_samples = ClusterSimilarity.transform(x, kmeans.labels)

De estos datos podemos obtener una variante de la visualización anterior, donde el color de cada punto indica la similaridad que tiene con su respectivo cluster. Esta gráfica nos ayuda a darnos una mejor idea de qué tan buena es la configuración de clusters obtenida de KMeans.

similarity_col = Nx.stack(silhouette_samples, axis: 1)

similarity_df =
  df_with_kmeans_labels
  |> DF.put("cluster_similarity", similarity_col)

VL.new(
  title: [
    text: "Silhouette Similarity for each sample"
  ],
  width: 700,
  height: 700,
  config: [
    axis: [
      grid: true,
      grid_color: "#dedede"
    ]
  ]
)
|> VL.layers([
  geo_vl
  |> VL.data_from_values(similarity_df)
  |> VL.encode_field(
    :color,
    "cluster_similarity",
    type: :quantitative,
    scale: [
      scheme: "plasma"
    ]
  ),
  clusters_vl
])

Tratamiento de features categóricos

Vimos que nuestro dataset contien un feature de tipo categórico ocean_proximity que tiene 5 posibles valores.

housing_df["ocean_proximity"]
|> S.frequencies()

La mayoría de los algoritmos de ML prefieren trabajar con números, por lo que es una buena idea convertir estas categorías a números. Una forma muy común de codificar valores categóricos cuando no existe una relación de órden entre los valores, es crear un atributo binario (booleano) para cada valor. A esta técnina se le denomina one-hot encoding.

housing_df
|> DF.mutate(%{"ocean_proximity" => cast(ocean_proximity, :category)})
|> DF.pull("ocean_proximity")
|> Nx.stack(axis: 1)
|> Nx.reshape({DF.n_rows(housing_df)})
|> Preprocessing.one_hot_encode(num_classes: 5)

Sin embargo, Explorer.DataFrame provee otra forma de hacer esto que es utilizando la función dummies/2.

housing_df
|> DF.dummies("ocean_proximity")

Pipelines de transformación

Como vimos, nuestros datos deben pasar por un montón de transformaciones antes de entrenar nuestro modelo. Cuando estos transformadores se ejecutan de manera secuencial sobre un conjunto de columnas se crea lo que se llama un pipeline de transformaciones. Un pipeline secuencial no es más que una composición de funciones secuencial.

Vamos a crear un módulo que baja a tierra algunos de los transformadores que vimos.

defmodule DataPipeline do
  def clean_data_pipeline(df, attrs, strategy \\ :nan) do
    for name <- attrs, into: [] do
      {name, S.fill_missing(df[name], strategy)}
    end
    |> DF.new()
  end

  def cast_data_pipeline(df, attrs, type \\ :category) do
    for name <- attrs, into: [] do
      {name, S.cast(df[name], type)}
    end
    |> DF.new()
  end

  def numerical_pipeline(df, attrs) do
    x =
      df
      |> clean_data_pipeline(attrs)
      |> DF.select(attrs)
      |> Nx.stack(axis: 1)

    x
    |> SimpleImputer.fit(strategy: :median)
    |> SimpleImputer.transform(x)
    |> Preprocessing.standard_scale(axes: [0])
  end

  def log_pipeline(df, attrs) do
    x =
      df
      |> clean_data_pipeline(attrs)
      |> DF.select(attrs)
      |> Nx.stack(axis: 1)
      |> Nx.log()

    x
    |> SimpleImputer.fit(strategy: :median)
    |> SimpleImputer.transform(x)
    |> Preprocessing.standard_scale(axes: [0])
  end

  def ratio_pipeline(df, attr1, attr2) do
    x =
      df
      |> clean_data_pipeline([attr1, attr2])
      |> DF.select([attr1, attr2])
      |> Nx.stack(axis: 1)

    x_new =
      x
      |> SimpleImputer.fit(strategy: :median)
      |> SimpleImputer.transform(x)

    col1 = Nx.take(x_new, 0, axis: 1)
    col2 = Nx.take(x_new, 1, axis: 1)

    Nx.divide(col1, col2)
    |> Nx.reshape({DF.n_rows(df), 1})
    |> Preprocessing.standard_scale(axes: [0])
  end

  def categorical_pipeline(df, attrs) do
    x =
      df
      |> cast_data_pipeline(attrs)
      |> Nx.stack(axis: 1)

    x
    |> SimpleImputer.fit(strategy: :mode)
    |> SimpleImputer.transform(x)
  end

  def cluster_similarity(df, attrs) do
    df
    |> DF.select(attrs)
    |> Nx.stack(axis: 1)
    |> ClusterSimilarity.fit_transform()
    |> Nx.stack(axis: 1)
  end
end

Cada uno de estos transformadores será utilizado para ciertas columnas del dataframe, y finalmente podremos integrar todas las columnas transformadas en un tensor para poder entrenar nuestro modelo. El siguiente módulo crea un función llamada preprocessing/1 que toma el dataframe de datos y devuelve un tensor con todas las columnas transformadas.

Notese que ejecutamos cada pipeline de transformaciones en un proceso nuevo, haciendo uso de la concurrencia que provee Elixir.

defmodule ColumnTransformer do
  alias DataPipeline, as: Pipeline

  def preprocessing(df) do
    cat_attrs = ["ocean_proximity"]
    geo_attrs = ["latitude", "longitude"]

    num_attrs = [
      "longitude",
      "latitude",
      "housing_median_age",
      "total_rooms",
      "total_bedrooms",
      "population",
      "households",
      "median_income"
    ]

    heavy_tail_attrs = [
      "total_bedrooms",
      "total_rooms",
      "population",
      "households",
      "median_income"
    ]

    Task.await_many(
      [
        Task.async(Pipeline, :ratio_pipeline, [df, "total_bedrooms", "total_rooms"]),
        Task.async(Pipeline, :ratio_pipeline, [df, "total_rooms", "households"]),
        Task.async(Pipeline, :ratio_pipeline, [df, "population", "households"]),
        Task.async(Pipeline, :log_pipeline, [df, heavy_tail_attrs]),
        Task.async(Pipeline, :cluster_similarity, [df, geo_attrs]),
        Task.async(Pipeline, :numerical_pipeline, [df, num_attrs]),
        Task.async(Pipeline, :categorical_pipeline, [df, cat_attrs])
      ],
      :infinity
    )
    |> Nx.concatenate(axis: 1)
  end
end

ColumnTransformer.preprocessing(housing_df)

Creando un conjunto de test

Una parte necesaria la hora de crear un modelo es evaluarlo, es decir, determinar la calidad de sus predicciones. Para esto, vamos a dividir nuestro conjunto original de datos en dos partes: un conjunto de entrenamiento (train set) y un conjunto de test (test set). A continuación definimos la función shuffle_and_split_data/2 que se encarga de esto.

defmodule Utils do
  def shuffle_and_split_data(%DF{} = dataframe, test_ratio \\ 0.20) do
    shuffled_data =
      dataframe
      # seed is for having reproducible results
      |> DF.shuffle(seed: 42)

    test_data_size =
      shuffled_data
      |> DF.n_rows()
      |> Kernel.*(test_ratio)
      |> trunc()

    train_data_size =
      shuffled_data
      |> DF.n_rows()
      |> Kernel.-(test_data_size)

    {DF.head(shuffled_data, train_data_size), DF.tail(shuffled_data, test_data_size)}
  end
end

Esta función se basa en muestreo aleatorio, lo que generalmente funciona bien en datasets que son grandes. Si no lo es, se corre un gran riesgo de introducir sesgo muestral (sampling bias).

Supongamos que al hablar con un experto, este nos dice que el ingreso medio en un CBG es dato de suma importancia a la hora de determinar el precio medio de una casa. Por ello, vamos a querer que nuestro test set sea representativo de las distintas categorías de ingreso medio que existen en nuestro conjunto. Para esto podríamos crear un nuevo feature income_category y estratificar de acuerdo a esta categoría. A esta técnia se le denomina muestreo estratificado (stratified sampling).

add_income_category_column = fn housing_df ->
  bin_median_income = fn income_value ->
    [0, 1.5, 3.0, 4.5, 6.0, :infinity]
    |> Enum.find_index(&amp;(income_value <= &amp;1))
    |> Integer.to_string()
  end

  income_cat =
    housing_df
    |> DF.pull("median_income")
    |> S.transform(bin_median_income)
    |> S.cast(:category)

  DF.put(housing_df, :income_category, income_cat)
end

Veamos un histograma de las categorías que creamos.

income_cat_df = add_income_category_column.(housing_df)

VL.new(
  title: [
    text: "Income category histogram"
  ],
  width: 500,
  height: 500,
  columns: 3,
  config: [
    axis: [
      grid: true,
      grid_color: "#dedede"
    ]
  ]
)
|> VL.data_from_values(income_cat_df)
|> VL.mark(:bar)
|> VL.encode_field(:x, "income_category", axis: [ticks: true])
|> VL.encode_field(:y, "value count", aggregate: :count)

Entrenando y evaluando nuestro modelo

Decidimos entrenar un modelo básico de regresión lineal. En el fondo, el modelo de regresión lineal resuelve el problema de mínimos cuadrados lineal (PMCL) asociado a el input x y los puntos y dados. Es decir que va a encontrar los coeficientes de una función de ajuste, de forma tal que se minimiza $||Xw - Y||^2$ donde $X$ es la matriz de datos preprocesados de entrenamiento e $Y$ son las etiquetas.

Como medida de rendiemiento (o error) vamos a tomar la distancia cuadrática media.

$$ \begin{equation} RMSE(y, \hat{y}) = \sqrt{\sum_{i=0}^{n - 1}{\frac{1}{n}(y_i - \hat{y}_i)^2}} \end{equation} $$

defmodule Housing do
  alias Scholar.Linear.LinearRegression
  alias Scholar.Metrics

  def train_and_evaluate_model(housing_df) do
    # split and shuffle data into training and test
    {train_data_df, test_data_df} = Utils.shuffle_and_split_data(housing_df)

    IO.puts("Train data size: #{DF.n_rows(train_data_df)}")
    IO.puts("Test data size: #{DF.n_rows(test_data_df)}")

    # labels
    y = Nx.concatenate(housing_df[["median_house_value"]])
    y_train = Nx.concatenate(train_data_df[["median_house_value"]])
    y_test = Nx.concatenate(test_data_df[["median_house_value"]])

    # preprocess both train and test data
    IO.puts("\nRunning data pipeline ...")

    [train_prepared_df, test_prepared_df] =
      Task.await_many(
        [
          Task.async(ColumnTransformer, :preprocessing, [train_data_df]),
          Task.async(ColumnTransformer, :preprocessing, [test_data_df])
        ],
        :infinity
      )

    x_train = train_prepared_df
    x_test = test_prepared_df

    IO.puts("\nPreprocessing done. Training model ...")

    # train linear model
    model = LinearRegression.fit(x_train, y_train)

    IO.puts("\nTraining Done.\n")

    # predict on test set
    predictions = LinearRegression.predict(model, x_test)

    # calculate errors
    rmse = Metrics.mean_square_error(y_test, predictions) |> Nx.sqrt()
    mae = Metrics.mean_absolute_error(y_test, predictions)

    IO.puts(":: performance report ::\n")
    IO.puts("> target mean (reference): #{Nx.mean(y) |> Nx.to_number()}")
    IO.puts("> RMSE: #{Nx.to_number(rmse)}")
    IO.puts("> MAE: #{Nx.to_number(mae)}\n")

    model
  end
end

model = Housing.train_and_evaluate_model(housing_df)