Apache Iceberg and time travel queries with Scala/Spark

Iceberg provides time travel functionality, allowing you to query the state of a table at a previous point in time. You can query the table as it existed at a specific snapshot or as of a certain timestamp.

Example: querying an Iceberg table at a specific snapshot

Each time you make changes (like insert or update) to an Iceberg table, a snapshot is created. You can query data from a particular snapshot.

def timeTravelBySnapshot(spark: SparkSession, snapshotId: Long): Unit = {
  val timeTravelDF = spark.read
    .option("snapshot-id", snapshotId)
    .table("spark_catalog.default.iceberg_table")
  
  timeTravelDF.show()
  println(s"Data retrieved for snapshot: $snapshotId")
}

We can find the snapshot ID using:

val historyDF = spark.sql("SELECT * FROM spark_catalog.default.iceberg_table.history")
historyDF.show()

Once we have the snapshot ID, pass it to timeTravelBySnapshot() to query the data as of that snapshot.

Example: querying an Iceberg table as of a timestamp

def timeTravelByTimestamp(spark: SparkSession, timestamp: String): Unit = {
  val timeTravelDF = spark.read
    .option("as-of-timestamp", timestamp)
    .table("spark_catalog.default.iceberg_table")
  
  timeTravelDF.show()
  println(s"Data retrieved as of timestamp: $timestamp")
}
// Example timestamp format: "2024-10-16T15:00:00.000Z"
// To execute:
timeTravelByTimestamp(spark, "2024-10-16T15:00:00.000Z")

In this way we can to handle time travel queries using Apache Iceberg and Spark.

How to create neural network with Scala?

Creating a neural network in Scala typically involves either building the network from scratch using linear algebra or leveraging libraries like Breeze or DeepLearning4j (which has Scala bindings). Here’s a basic outline of how you could create a neural network in Scala:

Dependencies

First, we’ll need some libraries for mathematical operations. Using sbt (Simple Build Tool) we can add this to our build.sbt file:

libraryDependencies ++= Seq(
  "org.scalanlp" %% "breeze" % "1.2",
  "org.deeplearning4j" % "deeplearning4j-core" % "1.0.0-beta7"
)

Creating a neural network from scratch

Here’s a very simplified feed-forward neural network (without any optimization or training) built from scratch using Breeze for matrix operations:

  • Input layer, weights, and biases initialization – we need a way to store the weights and biases between layers.
  • Activation functions –implement a simple activation function (e.g., sigmoid).
  • Feedforward – implement the feedforward process where the input data is passed through the network.
  • Backpropagation (if we need training).
import breeze.linalg._
import breeze.numerics._
// Activation function (sigmoid)
def sigmoid(x: DenseMatrix[Double]): DenseMatrix[Double] = {
  1.0 /:/ (exp(-x) + 1.0)
}
// Derivative of sigmoid
def sigmoidDerivative(x: DenseMatrix[Double]): DenseMatrix[Double] = {
  x :* (1.0 - x)
}
// Feedforward step
def feedForward(input: DenseMatrix[Double], weights: Seq[DenseMatrix[Double]], biases: Seq[DenseMatrix[Double]]): Seq[DenseMatrix[Double]] = {
  var activations = Seq(input)
  for (i <- weights.indices) {
    val z = (weights(i) * activations.last) + biases(i)
    val a = sigmoid(z)
    activations = activations :+ a
  }
  activations
}
// Backpropagation (simplified version for gradient calculation, not full)
def backpropagation(expected: DenseMatrix[Double], activations: Seq[DenseMatrix[Double]], weights: Seq[DenseMatrix[Double]]): (Seq[DenseMatrix[Double]], Seq[DenseMatrix[Double]]) = {
  val error = activations.last - expected
  // Backpropagation steps here (update weights and biases)
  (weights, activations)
}
// Example usage
val inputLayerSize = 2
val hiddenLayerSize = 3
val outputLayerSize = 1
// Initialize weights and biases randomly
val weights1 = DenseMatrix.rand[Double](hiddenLayerSize, inputLayerSize)
val weights2 = DenseMatrix.rand[Double](outputLayerSize, hiddenLayerSize)
val bias1 = DenseMatrix.rand[Double](hiddenLayerSize, 1)
val bias2 = DenseMatrix.rand[Double](outputLayerSize, 1)
// Input data
val X = DenseMatrix((0.0, 1.0), (1.0, 0.0)) // 2 input features
val Y = DenseMatrix(1.0, 0.0)               // Expected output
// Feedforward
val weights = Seq(weights1, weights2)
val biases = Seq(bias1, bias2)
val activations = feedForward(X, weights, biases)
// Print activations
activations.foreach(println)

The first step is importing the necessary libraries. Here, we use Breeze, a Scala library that provides powerful tools for numerical processing like linear algebra, which is essential for machine learning.

  • breeze.linalg._: This gives us tools for handling matrices and vectors (like DenseMatrix, which is similar to a 2D array or matrix),
  • breeze.numerics._: This provides basic mathematical functions like exp (exponentiation).

