Cloud

GCP, Dataproc

Here is a demo of running a learning algorithm on Google Cloud Platform’s Dataproc.

 gcloud dataproc batches submit pyspark \
     pyspark-demo.py \
     --region=us-central1 \
     --version=2.0 \
     --deps-bucket=my-dataproc-deps-bucket \
     --py-files=pybbn-3.2.3-py3.9.egg,pysparkbbn-0.0.3-py3.9.egg \
     -- --input gs://my-gcs-folder/data/data-binary.csv

The driver code pyspark-demo.py is as follows.

import argparse
import json
import sys
from typing import List

from pybbn.pptc.inferencecontroller import InferenceController
from pyspark.sql import SparkSession

from pyspark_bbn.discrete.bbn import get_bbn, get_darkstar_data, get_pybbn_data
from pyspark_bbn.discrete.data import DiscreteData
from pyspark_bbn.discrete.plearn import ParamLearner
from pyspark_bbn.discrete.scblearn import Naive


def parse_pargs(args: List[str]) -> argparse.Namespace:
    """Parse command-line arguments.

    :param args: command-line arguments.
    :type args: List[str]
    :return: parsed arguments.
    :rtype: argparse.Namespace
    """
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", type=str, required=True, help="Input CSV file")

    return parser.parse_args(args)


def start(input_path: str) -> None:
    """Start the PySpark demo.

    :param input_path: path to the input CSV file.
    :type input_path: str
    :return: None
    :rtype: None
    """
    spark = SparkSession.builder.appName("learn-naive").getOrCreate()

    sdf = spark.read.option("header", True).option("inferSchema", True).csv(input_path)

    print("data schema")
    sdf.printSchema()

    print("")
    print("data sample")
    sdf.show(10)

    data = DiscreteData(sdf)
    naive = Naive(data, "e")
    g = naive.get_network()

    print("")
    print("nodes")
    print("-" * 10)
    for n in g.nodes():
        print(f"{n}")

    print("")
    print("edges")
    print("-" * 10)
    for pa, ch in g.edges():
        print(f"{pa} -> {ch}")

    param_learner = ParamLearner(data, g)
    p = param_learner.get_params()

    print("")
    print("params")
    print("-" * 10)
    print(json.dumps(p, indent=2))

    print("")
    print("py-bbn, posteriors")
    print("-" * 10)
    bbn = get_bbn(g, p, data.get_profile())
    join_tree = InferenceController.apply(bbn)

    for node, posteriors in join_tree.get_posteriors().items():
        p_str = ", ".join([f"{val}={prob:.5f}" for val, prob in posteriors.items()])
        print(f"{node} : {p_str}")

    print("")
    print("py-bbn, data")
    print("-" * 10)
    pybbn_data = get_pybbn_data(g, p, data.get_profile())
    print(json.dumps(pybbn_data, indent=2))

    print("")
    print("darkstar, data")
    print("-" * 10)
    darkstar_data = get_darkstar_data(g, p, data.get_profile())
    print(json.dumps(darkstar_data, indent=2))


if __name__ == "__main__":
    args = parse_pargs(sys.argv[1:])

    input_path = args.input
    start(input_path)

The output should look something like the following.

Batch [7732907e5b8843f98c5f6c2ccffbd85d] submitted.
Using the default container image
Waiting for container log creation
PYSPARK_PYTHON=/opt/dataproc/conda/bin/python
JAVA_HOME=/usr/lib/jvm/temurin-17-jdk-amd64
SPARK_EXTRA_CLASSPATH=
:: loading settings :: file = /etc/spark/conf/ivysettings.xml
data schema
root
 |-- a: integer (nullable = true)
 |-- b: integer (nullable = true)
 |-- c: integer (nullable = true)
 |-- d: integer (nullable = true)
 |-- e: integer (nullable = true)


