Масштабируемая Big Data система в Kubernetes с использованием Spark и Cassandra03.06.2023 19:01

  org.glassfish.web
  jakarta.servlet.jsp.jstl
  3.0.1


  jakarta.servlet
  jakarta.servlet-api
  6.0.0
  provided

  
  jakarta.servlet.jsp.jstl
  jakarta.servlet.jsp.jstl-api
  3.0.0


  javax.servlet
  javax.servlet-api
  4.0.1
  provided


  ch.qos.logback
  logback-classic
  1.2.9

wget https://developer.download.nvidia.com/compute/cuda/repos/ubuntu2204/x86_64/cuda-ubuntu2204.pin
sudo mv cuda-ubuntu2204.pin /etc/apt/preferences.d/cuda-repository-pin-600
wget https://developer.download.nvidia.com/compute/cuda/12.1.0/local_installers/cuda-repo-ubuntu2204-12-1-local_1
sudo dpkg -i cuda-repo-ubuntu2204-12-1-local_12.1.0-530.30.02-1_amd64.deb
sudo cp /var/cuda-repo-ubuntu2204-12-1-local/cuda-*-keyring.gpg /usr/share/keyrings/
sudo apt-get update
sudo apt-get -y install cuda

nvidia-smi
Sun Mar 19 20:56:18 2023       
+---------------------------------------------------------------------------------------+
| NVIDIA-SMI 530.30.02              Driver Version: 530.30.02    CUDA Version: 12.1     |
|-----------------------------------------+----------------------+----------------------+
| GPU  Name                  Persistence-M| Bus-Id        Disp.A | Volatile Uncorr. ECC |
| Fan  Temp  Perf            Pwr:Usage/Cap|         Memory-Usage | GPU-Util  Compute M. |
|                                         |                      |               MIG M. |
|=========================================+======================+======================|
|   0  NVIDIA GeForce GTX 1650         On | 00000000:02:00.0 Off |                  N/A |
|  0%   47C    P2               N/A /  75W|    332MiB /  4096MiB |      2%      Default |
|                                         |                      |                  N/A |
+-----------------------------------------+----------------------+----------------------+
                                                                                         
+---------------------------------------------------------------------------------------+
| Processes:                                                                            |
|  GPU   GI   CI        PID   Type   Process name                            GPU Memory |
|        ID   ID                                                             Usage      |
|=======================================================================================|
|    0   N/A  N/A      1346      G   /usr/lib/xorg/Xorg                           62MiB |
|    0   N/A  N/A      1905    C+G   ...libexec/gnome-remote-desktop-daemon      132MiB |
|    0   N/A  N/A      2013      G   /usr/bin/gnome-shell                        131MiB |
+---------------------------------------------------------------------------------------+

sudo -i
apt-get update && apt-get install -y nvidia-container-toolkit

distribution=$(. /etc/os-release;echo $ID$VERSION_ID)
curl -s -L https://nvidia.github.io/nvidia-docker/gpgkey | sudo apt-key add -
curl -s -L https://nvidia.github.io/nvidia-docker/$distribution/nvidia-docker.list | sudo tee /etc/apt/sources.list.d/nvidia-docker.list

sudo apt-get update && sudo apt-get install -y nvidia-container-toolkit
sudo systemctl restart docker

nvidia-ctk runtime configure --runtime=docker

version = 2
root = "/var/lib/containerd"
state = "/run/containerd"
oom_score = 0

[grpc]
  max_recv_message_size = 16777216
  max_send_message_size = 16777216

[debug]
  level = "info"

[metrics]
  address = ""
  grpc_histogram = false

