Domino Distributed Computing

from domino import Domino
domino = Domino("project-owner/project-name")
# Start workspace with Spark clusterworkspace = domino.workspace_start(    hardware_tier_name="medium",    cluster_config={        "clusterType": "Spark",        "workerCount": 4,        "workerHardwareTier": "medium",        "masterHardwareTier": "medium"    })

from pyspark.sql import SparkSession
# Domino auto-configures Sparkspark = SparkSession.builder.getOrCreate()
# Check configurationprint(f"Spark version: {spark.version}")print(f"Executors: {spark.sparkContext.defaultParallelism}")

# Read CSVdf = spark.read.csv("/mnt/data/dataset/data.csv", header=True, inferSchema=True)
# Read Parquetdf = spark.read.parquet("/mnt/data/dataset/")
# Read from databasedf = spark.read.jdbc(    url="jdbc:postgresql://host:5432/db",    table="schema.table",    properties={"user": "user", "password": "pass"})

from pyspark.sql import functions as F
# Transformationsresult = df.filter(F.col("value") > 100) \    .groupBy("category") \    .agg(F.mean("value").alias("avg_value")) \    .orderBy("avg_value", ascending=False)
# Show resultsresult.show()

from pyspark.ml.feature import VectorAssemblerfrom pyspark.ml.classification import RandomForestClassifierfrom pyspark.ml import Pipeline
# Prepare featuresassembler = VectorAssembler(    inputCols=["feature1", "feature2", "feature3"],    outputCol="features")
# Create modelrf = RandomForestClassifier(    featuresCol="features",    labelCol="label",    numTrees=100)
# Build pipelinepipeline = Pipeline(stages=[assembler, rf])model = pipeline.fit(train_df)predictions = model.transform(test_df)

# Write Parquet (recommended)result.write.parquet("/mnt/artifacts/output/", mode="overwrite")
# Write CSVresult.write.csv("/mnt/artifacts/output.csv", header=True)

import ray
# Domino auto-initializes Ray# Or manually connectray.init(address="auto")
print(f"Cluster resources: {ray.cluster_resources()}")

import ray
@ray.remotedef process_item(item):    # Your processing logic    return item * 2
# Run in parallelitems = [1, 2, 3, 4, 5]futures = [process_item.remote(item) for item in items]results = ray.get(futures)print(results)  # [2, 4, 6, 8, 10]

from ray import trainfrom ray.train import ScalingConfigfrom ray.train.torch import TorchTrainer
def train_func():    # Training logic    model = create_model()    for epoch in range(10):        train_epoch(model)        train.report({"loss": loss})
trainer = TorchTrainer(    train_func,    scaling_config=ScalingConfig(num_workers=4, use_gpu=True))result = trainer.fit()

from ray import tunefrom ray.tune import CLIReporter
def objective(config):    # Training with hyperparameters    model = train_model(        learning_rate=config["lr"],        batch_size=config["batch_size"]    )    return {"accuracy": accuracy}
analysis = tune.run(    objective,    config={        "lr": tune.loguniform(1e-4, 1e-1),        "batch_size": tune.choice([32, 64, 128])    },    num_samples=20,    progress_reporter=CLIReporter())
print(f"Best config: {analysis.best_config}")

from dask.distributed import Client
# Domino auto-configures Daskclient = Client()
print(f"Dashboard: {client.dashboard_link}")print(f"Workers: {len(client.scheduler_info()['workers'])}")

import dask.dataframe as dd
# Read large CSV filesdf = dd.read_csv("/mnt/data/dataset/*.csv")
# Parallel operations (lazy)result = df.groupby("category")["value"].mean()
# Executecomputed_result = result.compute()

import dask.array as da
# Create large arrayx = da.random.random((100000, 100000), chunks=(1000, 1000))
# Operations (lazy)result = x.mean()
# Computevalue = result.compute()

from dask_ml.model_selection import GridSearchCVfrom sklearn.ensemble import RandomForestClassifier
# Distributed hyperparameter searchparam_grid = {    "n_estimators": [100, 200, 300],    "max_depth": [10, 20, 30]}
grid_search = GridSearchCV(    RandomForestClassifier(),    param_grid,    cv=3)
grid_search.fit(X_train, y_train)print(f"Best params: {grid_search.best_params_}")

# Use GPU-accelerated Sparkspark = SparkSession.builder \    .config("spark.rapids.sql.enabled", "true") \    .getOrCreate()
# Operations automatically use GPUdf = spark.read.parquet("/mnt/data/large_dataset/")result = df.groupBy("category").agg({"value": "mean"})

@ray.remote(num_gpus=1)def train_on_gpu():    import torch    device = torch.device("cuda")    # GPU training logic    return model
# Run on GPU workersfutures = [train_on_gpu.remote() for _ in range(4)]

cluster_config = {    "clusterType": "Spark",    "workerCount": 2,    "maxWorkerCount": 10,  # Scale up to 10    "autoScaling": True}

# Keep data close to compute# Use Domino Datasets or cloud storage in same regiondf = spark.read.parquet("/mnt/data/dataset/")

# Cache frequently used DataFramesdf.cache()df.persist()