data sample
+---+---+---+---+---+
|  a|  b|  c|  d|  e|
+---+---+---+---+---+
|  1|  0|  0|  0|  0|
|  1|  0|  0|  0|  0|
|  1|  0|  0|  1|  1|
|  0|  0|  0|  0|  1|
|  0|  0|  0|  0|  0|
|  1|  0|  0|  0|  1|
|  1|  0|  0|  0|  0|
|  1|  0|  0|  1|  1|
|  0|  0|  0|  0|  1|
|  1|  0|  0|  0|  0|
+---+---+---+---+---+
only showing top 10 rows


nodes
----------
e
a
b
c
d

edges
----------
e -> a
e -> b
e -> c
e -> d

params
----------
{
  "e": [
    {
      "e": "0",
      "__p__": 0.7416
    },
    {
      "e": "1",
      "__p__": 0.2584
    }
  ],
  "a": [
    {
      "a": "0",
      "e": "0",
      "__p__": 0.18743257820927725
    },
    {
      "a": "1",
      "e": "0",
      "__p__": 0.8125674217907227
    },
    {
      "a": "0",
      "e": "1",
      "__p__": 0.1946594427244582
    },
    {
      "a": "1",
      "e": "1",
      "__p__": 0.8053405572755418
    }
  ],
  "b": [
    {
      "b": "0",
      "e": "0",
      "__p__": 0.8015102481121898
    },
    {
      "b": "1",
      "e": "0",
      "__p__": 0.19848975188781015
    },
    {
      "b": "0",
      "e": "1",
      "__p__": 0.8068885448916409
    },
    {
      "b": "1",
      "e": "1",
      "__p__": 0.19311145510835914
    }
  ],
  "c": [
    {
      "c": "0",
      "e": "0",
      "__p__": 0.6863538295577131
    },
    {
      "c": "1",
      "e": "0",
      "__p__": 0.31364617044228693
    },
    {
      "c": "0",
      "e": "1",
      "__p__": 0.6884674922600619
    },
    {
      "c": "1",
      "e": "1",
      "__p__": 0.31153250773993807
    }
  ],
  "d": [
    {
      "d": "0",
      "e": "0",
      "__p__": 0.9704692556634305
    },
    {
      "d": "1",
      "e": "0",
      "__p__": 0.02953074433656958
    },
    {
      "d": "0",
      "e": "1",
      "__p__": 0.2921826625386997
    },
    {
      "d": "1",
      "e": "1",
      "__p__": 0.7078173374613003
    }
  ]
}

py-bbn, posteriors
----------
e : 0=0.74160, 1=0.25840
a : 0=0.18930, 1=0.81070
b : 0=0.80290, 1=0.19710
c : 0=0.68690, 1=0.31310
d : 0=0.79520, 1=0.20480

py-bbn, data
----------
{
  "nodes": {
    "0": {
      "probs": [
        0.7416,
        0.2584
      ],
      "variable": {
        "id": 0,
        "name": "e",
        "values": [
          "0",
          "1"
        ]
      }
    },
    "1": {
      "probs": [
        0.18743257820927725,
        0.8125674217907227,
        0.1946594427244582,
        0.8053405572755418
      ],
      "variable": {
        "id": 1,
        "name": "a",
        "values": [
          "0",
          "1"
        ]
      }
    },
    "2": {
      "probs": [
        0.8015102481121898,
        0.19848975188781015,
        0.8068885448916409,
        0.19311145510835914
      ],
      "variable": {
        "id": 2,
        "name": "b",
        "values": [
          "0",
          "1"
        ]
      }
    },
    "3": {
      "probs": [
        0.6863538295577131,
        0.31364617044228693,
        0.6884674922600619,
        0.31153250773993807
      ],
      "variable": {
        "id": 3,
        "name": "c",
        "values": [
          "0",
          "1"
        ]
      }
    },
    "4": {
      "probs": [
        0.9704692556634305,
        0.02953074433656958,
        0.2921826625386997,
        0.7078173374613003
      ],
      "variable": {
        "id": 4,
        "name": "d",
        "values": [
          "0",
          "1"
        ]
      }
    }
  },
  "edges": [
    {
      "pa": 0,
      "ch": 1
    },
    {
      "pa": 0,
      "ch": 2
    },
    {
      "pa": 0,
      "ch": 3
    },
    {
      "pa": 0,
      "ch": 4
    }
  ]
}