[plugins]
  [plugins."io.containerd.grpc.v1.cri"]
    sandbox_image = "registry.k8s.io/pause:3.8"
    max_container_log_line_size = -1
    enable_unprivileged_ports = false
    enable_unprivileged_icmp = false
    [plugins."io.containerd.grpc.v1.cri".containerd]
      default_runtime_name = "nvidia"
      snapshotter = "overlayfs"

      [plugins."io.containerd.grpc.v1.cri".containerd.runtimes]
        [plugins."io.containerd.grpc.v1.cri".containerd.runtimes.nvidia]
          privileged_without_host_devices = false
          runtime_engine = ""
          runtime_root = ""
          runtime_type = "io.containerd.runc.v2"

          [plugins."io.containerd.grpc.v1.cri".containerd.runtimes.nvidia.options]
            BinaryName = "/usr/bin/nvidia-container-runtime"

        [plugins."io.containerd.grpc.v1.cri".containerd.runtimes.runc]
          runtime_type = "io.containerd.runc.v2"
          runtime_engine = ""
          runtime_root = ""
          base_runtime_spec = "/etc/containerd/cri-base.json"

          [plugins."io.containerd.grpc.v1.cri".containerd.runtimes.runc.options]
            systemdCgroup = true

    [plugins."io.containerd.grpc.v1.cri".registry]
      [plugins."io.containerd.grpc.v1.cri".registry.mirrors]
        [plugins."io.containerd.grpc.v1.cri".registry.mirrors."docker.io"]
          endpoint = ["https://registry-1.docker.io"]
      [plugins."io.containerd.grpc.v1.cri".registry.configs]
        [plugins."io.containerd.grpc.v1.cri".registry.configs."registry-1.docker.io".auth]
            username = "login"
            password = "password"
            auth = ""
            identitytoken = ""

kubectl create -f https://raw.githubusercontent.com/NVIDIA/k8s-device-plugin/v0.13.0/nvidia-device-plugin.yml

helm repo add nvidia https://nvidia.github.io/gpu-operator
helm repo update

kubectl create ns gpu-operator
     
helm install --wait gpu-operator \
     -n gpu-operator \
     nvidia/gpu-operator \
     --set operator.defaultRuntime=containerd

---
kind: Deployment
apiVersion: apps/v1
metadata:
  name: cuda-example
spec:
  replicas: 2
  selector:
    matchLabels:
      app: cuda-example
  template:
    metadata:
      labels:
        app: cuda-example
    spec:
      containers:
        - name: cuda-jdk8
          image: repo/cuda-jdk8-spark-3.3.2:v2
          resources:
            limits:
              cpu: 2
              memory: 4Gi
              nvidia.com/gpu: 1
            requests:
              cpu: "200m"
              memory: 0.5Gi
              nvidia.com/gpu: 1

ARG java_image_tag=17-jre
FROM ${java_image_tag}

ARG spark_uid=1001
ARG UID_GID=1001
ENV UID=${UID_GID}
ENV GID=${UID_GID}

ENV SPARK_RAPIDS_DIR=/opt/sparkRapidsPlugin
ENV SPARK_RAPIDS_PLUGIN_JAR=${SPARK_RAPIDS_DIR}/rapids-4-spark_2.12-23.04.0.jar

