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.

from typing import List
from pyspark.sql import SparkSession
import argparse
import sys
import json

from pybbn.pptc.inferencecontroller import InferenceController
from pysparkbbn.discrete.bbn import get_bbn, get_darkstar_data, get_pybbn_data
from pysparkbbn.discrete.data import DiscreteData
from pysparkbbn.discrete.scblearn import Naive
from pysparkbbn.discrete.plearn import ParamLearner


def parse_pargs(args: List[str]) -> 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):
    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.

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 pysparkbbn.discrete.bbn import get_bbn
from pysparkbbn.discrete.data import DiscreteData
from pysparkbbn.discrete.plearn import ParamLearner
from pysparkbbn.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:
    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):
    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 pysparkbbn.discrete.bbn import get_bbn, get_darkstar_data, get_pybbn_data
from pysparkbbn.discrete.data import DiscreteData
from pysparkbbn.discrete.plearn import ParamLearner
from pysparkbbn.discrete.scblearn import Naive


def parse_pargs(args: List[str]) -> 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):
    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)