Update to Domino Data Lab

{
  "description": "Work with distributed computing frameworks in Domino including Apache Spark, Ray, and Dask clusters. Covers cluster configuration, on-demand clusters, choosing between frameworks, PySpark usage, and scaling workloads. Use when processing large datasets, parallel ML training, or running distributed compute jobs.",
  "included_files": [],
  "name": "domino-distributed-computing",
  "skill_md_contents": "---\nname: domino-distributed-computing\ndescription: Work with distributed computing frameworks in Domino including Apache Spark, Ray, and Dask clusters. Covers cluster configuration, on-demand clusters, choosing between frameworks, PySpark usage, and scaling workloads. Use when processing large datasets, parallel ML training, or running distributed compute jobs.\n---\n\n# Domino Distributed Computing Skill\n\n## Description\nThis skill helps users work with distributed computing frameworks in Domino - Spark, Ray, and Dask clusters for scaling compute-intensive workloads.\n\n## Activation\nActivate this skill when users want to:\n- Run Spark, Ray, or Dask clusters in Domino\n- Scale data processing or ML training\n- Configure distributed cluster settings\n- Understand when to use each framework\n\n## Supported Frameworks\n\n| Framework | Best For |\n|-----------|----------|\n| **Apache Spark** | Large-scale data processing, SQL, ETL |\n| **Ray** | Distributed ML, hyperparameter tuning, RL |\n| **Dask** | Parallel pandas, NumPy at scale |\n| **MPI** | Scientific computing, HPC workloads |\n\n## When to Use Each Framework\n\n### Spark\n- Processing terabyte-scale data\n- SQL analytics on big data\n- ETL pipelines\n- Structured data processing\n\n### Ray\n- Distributed model training\n- Hyperparameter optimization\n- Reinforcement learning\n- Generic Python parallelization\n\n### Dask\n- Scaling pandas workflows\n- Parallel NumPy operations\n- Lazy evaluation needed\n- Familiar pandas/NumPy API preferred\n\n## Launching On-Demand Clusters\n\n### Via Domino UI\n1. Start a workspace or job\n2. Check **Attach compute cluster**\n3. Select:\n   - **Cluster Type**: Spark, Ray, or Dask\n   - **Worker Count**: Number of workers\n   - **Hardware Tier**: Resources per worker\n   - **Auto-scaling**: Enable/disable\n4. Launch\n\n### Via Python SDK\n```python\nfrom domino import Domino\n\ndomino = Domino(\"project-owner/project-name\")\n\n# Start workspace with Spark cluster\nworkspace = domino.workspace_start(\n    hardware_tier_name=\"medium\",\n    cluster_config={\n        \"clusterType\": \"Spark\",\n        \"workerCount\": 4,\n        \"workerHardwareTier\": \"medium\",\n        \"masterHardwareTier\": \"medium\"\n    }\n)\n```\n\n## Apache Spark\n\n### Connecting to Spark\n```python\nfrom pyspark.sql import SparkSession\n\n# Domino auto-configures Spark\nspark = SparkSession.builder.getOrCreate()\n\n# Check configuration\nprint(f\"Spark version: {spark.version}\")\nprint(f\"Executors: {spark.sparkContext.defaultParallelism}\")\n```\n\n### Reading Data\n```python\n# Read CSV\ndf = spark.read.csv(\"/mnt/data/dataset/data.csv\", header=True, inferSchema=True)\n\n# Read Parquet\ndf = spark.read.parquet(\"/mnt/data/dataset/\")\n\n# Read from database\ndf = spark.read.jdbc(\n    url=\"jdbc:postgresql://host:5432/db\",\n    table=\"schema.table\",\n    properties={\"user\": \"user\", \"password\": \"pass\"}\n)\n```\n\n### Processing Data\n```python\nfrom pyspark.sql import functions as F\n\n# Transformations\nresult = df.filter(F.col(\"value\") > 100) \\\n    .groupBy(\"category\") \\\n    .agg(F.mean(\"value\").alias(\"avg_value\")) \\\n    .orderBy(\"avg_value\", ascending=False)\n\n# Show results\nresult.show()\n```\n\n### Machine Learning with Spark MLlib\n```python\nfrom pyspark.ml.feature import VectorAssembler\nfrom pyspark.ml.classification import RandomForestClassifier\nfrom pyspark.ml import Pipeline\n\n# Prepare features\nassembler = VectorAssembler(\n    inputCols=[\"feature1\", \"feature2\", \"feature3\"],\n    outputCol=\"features\"\n)\n\n# Create model\nrf = RandomForestClassifier(\n    featuresCol=\"features\",\n    labelCol=\"label\",\n    numTrees=100\n)\n\n# Build pipeline\npipeline = Pipeline(stages=[assembler, rf])\nmodel = pipeline.fit(train_df)\npredictions = model.transform(test_df)\n```\n\n### Writing Results\n```python\n# Write Parquet (recommended)\nresult.write.parquet(\"/mnt/artifacts/output/\", mode=\"overwrite\")\n\n# Write CSV\nresult.write.csv(\"/mnt/artifacts/output.csv\", header=True)\n```\n\n## Ray\n\n### Connecting to Ray\n```python\nimport ray\n\n# Domino auto-initializes Ray\n# Or manually connect\nray.init(address=\"auto\")\n\nprint(f\"Cluster resources: {ray.cluster_resources()}\")\n```\n\n### Parallel Tasks\n```python\nimport ray\n\n@ray.remote\ndef process_item(item):\n    # Your processing logic\n    return item * 2\n\n# Run in parallel\nitems = [1, 2, 3, 4, 5]\nfutures = [process_item.remote(item) for item in items]\nresults = ray.get(futures)\nprint(results)  # [2, 4, 6, 8, 10]\n```\n\n### Distributed Training with Ray Train\n```python\nfrom ray import train\nfrom ray.train import ScalingConfig\nfrom ray.train.torch import TorchTrainer\n\ndef train_func():\n    # Training logic\n    model = create_model()\n    for epoch in range(10):\n        train_epoch(model)\n        train.report({\"loss\": loss})\n\ntrainer = TorchTrainer(\n    train_func,\n    scaling_config=ScalingConfig(num_workers=4, use_gpu=True)\n)\nresult = trainer.fit()\n```\n\n### Hyperparameter Tuning with Ray Tune\n```python\nfrom ray import tune\nfrom ray.tune import CLIReporter\n\ndef objective(config):\n    # Training with hyperparameters\n    model = train_model(\n        learning_rate=config[\"lr\"],\n        batch_size=config[\"batch_size\"]\n    )\n    return {\"accuracy\": accuracy}\n\nanalysis = tune.run(\n    objective,\n    config={\n        \"lr\": tune.loguniform(1e-4, 1e-1),\n        \"batch_size\": tune.choice([32, 64, 128])\n    },\n    num_samples=20,\n    progress_reporter=CLIReporter()\n)\n\nprint(f\"Best config: {analysis.best_config}\")\n```\n\n## Dask\n\n### Connecting to Dask\n```python\nfrom dask.distributed import Client\n\n# Domino auto-configures Dask\nclient = Client()\n\nprint(f\"Dashboard: {client.dashboard_link}\")\nprint(f\"Workers: {len(client.scheduler_info()['workers'])}\")\n```\n\n### Dask DataFrames (Parallel pandas)\n```python\nimport dask.dataframe as dd\n\n# Read large CSV files\ndf = dd.read_csv(\"/mnt/data/dataset/*.csv\")\n\n# Parallel operations (lazy)\nresult = df.groupby(\"category\")[\"value\"].mean()\n\n# Execute\ncomputed_result = result.compute()\n```\n\n### Dask Arrays (Parallel NumPy)\n```python\nimport dask.array as da\n\n# Create large array\nx = da.random.random((100000, 100000), chunks=(1000, 1000))\n\n# Operations (lazy)\nresult = x.mean()\n\n# Compute\nvalue = result.compute()\n```\n\n### Dask ML\n```python\nfrom dask_ml.model_selection import GridSearchCV\nfrom sklearn.ensemble import RandomForestClassifier\n\n# Distributed hyperparameter search\nparam_grid = {\n    \"n_estimators\": [100, 200, 300],\n    \"max_depth\": [10, 20, 30]\n}\n\ngrid_search = GridSearchCV(\n    RandomForestClassifier(),\n    param_grid,\n    cv=3\n)\n\ngrid_search.fit(X_train, y_train)\nprint(f\"Best params: {grid_search.best_params_}\")\n```\n\n## GPU Clusters\n\n### Spark RAPIDS\n```python\n# Use GPU-accelerated Spark\nspark = SparkSession.builder \\\n    .config(\"spark.rapids.sql.enabled\", \"true\") \\\n    .getOrCreate()\n\n# Operations automatically use GPU\ndf = spark.read.parquet(\"/mnt/data/large_dataset/\")\nresult = df.groupBy(\"category\").agg({\"value\": \"mean\"})\n```\n\n### Ray with GPUs\n```python\n@ray.remote(num_gpus=1)\ndef train_on_gpu():\n    import torch\n    device = torch.device(\"cuda\")\n    # GPU training logic\n    return model\n\n# Run on GPU workers\nfutures = [train_on_gpu.remote() for _ in range(4)]\n```\n\n## Autoscaling\n\n### Enable Autoscaling\nConfigure clusters to scale based on workload:\n```python\ncluster_config = {\n    \"clusterType\": \"Spark\",\n    \"workerCount\": 2,\n    \"maxWorkerCount\": 10,  # Scale up to 10\n    \"autoScaling\": True\n}\n```\n\n### Monitor Scaling\nView cluster status in Domino UI or via dashboard URLs.\n\n## Best Practices\n\n### 1. Choose Right Framework\n- SQL/ETL: Spark\n- ML/Parallel Python: Ray\n- Pandas at scale: Dask\n\n### 2. Right-size Clusters\n- Start small, scale up\n- Monitor resource usage\n- Use autoscaling when unsure\n\n### 3. Data Locality\n```python\n# Keep data close to compute\n# Use Domino Datasets or cloud storage in same region\ndf = spark.read.parquet(\"/mnt/data/dataset/\")\n```\n\n### 4. Persist Intermediate Results\n```python\n# Cache frequently used DataFrames\ndf.cache()\ndf.persist()\n```\n\n## Troubleshooting\n\n### Cluster Won't Start\n- Check hardware tier availability\n- Verify cluster environment builds\n- Review cluster logs\n\n### Out of Memory\n- Increase worker memory\n- Add more workers\n- Optimize code (reduce shuffles)\n\n### Slow Performance\n- Check data locality\n- Review partition sizes\n- Monitor cluster dashboard\n\n## Documentation Reference\n- [On-demand distributed computing](https://docs.dominodatalab.com/en/latest/user_guide/8b4418/on-demand-distributed-computing/)\n- [Spark, Dask, Ray comparison](https://domino.ai/blog/spark-dask-ray-choosing-the-right-framework)\n- [Access data with Dask](https://docs.dominodatalab.com/en/latest/user_guide/0919a6/access-data-with-dask/)\n"
}