Following the blog post on Turn AML compute into Ray and Dask , we've added a new exciting capability to run Spark within AML compute where Spark shares the same context with your ML code. The Spark version is 3.2.1 with support for Delta Lake and Synapse SQL read/write. This enables users of AML to perform powerful data transformation and even Spark ML within AML interactive notebook or in a job run. 

Following the blog post on Turning AML compute into Ray and Dask , we've added a new exciting capability to run Spark within AML compute where Spark shares the same context with your ML code. The Spark version is 3.2.1 with support for Delta Lake and Synapse SQL read/write. This enables users of AML to perform powerful data transformation and even Spark ML within AML interactive notebook or in a job run. 

Traditionally, Azure ML integrates with Spark Synapse or external compute services via a pipeline step or better via magic command like %synapse, but the computing context is separate from your AML logic so you still need to run Spark in a separate step and persist the output to some storage and load it in your AML script.

With this approach, Spark is available right within your AML code whether it's AML notebook, python script or pipeline step. It shares the common computing context and most of the cases you can just directly convert the Spark Dataframe to Pandas and Dask Dataframe without persisting first to an intermediary storage.

Here are examples of how Spark inside AML can benefit users:

1. Easy Spark setting up with credentials to access Azure Storage

from ray_on_aml.core import Ray_On_AML\nray_on_aml =Ray_On_AML(ws=ws, compute_cluster =\"d12-v2-ssh\", maxnode=5)\nray = ray_on_aml.getRay()\nstorage_account_name =\"adlsdatalakegen6\"\nstorage_account_key=ws.get_default_keyvault().get_secret(\"adlsdatalakegen6\")\nadditional_spark_configs ={f\"fs.azure.account.key.{storage_account_name}.dfs.core.windows.net\":f\"{storage_account_key}\"}\nspark = ray_on_aml.getSpark(executor_cores =3,num_executors =3 ,executor_memory='10GB', additional_spark_configs=additional_spark_configs)\n#Number of nodes (including head node) can be set as number of executor. 

2. Fast loading high volume data from delta lake in ADLS or Synapse SQL Pool

#Access delta lake data\nadls_data = spark.read.format(\"delta\").load(\"abfss://mltraining@adlsdatalakegen6.dfs.core.windows.net/ISDWeatherDelta\")

server_name = \"jdbc:sqlserver://sy2qwhqqkv7eacsws1.sql.azuresynapse.net:1433\"\ndatabase_name = \"sy2qwhqqkv7eacsws1p1\"\nurl = server_name + \";\" + \"databaseName=\" + database_name + \";\"\ntable_name = \"ISDWeatherDelta\"\nusername = \"azureuser\"\npassword = \"\" \njdbcDF = spark.read \\\n        .format(\"jdbc\") \\\n        .option(\"url\", url) \\\n        .option(\"dbtable\", table_name) \\\n        .option(\"driver\", \"com.microsoft.sqlserver.jdbc.SQLServerDriver\") \\\n        .option(\"user\", username) \\\n        .option(\"password\", password).load()\n\njdbcDF.show()

3. Using Koalas (pyspark Pandas) API for EDA and data transformation on big dataset.
Data scientists loves pandas but not everyone is comfortable with Pyspark so Koalas (Pandas on pyspark) is the answer. You can easily load dataset that is way bigger than the memory of a machine and use Pandas like APIs to manipulate and visualize big dataset.

#Reading 73M rows of data into a Pandas like dataframe\nfrom pyspark.pandas import read_delta\nkoalas_df= read_delta(\"abfss://mltraining@adlsdatalakegen6.dfs.core.windows.net/ISDWeatherDelta\")\nkoalas_df.describe()

and plotting with big dataset with plotly backend

koalas_df[\"snowDepth\"].plot.area()

4. Loading big dataset, transform with Spark then train XGBoost within a single ML step.
(credit: The below example is taken from raydp project)

data = spark.read.format(\"csv\").option(\"header\", \"true\") \\\n        .option(\"inferSchema\", \"true\") \\\n        .load(NYC_TRAIN_CSV)\n# Set spark timezone for processing datetime\nspark.conf.set(\"spark.sql.session.timeZone\", \"UTC\")\n# Transform the dataset\ndata = nyc_taxi_preprocess(data)\n# Split data into train_dataset and test_dataset\ntrain_df, test_df = random_split(data, [0.9, 0.1], 0)\n# Convert spark dataframe into ray dataset\ntrain_dataset = ray.data.from_spark(train_df)\ntest_dataset = ray.data.from_spark(test_df)\n# Then convert them into DMatrix used by xgboost\ndtrain = RayDMatrix(train_dataset, label='fare_amount')\ndtest = RayDMatrix(test_dataset, label='fare_amount')\n# Configure the XGBoost model\nconfig = {\n    \"tree_method\": \"hist\",\n    \"eval_metric\": [\"logloss\", \"error\"],\n}\nevals_result = {}\n# Train the model\nbst = train(\n        config,\n        dtrain,\n        evals=[(dtest, \"eval\")],\n        evals_result=evals_result,\n        ray_params=RayParams(max_actor_restarts=2, num_actors=2, cpus_per_actor=2),\n        num_boost_round=10)

5. Spark mllib

(Collaborative filtering example from https://spark.apache.org/docs/3.2.1/ml-collaborative-filtering.html)

from pyspark.ml.evaluation import RegressionEvaluator\nfrom pyspark.ml.recommendation import ALS\nfrom pyspark.sql import Row\n\nlines = spark.read.text(\"abfss://mltraining@adlsdatalakegen6.dfs.core.windows.net/mllib/sample_movielens_ratings.txt\").rdd\nparts = lines.map(lambda row: row.value.split(\"::\"))\nratingsRDD = parts.map(lambda p: Row(userId=int(p[0]), movieId=int(p[1]),\n                                     rating=float(p[2]), timestamp=int(p[3])))\nratings = spark.createDataFrame(ratingsRDD)\n(training, test) = ratings.randomSplit([0.8, 0.2])\n\n# Build the recommendation model using ALS on the training data\n# Note we set cold start strategy to 'drop' to ensure we don't get NaN evaluation metrics\nals = ALS(maxIter=5, regParam=0.01, userCol=\"userId\", itemCol=\"movieId\", ratingCol=\"rating\",\n          coldStartStrategy=\"drop\")\nmodel = als.fit(training)\n\n# Evaluate the model by computing the RMSE on the test data\npredictions = model.transform(test)\nevaluator = RegressionEvaluator(metricName=\"rmse\", labelCol=\"rating\",\n                                predictionCol=\"prediction\")\nrmse = evaluator.evaluate(predictions)\nprint(\"Root-mean-square error = \" + str(rmse))\n\n# Generate top 10 movie recommendations for each user\nuserRecs = model.recommendForAllUsers(10)\n# Generate top 10 user recommendations for each movie\nmovieRecs = model.recommendForAllItems(10)\n\n# Generate top 10 movie recommendations for a specified set of users\nusers = ratings.select(als.getUserCol()).distinct().limit(3)\nuserSubsetRecs = model.recommendForUserSubset(users, 10)\n# Generate top 10 user recommendations for a specified set of movies\nmovies = ratings.select(als.getItemCol()).distinct().limit(3)\nmovieSubSetRecs = model.recommendForItemSubset(movies, 10)

6. Use in a remote AML job (CLI v2 is supported)
You can easily turn your remote AML cluster into Spark in the same way you would do with ray in ray-on-aml.
In your python script, do something like this

if __name__ == \"__main__\":\n    run = Run.get_context()\n    ws = run.experiment.workspace\n    ray_on_aml =Ray_On_AML()\n    ray = ray_on_aml.getRay()\n    if ray: #in the headnode\n        spark = ray_on_aml.getSpark(executor_cores =3,num_executors =3 ,executor_memory='10GB')

This capability is based on the work from raydp which develops a framework to run Spark on ray

For installation, this capability is part of ray-on-aml library, so you just need to upgrade the library to the latest version to use the Spark capability.

Checkout Spark example notebook from ray-on-aml repo to learn more!

Following the blog post on Turning AML compute into Ray and Dask , we've added a new exciting capability to run Spark within AML compute where Spark shares the same context with your ML code. The Spark version is 3.2.1 with support for Delta Lake and Synapse SQL read/write. This enables users of AML to perform powerful data transformation and even Spark ML within AML interactive notebook or in a job run. 

Traditionally, Azure ML integrates with Spark Synapse or external compute services via a pipeline step or better via magic command like %synapse, but the computing context is separate from your AML logic so you still need to run Spark in a separate step and persist the output to some storage and load it in your AML script.

With this approach, Spark is available right within your AML code whether it's AML notebook, python script or pipeline step. It shares the common computing context and most of the cases you can just directly convert the Spark Dataframe to Pandas and Dask Dataframe without persisting first to an intermediary storage.

Here are examples of how Spark inside AML can benefit users:

1. Easy Spark setting up with credentials to access Azure Storage

2. Fast loading high volume data from delta lake in ADLS or Synapse SQL Pool

3. Using Koalas (pyspark Pandas) API for EDA and data transformation on big dataset.
Data scientists loves pandas but not everyone is comfortable with Pyspark so Koalas (Pandas on pyspark) is the answer. You can easily load dataset that is way bigger than the memory of a machine and use Pandas like APIs to manipulate and visualize big dataset.

and plotting with big dataset with plotly backend

4. Loading big dataset, transform with Spark then train XGBoost within a single ML step.
(credit: The below example is taken from raydp project)

5. Spark mllib

(Collaborative filtering example from https://spark.apache.org/docs/3.2.1/ml-collaborative-filtering.html)

6. Use in a remote AML job (CLI v2 is supported)
You can easily turn your remote AML cluster into Spark in the same way you would do with ray in ray-on-aml.
In your python script, do something like this

This capability is based on the work from raydp which develops a framework to run Spark on ray

For installation, this capability is part of ray-on-aml library, so you just need to upgrade the library to the latest version to use the Spark capability.

Checkout Spark example notebook from ray-on-aml repo to learn more!

Following the blog post on Turn AML compute into Ray and Dask , we've added a new exciting capability to run Spark within AML compute where Spark shares the same context with your ML code. The Spark version is 3.2.1 with support for Delta Lake and Synapse SQL read/write. This enables users of AML to perform powerful data transformation and even Spark ML within AML interactive notebook or in a job run.