RUN set -ex && \
    sed -i 's/http:\/\/deb.\(.*\)/https:\/\/deb.\1/g' /etc/apt/sources.list && \
    apt-get update && \
    ln -s /lib /lib64 && \
    apt install -y bash tini libc6 libpam-modules libnss3 procps nano iputils-ping net-tools iptables sudo \
    wget software-properties-common build-essential libnss3-dev zlib1g-dev libgdbm-dev libncurses5-dev \
    libssl-dev libffi-dev libreadline-dev libsqlite3-dev libbz2-dev python3 && \
    mkdir -p /opt/spark && \
    mkdir -p /opt/spark/examples && \
    mkdir -p /opt/spark/conf && \
    mkdir -p /opt/spark/work-dir && \
    mkdir -p /opt/sparkRapidsPlugin && \
    touch /opt/spark/RELEASE && \
    rm /bin/sh && \
    ln -sv /bin/bash /bin/sh && \
    echo "auth required pam_wheel.so use_uid" >> /etc/pam.d/su && \
    chgrp root /etc/passwd && chmod ug+rw /etc/passwd && \
    rm -rf /var/cache/apt/*
RUN apt-get install libnccl2 libnccl-dev -y --allow-change-held-packages && rm -rf /var/cache/apt/*

COPY jars /opt/spark/jars
COPY rapids ${SPARK_RAPIDS_DIR}
COPY extraclasspath /opt/spark/extraclasspath
COPY bin /opt/spark/bin
COPY sbin /opt/spark/sbin
COPY conf /opt/spark/conf
COPY kubernetes/dockerfiles/spark/entrypoint.sh /opt/
COPY kubernetes/dockerfiles/spark/decom.sh /opt/
COPY kubernetes/tests /opt/spark/tests
COPY data /opt/spark/data
COPY datasets /opt/spark/

ENV SPARK_HOME /opt/spark

WORKDIR /opt/spark/work-dir
RUN chmod g+w /opt/spark/work-dir
RUN chmod a+x /opt/decom.sh
RUN chmod a+x /opt/entrypoint.sh
RUN chmod a+x /opt/sparkRapidsPlugin/getGpusResources.sh
RUN ls -lah /opt

RUN groupadd --gid $UID appuser && useradd --uid $UID --gid appuser --shell /bin/bash --create-home appuser
RUN mkdir /var/logs && chown -R appuser:appuser /var/logs
RUN mkdir /opt/spark/logs && chown -R appuser:appuser /opt/spark/
RUN chown -R appuser:appuser /tmp

RUN ls -lah /home/appuser
RUN touch /home/appuser/.bashrc

RUN echo -e '\
export SPARK_HOME=/opt/spark\n\
export PATH=$PATH:$SPARK_HOME/bin:$SPARK_HOME/sbin\
' > /home/appuser/.bashrc

RUN chown -R appuser:appuser /home/appuser

EXPOSE 4040
EXPOSE 8081

USER ${spark_uid}

ENTRYPOINT [ "/opt/entrypoint.sh" ]

#!/bin/bash
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

# echo commands to the terminal output
set -ex

# Check whether there is a passwd entry for the container UID
myuid=$(id -u)
mygid=$(id -g)
# turn off -e for getent because it will return error code in anonymous uid case
set +e
uidentry=$(getent passwd $myuid)
set -e

# If there is no passwd entry for the container UID, attempt to create one
if [ -z "$uidentry" ] ; then
    if [ -w /etc/passwd ] ; then
	echo "$myuid:x:$myuid:$mygid:${SPARK_USER_NAME:-anonymous uid}:$SPARK_HOME:/bin/false" >> /etc/passwd
    else
	echo "Container ENTRYPOINT failed to add passwd entry for anonymous UID"
    fi
fi

SPARK_CLASSPATH="$SPARK_CLASSPATH:${SPARK_HOME}/jars/*"
env | grep SPARK_JAVA_OPT_ | sort -t_ -k4 -n | sed 's/[^=]*=\(.*\)/\1/g' > /tmp/java_opts.txt
readarray -t SPARK_EXECUTOR_JAVA_OPTS < /tmp/java_opts.txt

if [ -n "$SPARK_EXTRA_CLASSPATH" ]; then
  SPARK_CLASSPATH="$SPARK_CLASSPATH:$SPARK_EXTRA_CLASSPATH"
fi

if ! [ -z ${PYSPARK_PYTHON+x} ]; then
    export PYSPARK_PYTHON
fi
if ! [ -z ${PYSPARK_DRIVER_PYTHON+x} ]; then
    export PYSPARK_DRIVER_PYTHON
fi

# If HADOOP_HOME is set and SPARK_DIST_CLASSPATH is not set, set it here so Hadoop jars are available to the executor.
# It does not set SPARK_DIST_CLASSPATH if already set, to avoid overriding customizations of this value from elsewhere e.g. Docker/K8s.
if [ -n "${HADOOP_HOME}"  ] && [ -z "${SPARK_DIST_CLASSPATH}"  ]; then
  export SPARK_DIST_CLASSPATH="$($HADOOP_HOME/bin/hadoop classpath)"
fi

if ! [ -z ${HADOOP_CONF_DIR+x} ]; then
  SPARK_CLASSPATH="$HADOOP_CONF_DIR:$SPARK_CLASSPATH";
fi

if ! [ -z ${SPARK_CONF_DIR+x} ]; then
  SPARK_CLASSPATH="$SPARK_CONF_DIR:$SPARK_CLASSPATH";
elif ! [ -z ${SPARK_HOME+x} ]; then
  SPARK_CLASSPATH="$SPARK_HOME/conf:$SPARK_CLASSPATH";
fi

case "$1" in
  driver)
    shift 1
    CMD=(
      "$SPARK_HOME/bin/spark-submit"
      --conf "spark.driver.bindAddress=$SPARK_DRIVER_BIND_ADDRESS"
      --deploy-mode client
      "$@"
    )
    ;;
  executor)
    shift 1
    CMD=(
      ${JAVA_HOME}/bin/java
      "${SPARK_EXECUTOR_JAVA_OPTS[@]}"
      -Xms$SPARK_EXECUTOR_MEMORY
      -Xmx$SPARK_EXECUTOR_MEMORY
      -cp "$SPARK_CLASSPATH:$SPARK_DIST_CLASSPATH"
      org.apache.spark.executor.CoarseGrainedExecutorBackend
      --driver-url $SPARK_DRIVER_URL
      --executor-id $SPARK_EXECUTOR_ID
      --cores $SPARK_EXECUTOR_CORES
      --app-id $SPARK_APPLICATION_ID
      --hostname $SPARK_EXECUTOR_POD_IP
      --resourceProfileId $SPARK_RESOURCE_PROFILE_ID
    )
    ;;

  *)
    echo "Non-spark-on-k8s command provided, proceeding in pass-through mode..."
    CMD=("$@")
    ;;
esac

# Execute the container CMD under tini for better hygiene
exec /usr/bin/tini -s -- "${CMD[@]}"

#!/usr/bin/env bash

#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#


set -ex
echo "Asked to decommission"
# Find the pid to signal
date | tee -a ${LOG}
WORKER_PID=$(ps -o pid -C java | tail -n 1| awk '{ sub(/^[ \t]+/, ""); print }')
echo "Using worker pid $WORKER_PID"
kill -s SIGPWR ${WORKER_PID}
# For now we expect this to timeout, since we don't start exiting the backend.
echo "Waiting for worker pid to exit"
# If the worker does exit stop blocking the cleanup.
timeout 60 tail --pid=${WORKER_PID} -f /dev/null
date
echo "Done"
date
sleep 1

#!/bin/bash
echo "=========== 1st stage ==========="
echo "Docker repo: $DOCKER_REPO"
echo "### build base cuda11.8 java8 image ###"
docker build -f Dockerfile-cuda-java8 -t localhost:5000/cuda11.8-jdk8:v1 .

echo "=========== 2nd stage ==========="
echo "### build Spark and PySpark images ###"
cd spark-3.3.2

./bin/docker-image-tool.sh -r localhost:5000 \
      -t jdk8-3.3.2 -b java_image_tag=localhost:5000/cuda11.8-jdk8:v1 \
      -p ./kubernetes/dockerfiles/spark/bindings/python/Dockerfile \
      -n build

cd ..

docker tag localhost:5000/spark-py:jdk8-3.3.2 $DOCKER_REPO/cuda-jdk8-spark-py-3.3.2:v2
docker tag localhost:5000/spark:jdk8-3.3.2 $DOCKER_REPO/cuda-jdk8-spark-3.3.2:v2

echo "=========== 3rt stage ==========="
echo "### push Spark and PySpark images ###"
docker push $DOCKER_REPO/cuda-jdk8-spark-py-3.3.2:v2
docker push $DOCKER_REPO/cuda-jdk8-spark-3.3.2:v2

FROM nvcr.io/nvidia/cuda:11.8.0-runtime-ubuntu22.04

ENV LANG='en_US.UTF-8' LANGUAGE='en_US:en' LC_ALL='en_US.UTF-8'

ARG DEBIAN_FRONTEND=noninteractive

RUN apt-get update && apt install -y bash tini libc6 libpam-modules libnss3 procps nano iputils-ping net-tools

RUN apt-get update && \
	apt-get install -y openjdk-8-jdk && \
	apt-get install -y ant && \
	apt-get clean && \
	rm -rf /var/lib/apt/lists/* && \
	rm -rf /var/cache/oracle-jdk8-installer;

RUN apt-get update && \
	apt-get install -y ca-certificates-java && \
	apt-get clean && \
	update-ca-certificates -f && \
	rm -rf /var/lib/apt/lists/* && \
	rm -rf /var/cache/oracle-jdk8-installer;

ENV JAVA_HOME /usr/lib/jvm/java-8-openjdk-amd64/
RUN export JAVA_HOME

CMD ["tail", "-f", "/dev/null"]

---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: spark-driver
  namespace: default

---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: spark-driver-crb
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: cluster-admin
subjects:
  - kind: ServiceAccount
    name: spark-driver
    namespace: default

kind: Service
apiVersion: v1
metadata:
  labels:
    app: ml-app
  name: ml-app
spec:
  type: ClusterIP
  clusterIP: None
  clusterIPs:
    - None
  internalTrafficPolicy: Cluster
  ipFamilies:
    - IPv4
  ipFamilyPolicy: SingleStack
  ports:
    - name: driver-port
      port: 33139
      targetPort: 33139
      protocol: TCP
    - name: block-manager-port
      port: 45029
      targetPort: 45029
      protocol: TCP
  selector:
    app: ml-app
  sessionAffinity: None

---
apiVersion: v1
kind: Pod
metadata:
  labels:
    app: ml-app
  name: ml-app
spec:
  serviceAccountName: spark-driver
  containers:
    - name: ml-app
      image: repo/ml:cuda11.8-jdk8
      imagePullPolicy: Always
      env:
        - name: "SPARK_EXECUTORS"
          value: "2"
        - name: "SPARK_MODE"
          value: "client"
        - name: "POSTGRES_HOST"
          value: "192.168.0.125"
        - name: "CASSANDRA_HOST"
          value: "192.168.0.125,192.168.0.150"
          # add your own env vars from ConfigMaps/Secrets
      resources:
        limits:
          cpu: "2"
          memory: 4Gi
        requests:
          cpu: "200m"
          memory: 0.5Gi
      ports:
        - containerPort: 4040
        - containerPort: 9090
        - containerPort: 33139
        - containerPort: 45029
---
kind: Service
apiVersion: v1
metadata:
  labels:
    app: ml-app
  name: ml-app
spec:
  type: ClusterIP
  clusterIP: None
  clusterIPs:
    - None
  internalTrafficPolicy: Cluster
  ipFamilies:
    - IPv4
  ipFamilyPolicy: SingleStack
  ports:
    - name: driver-port
      port: 33139
      targetPort: 33139
      protocol: TCP
    - name: block-manager-port
      port: 45029
      targetPort: 45029
      protocol: TCP
  selector:
    app: ml-app
  sessionAffinity: None

abstract class AbstractCassandraRepository constructor(
    private val sparkSession: SparkSession
) {

    companion object {
        internal const val keyspace: String = "instrument_data"
    }

    fun cassandraDataset(keyspace: String, table: String): Dataset {
        val cassandraDataset: Dataset = sparkSession.read()
            .format("org.apache.spark.sql.cassandra")
            .option("keyspace", keyspace)
            .option("table", table)
            .load()

        cassandraDataset.createOrReplaceTempView(table)
        return cassandraDataset
    }
}

create table instrument_data.time_series_history
(
    ticker      text,
    task_number uuid,
    datetime    timestamp,
    timeframe   text,
    close       decimal,
    high        decimal,
    low         decimal,
    open        decimal,
    volume      bigint,
    primary key (ticker, task_number, datetime, timeframe)
    )
    with compression = {'chunk_length_in_kb': '64', 'class': 'org.apache.cassandra.io.compress.LZ4Compressor'};

create index time_series_history_datetime_index
    on instrument_data.time_series_history (datetime);

create index time_series_history_task_number_index
    on instrument_data.time_series_history (task_number);

create index time_series_history_timeframe_index
    on instrument_data.time_series_history (timeframe);

abstract class AbstractIndicatorRepository constructor(
    sparkSession: SparkSession,
    private val table : String
) : AbstractCassandraRepository(sparkSession) {

    fun getBaseDataSet(
        ticker: String,
        taskNumber : UUID,
        dateStart : LocalDate,
        dateEnd : LocalDate
    ): Dataset {
        return cassandraDataset(table)
            .filter(
                functions.col("ticker").equalTo(ticker)
                    .and(functions.col("task_number").equalTo(taskNumber.toString()))
                    .and(functions.col("datetime").between(dateStart, dateEnd)))
    }
}

@Component
class TimeSeriesRepository(
    sparkSession: SparkSession
) : AbstractIndicatorRepository(sparkSession, "time_series_history") {

    fun getDataset(
        ticker: String,
        taskNumber : UUID,
        dateStart : LocalDate,
        dateEnd : LocalDate
    ) : Dataset {
        val dataset = getBaseDataSet(ticker, taskNumber, dateStart, dateEnd)
            .selectExpr(
                "to_date(datetime) as dateTime",
                "CAST(close AS Double) as close"
            )

        dataset.createOrReplaceTempView("ts")
        return dataset
    }
}

private fun getMainDataset(
    ticker : String,
    taskNumber : UUID,
    dateStart : LocalDate,
    dateEnd : LocalDate,
    currentOffset : Int,
    batchSize : Int
) : Dataset {
    val timeSeries = timeSeriesRepository
        .getDataset(ticker, taskNumber, dateStart, dateEnd, currentOffset, batchSize)
        .`as`("ts")
    val emaDataSet = emaRepository
        .getEmaDataSet(ticker, dateStart, dateEnd, currentOffset, batchSize)
        .`as`("ema")
    val stochasticDataset = stochasticRepository
        .getStochasticDataSet(ticker, dateStart, dateEnd, currentOffset, batchSize)
        .`as`("stoch")
    val bBandsDataset = bBandIndicatorRepository
        .getBBandsDataSet(ticker, dateStart, dateEnd, currentOffset, batchSize)
        .`as`("bb")
    val macdDataset = macdRepository
        .getMacdDataSet(ticker, dateStart, dateEnd, currentOffset, batchSize)
        .`as`("macd")
    val rsiDataset = rsiRepository
        .getRsiDataSet(ticker, dateStart, dateEnd, currentOffset, batchSize)
        .`as`("rsi")
    val smaDataset = smaRepository
        .getSmaDataSet(ticker, dateStart, dateEnd, currentOffset, batchSize)
        .`as`("sma")
    val willrDataset = willrRepository
        .getWillrDataSet(ticker, dateStart, dateEnd, currentOffset, batchSize)
        .`as`("willr")

    return combineDatasets(
        timeSeries, emaDataSet, stochasticDataset, bBandsDataset, macdDataset, rsiDataset, smaDataset, willrDataset
    )
}

Здесь у Java/Kotlin разработчика может возникнуть желание сделать запрос данных через CompletableFuture> и выполнить их все в одном тредпуле. Однако, не стоит этого делать. Спарк сам распараллеливает работу и создает FutureTask’s, программисту не стоит распараллеливать код на разные потоки.

private fun combineDatasets(
    timeSeries : Dataset,
    emaDataSet : Dataset,
    stochasticDataset : Dataset,
    bBandsDataset : Dataset,
    macdDataset : Dataset,
    rsiDataset : Dataset,
    smaDataset : Dataset,
    willrDataset : Dataset
): Dataset {

    val result = timeSeries
        .join(emaDataSet, timeSeries.col("datetime")
                .equalTo(emaDataSet.col("datetime")), "leftouter")
        .join(stochasticDataset, timeSeries.col("datetime")
                .equalTo(stochasticDataset.col("datetime")), "leftouter")
        .join(bBandsDataset, timeSeries.col("datetime")
                .equalTo(bBandsDataset.col("datetime")), "leftouter")
        .join(macdDataset, timeSeries.col("datetime")
                .equalTo(macdDataset.col("datetime")), "leftouter")
        .join(rsiDataset, timeSeries.col("datetime")
                .equalTo(rsiDataset.col("datetime")), "leftouter")
        .join(smaDataset, timeSeries.col("datetime")
                .equalTo(smaDataset.col("datetime")), "leftouter")
        .join(willrDataset, timeSeries.col("datetime")
                .equalTo(willrDataset.col("datetime")), "leftouter")
        .selectExpr(
            "ts.dateTime as dateTime",
            "ema.emaTimePeriod as emaTimePeriod",
            "ema.ema as ema",
            "stoch.fastKPeriod as fastKPeriod",
            "stoch.slowKPeriod as slowKPeriod",
            "stoch.slowDPeriod as slowDPeriod",
            "stoch.slowD as slowD",
            "stoch.slowK as slowK",
            "bb.timePeriod as bbTimePeriod",
            "bb.sd as bbSd",
            "bb.lowerBand as lowerBand",
            "bb.middleBand as middleBand",
            "bb.upperBand as upperBand",
            "macd.signalPeriod as signalPeriod",
            "macd.fastPeriod as fastPeriod",
            "macd.slowPeriod as slowPeriod",
            "macd.macd as macd",
            "macd.macdHist as macdHist",
            "macd.macdSignal as macdSignal",
            "rsi.timePeriod as rsiTimePeriod",
            "rsi.rsi as rsi",
            "sma.timePeriod as smaTimePeriod",
            "sma.sma as sma",
            "willr.timePeriod as willrTimePeriod",
            "willr.willr as willr",
            "ts.close as close"
        )

    val windowSpec = Window.orderBy(functions.asc("dateTime"))
    return result
        .withColumn("id", functions.row_number().over(windowSpec))
        .filter(functions.col("dateTime").isNotNull)
}