In this neural network, we use the sigmoid function as the activation function. The sigmoid squashes input values between 0 and 1, which is useful in controlling the output of each neuron in a neural network.

def sigmoid(x: DenseMatrix[Double]): DenseMatrix[Double] = {
  1.0 /:/ (exp(-x) + 1.0)
}

The sigmoid derivative is used for calculating the gradient during backpropagation. The derivative is needed to adjust weights and biases during training.

def sigmoidDerivative(x: DenseMatrix[Double]): DenseMatrix[Double] = {
  x :* (1.0 - x)
}

The feedforward process is where the input passes through the network layer by layer to compute the output. At each layer, we multiply the input by the weights, add the biases, and then apply the activation function (sigmoid in this case).

Steps in feedForward:

Input: Start with the input as the initial activation.

For Each Layer:

  • calculate the weighted sum (z = W * a + b), where W is the weight matrix, a is the activation from the previous layer, and b is the bias,
  • apply the sigmoid activation function to z to get the next activation,
  • add the new activation to the sequence of activations.

This is a placeholder for the backpropagation step. Backpropagation is a method used to calculate the gradient (partial derivatives) needed for adjusting the weights and biases during training, based on the difference between predicted and actual values.

Breakdown

Neural Network Structure:

  • Input layer size: 2 neurons (for 2 input features).
  • Hidden layer size: 3 neurons.
  • Output layer size: 1 neuron (since we are outputting 1 value).

Random Weights and Biases:

  • weights1: Matrix of random values for the first layer, with size [3, 2] (3 neurons, 2 inputs),
  • weights2: Matrix of random values for the second layer, with size [1, 3] (1 output, 3 neurons in the hidden layer),
  • bias1 and bias2: Random biases for the layers.

Input Data:

  • X represents 2 inputs: (0.0, 1.0) and (1.0, 0.0).
  • Y is the expected output: 1.0, 0.0.

Feedforward Execution:

  • feedForward(X, weights, biases) computes the activations of each layer.

Printing the Activations:

  • After passing through the network, the activations of each layer are printed.

Using libraries

If we use DeepLearning4j (DL4J), the code will be simpler since the library abstracts most of the math and logic. We can build a neural network like this:

import org.deeplearning4j.nn.conf._
import org.deeplearning4j.nn.conf.layers._
import org.deeplearning4j.nn.multilayer._
import org.nd4j.linalg.factory.Nd4j
import org.nd4j.linalg.dataset.api.iterator.DataSetIterator
import org.deeplearning4j.datasets.iterator.impl.IrisDataSetIterator
// Build the configuration
val conf: MultiLayerConfiguration = new NeuralNetConfiguration.Builder()
  .seed(123)
  .updater(new Nesterovs(0.1, 0.9))
  .list()
  .layer(0, new DenseLayer.Builder().nIn(4).nOut(3)
    .activation("relu")
    .build())
  .layer(1, new OutputLayer.Builder(LossFunctions.LossFunction.MCXENT)
    .activation("softmax")
    .nIn(3).nOut(3)
    .build())
  .build()
// Initialize the model
val model = new MultiLayerNetwork(conf)
model.init()
// Load dataset (Iris for simplicity)
val irisIter: DataSetIterator = new IrisDataSetIterator(150, 150)
// Train the model
for (i <- 0 until 1000) {
  model.fit(irisIter)
}
// Evaluate or test model

Summary

This code demonstrates how to build a simple neural network using the basic principles of feedforward and (partially) backpropagation. Here’s what happens:

  • The input data is passed through the network, layer by layer, with weights and biases applied.
  • The sigmoid activation function ensures the values are normalized between 0 and 1.
  • Although the backpropagation logic is not fully implemented, this structure would allow for training once it is in place.

This example is highly simplified, but it provides a foundation for building more complex networks with multiple hidden layers, different activation functions, and learning algorithms.

Apache Iceberg partitioning data with Scala/Spark

Introduction

Partitioning helps improve query performance by reducing the amount of data scanned during read operations. Iceberg allows dynamic partitioning and supports various partition types such as identity, bucket, and truncate.

Example: partitioning data by a column (e.g., Age)

Let’s partition the table by a column like age (bucket partitioning).

def partitionTable(spark: SparkSession): Unit = {
  // Create a new partitioned table
  spark.sql(
    """
      |CREATE TABLE IF NOT EXISTS spark_catalog.default.iceberg_partitioned_table (
      | id BIGINT,
      | name STRING,
      | age INT
      |) USING iceberg
      | PARTITIONED BY (bucket(5, age))
      |""".stripMargin)
  println("Partitioned Iceberg table created!")
}
// To execute:
partitionTable(spark)

Here, we’ve partitioned the table using the age column into 5 buckets. This is useful for optimizing queries when filtering based on age.

Query example with partitioned table:

When you query partitioned data, Iceberg only scans the partitions relevant to the query:

def queryPartitionedTable(spark: SparkSession): Unit = {
  val partitionedDF = spark.sql("SELECT * FROM spark_catalog.default.iceberg_partitioned_table WHERE age > 25")
  partitionedDF.show()
}
// To execute:
queryPartitionedTable(spark)