darkstar, data
----------
{
  "nodes": {
    "e": {
      "id": 0,
      "values": {
        "0": 0,
        "1": 1
      }
    },
    "a": {
      "id": 1,
      "values": {
        "0": 0,
        "1": 1
      }
    },
    "b": {
      "id": 2,
      "values": {
        "0": 0,
        "1": 1
      }
    },
    "c": {
      "id": 3,
      "values": {
        "0": 0,
        "1": 1
      }
    },
    "d": {
      "id": 4,
      "values": {
        "0": 0,
        "1": 1
      }
    }
  },
  "edges": [
    {
      "parent": "e",
      "child": "a"
    },
    {
      "parent": "e",
      "child": "b"
    },
    {
      "parent": "e",
      "child": "c"
    },
    {
      "parent": "e",
      "child": "d"
    }
  ],
  "parameters": {
    "e": [
      {
        "e": "0",
        "__p__": 0.7416
      },
      {
        "e": "1",
        "__p__": 0.2584
      }
    ],
    "a": [
      {
        "a": "0",
        "e": "0",
        "__p__": 0.18743257820927725
      },
      {
        "a": "1",
        "e": "0",
        "__p__": 0.8125674217907227
      },
      {
        "a": "0",
        "e": "1",
        "__p__": 0.1946594427244582
      },
      {
        "a": "1",
        "e": "1",
        "__p__": 0.8053405572755418
      }
    ],
    "b": [
      {
        "b": "0",
        "e": "0",
        "__p__": 0.8015102481121898
      },
      {
        "b": "1",
        "e": "0",
        "__p__": 0.19848975188781015
      },
      {
        "b": "0",
        "e": "1",
        "__p__": 0.8068885448916409
      },
      {
        "b": "1",
        "e": "1",
        "__p__": 0.19311145510835914
      }
    ],
    "c": [
      {
        "c": "0",
        "e": "0",
        "__p__": 0.6863538295577131
      },
      {
        "c": "1",
        "e": "0",
        "__p__": 0.31364617044228693
      },
      {
        "c": "0",
        "e": "1",
        "__p__": 0.6884674922600619
      },
      {
        "c": "1",
        "e": "1",
        "__p__": 0.31153250773993807
      }
    ],
    "d": [
      {
        "d": "0",
        "e": "0",
        "__p__": 0.9704692556634305
      },
      {
        "d": "1",
        "e": "0",
        "__p__": 0.02953074433656958
      },
      {
        "d": "0",
        "e": "1",
        "__p__": 0.2921826625386997
      },
      {
        "d": "1",
        "e": "1",
        "__p__": 0.7078173374613003
      }
    ]
  }
}
Batch [7732907e5b8843f98c5f6c2ccffbd85d] finished.
metadata:
  '@type': type.googleapis.com/google.cloud.dataproc.v1.BatchOperationMetadata
  batch: projects/rocketvector/locations/us-central1/batches/7732907e5b8843f98c5f6c2ccffbd85d
  batchUuid: 96fc6be5-bb7b-45cf-9123-669ff6fa1a05
  createTime: '2023-06-08T08:17:28.049693Z'
  description: Batch
  operationType: BATCH
name: projects/rocketvector/regions/us-central1/operations/61f78fed-c3c5-38b6-a0d8-1b492d3d210d

AWS, Spark Processing Job

Here is a demo of running a learning algorithm on AWS Sagemaker using Spark Processing Jobs. We can kick off the job by running a driver program on the command line.

1 python driver.py

The driver program driver.py looks like the following. Note that you should build a Docker image and push it to ECR. That Docker image should have all the APIs you need installed.

from sagemaker.spark.processing import PySparkProcessor

job = PySparkProcessor(
    **{
        "role": "your_aws_role",
        "instance_type": "ml.c5.xlarge",
        "instance_count": 1,
        "base_job_name": "pyspark-bbn",
        "image_uri": "your_docker_image_uri",
    }
)

job.run(
    submit_app="learn.py",
    arguments=[
        "--input_bucket",
        "your_input_bucket",
        "--input_key",
        "temp/data-from-structure.csv",
        "--output_bucket",
        "your_output_bucket",
        "--output_key",
        "temp/output/data-from-structure/bbn-naive.json",
        "--clazz_var",
        "your_clazz_variable",
    ],
)

The learning program learn.py looks like the following. This learning program simply learns a Naive bayes nodel.

import argparse
import json
import logging
import sys
from typing import List

import boto3
from pybbn.graph.dag import Bbn
from pyspark.sql import SparkSession

from pyspark_bbn.discrete.bbn import get_bbn
from pyspark_bbn.discrete.data import DiscreteData
from pyspark_bbn.discrete.plearn import ParamLearner
from pyspark_bbn.discrete.scblearn import Naive

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
spark = SparkSession.builder.appName("learn-naive").getOrCreate()


def parse_pargs(args: List[str]) -> argparse.Namespace:
    """Parse command-line arguments.

    :param args: command-line arguments.
    :type args: List[str]
    :return: parsed arguments.
    :rtype: argparse.Namespace
    """
    parser = argparse.ArgumentParser()
    parser.add_argument("--input_bucket", type=str, required=True)
    parser.add_argument("--input_key", type=str, required=True)
    parser.add_argument("--output_bucket", type=str, required=True)
    parser.add_argument("--output_key", type=str, required=True)
    parser.add_argument("--clazz_var", type=str, default=None)

    return parser.parse_args(args)


def upload(src: str, bucket: str, key: str) -> None:
    """Upload a file to Amazon S3.

    :param src: local file path.
    :type src: str
    :param bucket: destination S3 bucket.
    :type bucket: str
    :param key: object key within the bucket.
    :type key: str
    :return: None
    :rtype: None
    """
    s3 = boto3.client("s3")
    response = s3.upload_file(src, bucket, key)
    logging.info(f"uploaded {src} to {bucket}/{key}")
    logging.info(f"response={response}")


if __name__ == "__main__":
    args = parse_pargs(sys.argv[1:])

    logging.info("Job Starting")

    logging.info("Parsed Arguments")
    logging.info(f"args={args}")

    data_path = f"s3://{args.input_bucket}/{args.input_key}"
    logging.info(f"data_path={data_path}")

    sdf = spark.read.option("header", "true").csv(data_path)

    n_rows, n_cols = sdf.count(), len(sdf.columns)
    logging.info("Read Data")
    logging.info(f"data dimensions: rows={n_rows:,}, cols={n_cols:,}")

    data = DiscreteData(sdf)

    structure_learner = Naive(data, args.clazz_var)
    logging.info("Learned Structure")
    logging.info(f"structure learn type: {type(structure_learner)}")

    g = structure_learner.get_network()
    logging.info(f"learned structure: nodes={len(g.nodes())}, edges={len(g.edges())}")

    parameter_learner = ParamLearner(data, g)
    p = parameter_learner.get_params()
    logging.info("Learned Parameters")
    logging.info(f"learned parameters: {len(p)}")

    bbn = get_bbn(g, p, data.get_profile())
    logging.info("Constructed BBN")
    logging.info(f"bbn: nodes={len(bbn.nodes)}, edges={len(bbn.edges)}")

    j_data = json.dumps(Bbn.to_dict(bbn), indent=2)
    j_path = "/tmp/bbn.json"
    with open(j_path, "w") as f:
        f.write(j_data)

    logging.info("Serialized BBN")
    logging.info(f"saved bbn to {j_path}")
    upload(j_path, args.output_bucket, args.output_key)

    spark.stop()

    print("Finished")

Azure, Machine Learning

Here is a demo of running a Serverless Spark Standalone job in Azure Machine Learning. We can submit a job via the Azure CLI as follows.

az ml job create \
             -f learn-naive.yaml \
             -g your_resource_group \
             -w your_aml_workspace \
             --subscription your_subscription_id

The YAML file learn-naive.yaml looks like the following.

$schema: http://azureml/sdk-2-0/SparkJob.json
type: spark

code: ./src
entry:
  file: learn-naive.py

py_files:
  - pybbn-3.2.3-py3.9.egg
  - pysparkbbn-0.0.3-py3.9.egg

conf:
  spark.driver.cores: 1
  spark.driver.memory: 2g
  spark.executor.cores: 2
  spark.executor.memory: 2g
  spark.executor.instances: 2

inputs:
  input_data:
    type: uri_file
    path: abfss://your_container@your_storage_account.dfs.core.windows.net/input/data-binary.csv
    mode: direct
  clazz: "e"

args: >-
  --input_data ${{inputs.input_data}} --clazz ${{inputs.clazz}}

identity:
  type: user_identity

resources:
  instance_type: standard_e4s_v3
  runtime_version: "3.2"

The Python program learn-naive.py looks like the following.

import argparse
import json
import sys
from typing import List

from pybbn.pptc.inferencecontroller import InferenceController
from pyspark.sql import SparkSession

from pyspark_bbn.discrete.bbn import get_bbn, get_darkstar_data, get_pybbn_data
from pyspark_bbn.discrete.data import DiscreteData
from pyspark_bbn.discrete.plearn import ParamLearner
from pyspark_bbn.discrete.scblearn import Naive


def parse_pargs(args: List[str]) -> argparse.Namespace:
    """Parse command-line arguments.

    :param args: command-line arguments.
    :type args: List[str]
    :return: parsed arguments.
    :rtype: argparse.Namespace
    """
    parser = argparse.ArgumentParser()
    parser.add_argument("--input_data", type=str, required=True, help="Input CSV file")
    parser.add_argument("--clazz", type=str, required=False, help="Clazz variable")

    return parser.parse_args(args)


def start(input_data: str, clazz: str) -> None:
    """Start the naive learning demo.

    :param input_data: path to input CSV file.
    :type input_data: str
    :param clazz: class variable name.
    :type clazz: str
    :return: None
    :rtype: None
    """
    spark = SparkSession.builder.appName("learn-naive").getOrCreate()

    sdf = spark.read.option("header", True).option("inferSchema", True).csv(input_data)

    print("data schema")
    sdf.printSchema()

    print("")
    print("data sample")
    sdf.show(10)

    data = DiscreteData(sdf)
    naive = Naive(data, clazz)
    g = naive.get_network()

    print("")
    print("nodes")
    print("-" * 10)
    for n in g.nodes():
        print(f"{n}")

    print("")
    print("edges")
    print("-" * 10)
    for pa, ch in g.edges():
        print(f"{pa} -> {ch}")

    param_learner = ParamLearner(data, g)
    p = param_learner.get_params()

    print("")
    print("params")
    print("-" * 10)
    print(json.dumps(p, indent=2))

    print("")
    print("py-bbn, posteriors")
    print("-" * 10)
    bbn = get_bbn(g, p, data.get_profile())
    join_tree = InferenceController.apply(bbn)

    for node, posteriors in join_tree.get_posteriors().items():
        p_str = ", ".join([f"{val}={prob:.5f}" for val, prob in posteriors.items()])
        print(f"{node} : {p_str}")

    print("")
    print("py-bbn, data")
    print("-" * 10)
    pybbn_data = get_pybbn_data(g, p, data.get_profile())
    print(json.dumps(pybbn_data, indent=2))

    print("")
    print("darkstar, data")
    print("-" * 10)
    darkstar_data = get_darkstar_data(g, p, data.get_profile())
    print(json.dumps(darkstar_data, indent=2))


if __name__ == "__main__":
    args = parse_pargs(sys.argv[1:])

    input_data = args.input_data
    clazz = args.clazz
    start(input_data, clazz)