Apache Spark

https://spark.apache.org/docs/3.5.6
Admin
Apache Spark is a unified analytics engine for large-scale data processing. It offers high-level...
Tokens:12,568,006
Snippets:119,139
Trust Score:10
Update:4 hours ago
Show doc for...
Tokens:
Raw
### Start SparkR session with existing Spark installation using sparkHome

Source: https://spark.apache.org/docs/3.5.6/api/R/articles/sparkr-vignettes.html

Pass the sparkHome argument to sparkR.session to point to an existing Spark installation, avoiding reinstallation.

```R
sparkR.session(sparkHome = "/HOME/spark")
```

--------------------------------

### Create SparkSession in Java using SparkSession.builder()

Source: https://spark.apache.org/docs/3.5.6/sql-getting-started.html

Initializes a SparkSession in Java with an application name and a custom configuration option. Use this as the entry point for Spark SQL functionality in Java.

```java
import org.apache.spark.sql.SparkSession;

SparkSession spark = SparkSession
  .builder()
  .appName("Java Spark SQL basic example")
  .config("spark.some.config.option", "some-value")
  .getOrCreate();
```

--------------------------------

### Java: Create Datasets with beans, primitives, and JSON

Source: https://spark.apache.org/docs/3.5.6/sql-getting-started.html

Shows how to create Datasets in Java using a Bean class, primitive types, and reading JSON into a Dataset. Encoders are created with Encoders.bean() and Encoders.LONG(). The example includes the output of show() and the result of collect().

```java
import java.util.Arrays;
import java.util.Collections;
import java.io.Serializable;

import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.Encoder;
import org.apache.spark.sql.Encoders;

public static class Person implements Serializable {
  private String name;
  private long age;

  public String getName() {
    return name;
  }

  public void setName(String name) {
    this.name = name;
  }

  public long getAge() {
    return age;
  }

  public void setAge(long age) {
    this.age = age;
  }
}

// Create an instance of a Bean class
Person person = new Person();
person.setName("Andy");
person.setAge(32);

// Encoders are created for Java beans
Encoder<Person> personEncoder = Encoders.bean(Person.class);
Dataset<Person> javaBeanDS = spark.createDataset(
  Collections.singletonList(person),
  personEncoder
);
javaBeanDS.show();
// +---+----+
// |age|name|
// +---+----+
// | 32|Andy|
// +---+----+

// Encoders for most common types are provided in class Encoders
Encoder<Long> longEncoder = Encoders.LONG();
Dataset<Long> primitiveDS = spark.createDataset(Arrays.asList(1L, 2L, 3L), longEncoder);
Dataset<Long> transformedDS = primitiveDS.map(
    (MapFunction<Long, Long>) value -> value + 1L,
    longEncoder);
transformedDS.collect(); // Returns [2, 3, 4]

// DataFrames can be converted to a Dataset by providing a class. Mapping based on name
String path = "examples/src/main/resources/people.json";
Dataset<Person> peopleDS = spark.read().json(path).as(personEncoder);
peopleDS.show();
// +----+-------+
// | age|   name|
// +----+-------+
// |null|Michael|
// |  30|   Andy|
// |  19| Justin|
// +----+-------+
```

--------------------------------

### mapReduceTriplets usage example

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Example of using the legacy mapReduceTriplets operator to send a message to each destination vertex. The reduce function concatenates messages with spaces. This code is rewritten using aggregateMessages in the following example.

```scala
val graph: Graph[Int, Float] = ...
def msgFun(triplet: Triplet[Int, Float]): Iterator[(Int, String)] = {
  Iterator((triplet.dstId, "Hi"))
}
def reduceFun(a: String, b: String): String = a + " " + b
val result = graph.mapReduceTriplets[String](msgFun, reduceFun)
```

--------------------------------

### Run Spark examples with run-example and spark-submit

Source: https://spark.apache.org/docs/3.5.6/quick-start.html

Commands to run built-in Spark examples. Use ./bin/run-example for Scala/Java (e.g., SparkPi) and ./bin/spark-submit for Python (examples/src/main/python/pi.py) and R (examples/src/main/r/dataframe.R) examples.

```bash
# For Scala and Java, use run-example:
./bin/run-example SparkPi

# For Python examples, use spark-submit directly:
./bin/spark-submit examples/src/main/python/pi.py

# For R examples, use spark-submit directly:
./bin/spark-submit examples/src/main/r/dataframe.R
```

--------------------------------

### Create SparkSession in Scala using SparkSession.builder()

Source: https://spark.apache.org/docs/3.5.6/sql-getting-started.html

Initializes a SparkSession in Scala with an application name and a custom configuration option. Use this as the entry point for Spark SQL functionality in Scala.

```scala
import org.apache.spark.sql.SparkSession

val spark = SparkSession
  .builder()
  .appName("Spark SQL basic example")
  .config("spark.some.config.option", "some-value")
  .getOrCreate()
```

--------------------------------

### Launch Spark server with Spark Connect using start-connect-server.sh

Source: https://spark.apache.org/docs/3.5.6/api/python/getting_started/quickstart_connect.html

Runs the start-connect-server.sh script to launch a Spark server with Spark Connect support. Requires environment variables loaded from ~/.profile and uses the Spark version from $SPARK_VERSION.

```bash
%%bash
source ~/.profile # Make sure environment variables are loaded.
$HOME/sbin/start-connect-server.sh --packages org.apache.spark:spark-connect_2.12:$SPARK_VERSION
```

--------------------------------

### Import Spark and GraphX

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Required imports to use Spark and GraphX in a project. Also imports RDD for examples that need it. If not using the Spark shell, a SparkContext is also needed.

```scala
import org.apache.spark._
import org.apache.spark.graphx._
// To make some of the examples work we will also need RDD
import org.apache.spark.rdd.RDD
```

--------------------------------

### SHOW PARTITIONS examples

Source: https://spark.apache.org/docs/3.5.6/sql-ref-syntax-aux-show-partitions.html

Examples showing how to create a partitioned table, insert rows, and list partitions using SHOW PARTITIONS with various partition specs. The output for each SHOW PARTITIONS command is displayed.

```sql
-- create a partitioned table and insert a few rows.
USE salesdb;
CREATE TABLE customer(id INT, name STRING) PARTITIONED BY (state STRING, city STRING);
INSERT INTO customer PARTITION (state = 'CA', city = 'Fremont') VALUES (100, 'John');
INSERT INTO customer PARTITION (state = 'CA', city = 'San Jose') VALUES (200, 'Marry');
INSERT INTO customer PARTITION (state = 'AZ', city = 'Peoria') VALUES (300, 'Daniel');
```

```sql
-- Lists all partitions for table `customer`
SHOW PARTITIONS customer;
+----------------------+
|             partition|
+----------------------+
|  state=AZ/city=Peoria|
| state=CA/city=Fremont|
|state=CA/city=San Jose|
+----------------------+
```

```sql
-- Lists all partitions for the qualified table `customer`
SHOW PARTITIONS salesdb.customer;
+----------------------+
|             partition|
+----------------------+
|  state=AZ/city=Peoria|
| state=CA/city=Fremont|
|state=CA/city=San Jose|
+----------------------+
```

```sql
-- Specify a full partition spec to list specific partition
SHOW PARTITIONS customer PARTITION (state = 'CA', city = 'Fremont');
+---------------------+
|            partition|
+---------------------+
|state=CA/city=Fremont|
+---------------------+
```

```sql
-- Specify a partial partition spec to list the specific partitions
SHOW PARTITIONS customer PARTITION (state = 'CA');
+----------------------+
|             partition|
+----------------------+
| state=CA/city=Fremont|
|state=CA/city=San Jose|
+----------------------+
```

```sql
-- Specify a partial spec to list specific partition
SHOW PARTITIONS customer PARTITION (city =  'San Jose');
+----------------------+
|             partition|
+----------------------+
|state=CA/city=San Jose|
+----------------------+
```

--------------------------------

### Start a streaming query for awaitTermination example

Source: https://spark.apache.org/docs/3.5.6/api/python/reference/pyspark.ss/api/pyspark.sql.streaming.StreamingQuery.awaitTermination.html

Creates a rate-based streaming DataFrame and starts a memory sink query named 'query_awaitTermination'. This is the setup for the awaitTermination examples that follow.

```python
>>> sdf = spark.readStream.format("rate").load()
>>> sq = sdf.writeStream.format('memory').queryName('query_awaitTermination').start()
```

--------------------------------

### Run Java/Scala example with bin/run-example

Source: https://spark.apache.org/docs/3.5.6/rdd-programming-guide.html

Use this command to run a Java or Scala example by passing the class name to Spark's bin/run-example script. The example shown runs SparkPi.

```bash
./bin/run-example SparkPi
```

--------------------------------

### aggregateMessages equivalent of mapReduceTriplets

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Rewrites the previous mapReduceTriplets example using the aggregateMessages operator. Uses EdgeContext to send messages directly to the destination vertex, avoiding iterator overhead and bytecode inspection.

```scala
val graph: Graph[Int, Float] = ...
def msgFun(triplet: EdgeContext[Int, Float, String]) {
  triplet.sendToDst("Hi")
}
def reduceFun(a: String, b: String): String = a + " " + b
val result = graph.aggregateMessages[String](msgFun, reduceFun)
```

--------------------------------

### startswith function examples

Source: https://spark.apache.org/docs/3.5.6/api/python/_modules/pyspark/sql/functions.html

Examples showing usage of startswith with string and binary columns, including expected output.

```python
>>> df = spark.createDataFrame([("Spark SQL", "Spark",)], ["a", "b"])
>>> df.select(startswith(df.a, df.b).alias('r')).collect()
[Row(r=True)]
```

```python
>>> df = spark.createDataFrame([("414243", "4142",)], ["e", "f"])
>>> df = df.select(to_binary("e").alias("e"), to_binary("f").alias("f"))
>>> df.printSchema()
root
 |-- e: binary (nullable = true)
 |-- f: binary (nullable = true)
>>> df.select(startswith("e", "f"), startswith("f", "e")).show()
+----------------+----------------+
|startswith(e, f)|startswith(f, e)|
+----------------+----------------+
|            true|           false|
+----------------+----------------+
```

--------------------------------

### Compute connected components on social network dataset

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Loads the same graph as the PageRank example, computes connected components with graph.connectedComponents(), joins component IDs with usernames, and prints the result. Uses GraphLoader.edgeListFile and Graph.connectedComponents.

```scala
import org.apache.spark.graphx.GraphLoader

// Load the graph as in the PageRank example
val graph = GraphLoader.edgeListFile(sc, "data/graphx/followers.txt")
// Find the connected components
val cc = graph.connectedComponents().vertices
// Join the connected components with the usernames
val users = sc.textFile("data/graphx/users.txt").map { line =>
  val fields = line.split(",")
  (fields(0).toLong, fields(1))
}
val ccByUsername = users.join(cc).map {
  case (id, (username, cc)) => (username, cc)
}
// Print the result
println(ccByUsername.collect().mkString("\n"))
```

--------------------------------

### Create person table and insert data for FIRST example

Source: https://spark.apache.org/docs/3.5.6/sql-ref-syntax-qry-select-groupby.html

Sets up the person table used in the FIRST(age) example. Includes a row with NULL age.

```sql
--Prepare data for ignore nulls example
CREATE TABLE person (id INT, name STRING, age INT);
INSERT INTO person VALUES
    (100, 'Mary', NULL),
    (200, 'John', 30),
    (300, 'Mike', 80),
    (400, 'Dan', 50);
```

--------------------------------

### Graph.cache

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Caches the graph in memory.

```APIDOC
## [SCALA] Graph.cache

### Description
Caches the graph in memory.

### Method
Function (def)

### Endpoint
Graph.cache(): Graph[VD, ED]

### Parameters
None

### Returns
- **Graph[VD, ED]** - The cached graph.
```

--------------------------------

### startApplication(SparkAppHandle.Listener... listeners)

Source: https://spark.apache.org/docs/3.5.6/api/java/org/apache/spark/launcher/AbstractLauncher.html

Starts a Spark application. Returns a SparkAppHandle.

```APIDOC
## startApplication(SparkAppHandle.Listener... listeners)

### Description
Starts a Spark application.

### Method
startApplication

### Parameters
- `listeners` (SparkAppHandle.Listener...) - Required - Listeners for the application.

### Returns
SparkAppHandle
```

--------------------------------

### sparkR.conf examples (dontrun)

Source: https://spark.apache.org/docs/3.5.6/api/R/reference/sparkR.conf.html

Example usage wrapped in if (FALSE) { # \dontrun{ ... } # } to prevent execution during package checks. Starts a SparkSession, retrieves all configs, gets the spark.master value, and gets spark.executor.memory with a default of "0g".

```r
if (FALSE) { # \dontrun{
sparkR.session()
allConfigs <- sparkR.conf()
masterValue <- unlist(sparkR.conf("spark.master"))
namedConfig <- sparkR.conf("spark.executor.memory", "0g")
} # }
```

--------------------------------

### Binarizer usage example with threshold and params

Source: https://spark.apache.org/docs/3.5.6/api/python/_modules/pyspark/ml/feature.html

Shows how to create a Binarizer, set threshold and input/output columns, transform a DataFrame, and save/load the model. The example also demonstrates using setParams and passing a params dictionary to transform.

```python
>>> df = spark.createDataFrame([(0.5,)], ["values"])
>>> binarizer = Binarizer(threshold=1.0, inputCol="values", outputCol="features")
>>> binarizer.setThreshold(1.0)
Binarizer...
>>> binarizer.setInputCol("values")
Binarizer...
>>> binarizer.setOutputCol("features")
Binarizer...
>>> binarizer.transform(df).head().features
0.0
>>> binarizer.setParams(outputCol="freqs").transform(df).head().freqs
0.0
>>> params = {binarizer.threshold: -0.5, binarizer.outputCol: "vector"}
>>> binarizer.transform(df, params).head().vector
1.0
>>> binarizerPath = temp_path + "/binarizer"
>>> binarizer.save(binarizerPath)
>>> loadedBinarizer = Binarizer.load(binarizerPath)
>>> loadedBinarizer.getThreshold() == binarizer.getThreshold()
True
>>> loadedBinarizer.transform(df).take(1) == binarizer.transform(df).take(1)
True
```

--------------------------------

### Graph.connectedComponents

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Computes the connected components of the graph.

```APIDOC
## [SCALA] Graph.connectedComponents

### Description
Computes the connected components of the graph.

### Method
Function (def)

### Endpoint
Graph.connectedComponents(): Graph[VertexId, ED]

### Parameters
None

### Returns
- **Graph[VertexId, ED]** - A graph where each vertex attribute is the ID of its connected component.
```

--------------------------------

### start()

Source: https://spark.apache.org/docs/3.5.6/api/java/org/apache/spark/scheduler/SchedulerBackend.html

Starts the scheduler backend. This method is called to initialize and begin the backend's operation.

```APIDOC
## [METHOD] start()

### Description
Starts the scheduler backend.

### Method
start

### Signature
void start()

### Parameters
None

### Returns
void
```

--------------------------------

### Graph.pageRank

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Computes the PageRank of each vertex in the graph.

```APIDOC
## [SCALA] Graph.pageRank

### Description
Computes the PageRank of each vertex in the graph.

### Method
Function (def)

### Endpoint
Graph.pageRank(tol: Double, resetProb: Double = 0.15): Graph[Double, Double]

### Parameters
- **tol** (Double) - Required - The tolerance for convergence.
- **resetProb** (Double) - Optional - The random reset probability. Defaults to 0.15.

### Returns
- **Graph[Double, Double]** - A graph where each vertex attribute is its PageRank value and each edge attribute is its weight.
```

--------------------------------

### Setup: employee and department tables

Source: https://spark.apache.org/docs/3.5.6/sql-ref-syntax-qry-select-join.html

Creates the employee and department tables used in all join examples. The employee table has columns id, name, and deptno; the department table has deptno and deptname.

```sql
SELECT * FROM employee;
+---+-----+------+
| id| name|deptno|
+---+-----+------+
|105|Chloe|     5|
|103| Paul|     3|
|101| John|     1|
|102| Lisa|     2|
|104| Evan|     4|
|106|  Amy|     6|
+---+-----+------+
```

```sql
SELECT * FROM department;
+------+-----------+
|deptno|   deptname|
+------+-----------+
|     3|Engineering|
|     2|      Sales|
|     1|  Marketing|
+------+-----------+
```

--------------------------------

### Graph.reverse

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Reverses the direction of all edges in the graph.

```APIDOC
## [SCALA] Graph.reverse

### Description
Reverses the direction of all edges in the graph.

### Method
Function (def)

### Endpoint
Graph.reverse: Graph[VD, ED]

### Parameters
None

### Returns
- **Graph[VD, ED]** - A new graph with all edges reversed.
```

--------------------------------

### SHOW DATABASES Examples

Source: https://spark.apache.org/docs/3.5.6/sql-ref-syntax-aux-show-databases.html

Examples demonstrating creating databases and listing them with SHOW DATABASES, SHOW DATABASES LIKE 'pay*', and SHOW SCHEMAS. The output tables show the databaseName column with matching databases.

```sql
-- Create database. Assumes a database named `default` already exists in
-- the system. 
CREATE DATABASE payroll_db;
CREATE DATABASE payments_db;

-- Lists all the databases. 
SHOW DATABASES;
+------------+
|databaseName|
+------------+
|     default|
| payments_db|
|  payroll_db|
+------------+
  
-- Lists databases with name starting with string pattern `pay`
SHOW DATABASES LIKE 'pay*';
+------------+
|databaseName|
+------------+
| payments_db|
|  payroll_db|
+------------+
  
-- Lists all databases. Keywords SCHEMAS and DATABASES are interchangeable. 
SHOW SCHEMAS;
+------------+
|databaseName|
+------------+
|     default|
| payments_db|
|  payroll_db|
+------------+
```

--------------------------------

### Graph.edges

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Returns the edges of the graph as an EdgeRDD.

```APIDOC
## [SCALA] Graph.edges

### Description
Returns the edges of the graph as an EdgeRDD.

### Method
Property accessor (val)

### Endpoint
Graph.edges

### Parameters
None

### Returns
- **EdgeRDD[ED]** - An EdgeRDD containing the edges and their attributes.
```

--------------------------------

### Graph.vertices

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Returns the vertices of the graph as a VertexRDD.

```APIDOC
## [SCALA] Graph.vertices

### Description
Returns the vertices of the graph as a VertexRDD.

### Method
Property accessor (val)

### Endpoint
Graph.vertices

### Parameters
None

### Returns
- **VertexRDD[VD]** - A VertexRDD containing the vertex IDs and their attributes.
```

--------------------------------

### Run Scala streaming example with run-example

Source: https://spark.apache.org/docs/3.5.6/streaming-programming-guide.html

Runs the Scala NetworkWordCount example using the run-example script, connecting to localhost:9999. Run in a different terminal after starting Netcat.

```bash
$ ./bin/run-example streaming.NetworkWordCount localhost 9999
```

--------------------------------

### Graph.numVertices

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Returns the number of vertices in the graph.

```APIDOC
## [SCALA] Graph.numVertices

### Description
Returns the number of vertices in the graph.

### Method
Property accessor (val)

### Endpoint
Graph.numVertices

### Parameters
None

### Returns
- **Long** - The number of vertices in the graph.
```

--------------------------------

### Run Scala/Java sample program SparkPi

Source: https://spark.apache.org/docs/3.5.6/

Uses bin/run-example to launch a Scala or Java sample program. Behind the scenes, this invokes the more general spark-submit script.

```bash
./bin/run-example SparkPi 10
```

--------------------------------

### sbt directory layout, package, and spark-submit commands

Source: https://spark.apache.org/docs/3.5.6/quick-start.html

Shows the required sbt project directory layout, the sbt package command to create a JAR, and the spark-submit command to run the application with --class SimpleApp and --master local[4]. The expected output includes the packaged JAR path and the application's line counts.

```bash
# Your directory layout should look like this
$ find .
.
./build.sbt
./src
./src/main
./src/main/scala
./src/main/scala/SimpleApp.scala

# Package a jar containing your application
$ sbt package
...
[info] Packaging {..}/{..}/target/scala-2.12/simple-project_2.12-1.0.jar

# Use spark-submit to run your application
$ YOUR_SPARK_HOME/bin/spark-submit \
  --class "SimpleApp" \
  --master local[4] \
  target/scala-2.12/simple-project_2.12-1.0.jar
...
Lines with a: 46, Lines with b: 23

```

--------------------------------

### Graph.numEdges

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Returns the number of edges in the graph.

```APIDOC
## [SCALA] Graph.numEdges

### Description
Returns the number of edges in the graph.

### Method
Property accessor (val)

### Endpoint
Graph.numEdges

### Parameters
None

### Returns
- **Long** - The number of edges in the graph.
```

--------------------------------

### BisectingKMeans usage example with fit, predict, and save/load

Source: https://spark.apache.org/docs/3.5.6/api/python/_modules/pyspark/ml/clustering.html

Demonstrates creating a BisectingKMeans model, setting parameters, fitting, predicting, computing cost, accessing summary, transforming data, and saving/loading the model and estimator. Shows default maxIter is 20 after clear, and that loaded models have no summary.

```python
>>> from pyspark.ml.linalg import Vectors
>>> data = [(Vectors.dense([0.0, 0.0]), 2.0), (Vectors.dense([1.0, 1.0]), 2.0),
...         (Vectors.dense([9.0, 8.0]), 2.0), (Vectors.dense([8.0, 9.0]), 2.0)]
>>> df = spark.createDataFrame(data, ["features", "weighCol"])
>>> bkm = BisectingKMeans(k=2, minDivisibleClusterSize=1.0)
>>> bkm.setMaxIter(10)
BisectingKMeans...
>>> bkm.getMaxIter()
10
>>> bkm.clear(bkm.maxIter)
>>> bkm.setSeed(1)
BisectingKMeans...
>>> bkm.setWeightCol("weighCol")
BisectingKMeans...
>>> bkm.getSeed()
1
>>> bkm.clear(bkm.seed)
>>> model = bkm.fit(df)
>>> model.getMaxIter()
20
>>> model.setPredictionCol("newPrediction")
BisectingKMeansModel...
>>> model.predict(df.head().features)
0
>>> centers = model.clusterCenters()
>>> len(centers)
2
>>> model.computeCost(df)
2.0
>>> model.hasSummary
True
>>> summary = model.summary
>>> summary.k
2
>>> summary.clusterSizes
[2, 2]
>>> summary.trainingCost
4.000...
>>> transformed = model.transform(df).select("features", "newPrediction")
>>> rows = transformed.collect()
>>> rows[0].newPrediction == rows[1].newPrediction
True
>>> rows[2].newPrediction == rows[3].newPrediction
True
>>> bkm_path = temp_path + "/bkm"
>>> bkm.save(bkm_path)
>>> bkm2 = BisectingKMeans.load(bkm_path)
>>> bkm2.getK()
2
>>> bkm2.getDistanceMeasure()
'euclidean'
>>> model_path = temp_path + "/bkm_model"
>>> model.save(model_path)
>>> model2 = BisectingKMeansModel.load(model_path)
>>> model2.hasSummary
False
>>> model.clusterCenters()[0] == model2.clusterCenters()[0]
array([ True,  True], dtype=bool)
>>> model.clusterCenters()[1] == model2.clusterCenters()[1]
array([ True,  True], dtype=bool)
>>> model.transform(df).take(1) == model2.transform(df).take(1)
True
```

--------------------------------

### Example using aggregateUsingIndex and innerJoin on VertexRDD

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Shows how to efficiently construct a new VertexRDD from an RDD[(VertexId, A)] using aggregateUsingIndex when a super-set VertexRDD exists. The example creates setA with 100 vertices, rddB with 200 entries, aggregates rddB into setB with 100 entries, then performs a fast innerJoin between setA and setB.

```scala
val setA: VertexRDD[Int] = VertexRDD(sc.parallelize(0L until 100L).map(id => (id, 1)))
val rddB: RDD[(VertexId, Double)] = sc.parallelize(0L until 100L).flatMap(id => List((id, 1.0), (id, 2.0)))
// There should be 200 entries in rddB
rddB.count
val setB: VertexRDD[Double] = setA.aggregateUsingIndex(rddB, _ + _)
// There should be 100 entries in setB
setB.count
// Joining A and B should now be fast!
val setC: VertexRDD[Double] = setA.innerJoin(setB)((id, a, b) => a + b)
```

--------------------------------

### position function examples

Source: https://spark.apache.org/docs/3.5.6/api/python/_modules/pyspark/sql/functions.html

Examples showing usage of position function with and without start parameter, including expected output.

```python
>>> import pyspark.sql.functions as sf
>>> spark.createDataFrame(
...     [("bar", "foobarbar", 5,)], ["a", "b", "c"]
... ).select(sf.position("a", "b", "c")).show()
+-----------------+
|position(a, b, c)|
+-----------------+
|                7|
+-----------------+
```

```python
>>> spark.createDataFrame(
...     [("bar", "foobarbar", 5,)], ["a", "b", "c"]
... ).select(sf.position("a", "b")).show()
+-----------------+
|position(a, b, 1)|
+-----------------+
|                4|
+-----------------+
```

--------------------------------

### joinVertices with explicit type annotations

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Shows the alternative non-curried form of joinVertices that requires explicit type annotations because type inference on the function parameters does not depend on the table argument. This is contrasted with the curried form used in earlier examples.

```scala
val joinedGraph = graph.joinVertices(uniqueCosts,
  (id: VertexId, oldCost: Double, extraCost: Double) => oldCost + extraCost)
```

--------------------------------

### Graph.stronglyConnectedComponents

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Computes the strongly connected components of the graph.

```APIDOC
## [SCALA] Graph.stronglyConnectedComponents

### Description
Computes the strongly connected components of the graph.

### Method
Function (def)

### Endpoint
Graph.stronglyConnectedComponents(numIter: Int): Graph[VertexId, ED]

### Parameters
- **numIter** (Int) - Required - The number of iterations to run.

### Returns
- **Graph[VertexId, ED]** - A graph where each vertex attribute is the ID of its strongly connected component.
```

--------------------------------

### startApplication(SparkAppHandle.Listener... listeners)

Source: https://spark.apache.org/docs/3.5.6/api/java/org/apache/spark/launcher/AbstractLauncher.html

Starts a Spark application. Returns a handle that provides information about the running application and can be used for basic interaction. Throws java.io.IOException.

```APIDOC
## startApplication(SparkAppHandle.Listener... listeners)

### Description
Starts a Spark application. This method returns a handle that provides information about the running application and can be used to do basic interaction with it. The returned handle assumes that the application will instantiate a single SparkContext during its lifetime. Once that context reports a final state (one that indicates the SparkContext has stopped), the handle will not perform new state transitions, so anything that happens after that cannot be monitored. If the underlying application is launched as a child process, `SparkAppHandle.kill()` can still be used to kill the child process.

### Method
startApplication

### Parameters
- **listeners** (SparkAppHandle.Listener...) - Required - Listeners to add to the handle before the app is launched.

### Returns
A handle for the launched application.

### Throws
- **java.io.IOException**

### Since
1.6.0
```

--------------------------------

### Run Java streaming example with run-example

Source: https://spark.apache.org/docs/3.5.6/streaming-programming-guide.html

Runs the Java JavaNetworkWordCount example using the run-example script, connecting to localhost:9999. Run in a different terminal after starting Netcat.

```bash
$ ./bin/run-example streaming.JavaNetworkWordCount localhost 9999
```

--------------------------------

### Graph.triangleCount

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Computes the number of triangles each vertex is part of.

```APIDOC
## [SCALA] Graph.triangleCount

### Description
Computes the number of triangles each vertex is part of.

### Method
Function (def)

### Endpoint
Graph.triangleCount(): Graph[Int, ED]

### Parameters
None

### Returns
- **Graph[Int, ED]** - A graph where each vertex attribute is the number of triangles the vertex participates in.
```

--------------------------------

### Create SparkSession

Source: https://spark.apache.org/docs/3.5.6/api/python/getting_started/quickstart_ps.html

Creates or gets a SparkSession using the builder pattern.

```python
spark = SparkSession.builder.getOrCreate()
```

--------------------------------

### Graph.triplets

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Returns the triplets of the graph as an RDD of EdgeTriplet.

```APIDOC
## [SCALA] Graph.triplets

### Description
Returns the triplets of the graph as an RDD of EdgeTriplet.

### Method
Property accessor (val)

### Endpoint
Graph.triplets

### Parameters
None

### Returns
- **RDD[EdgeTriplet[VD, ED]]** - An RDD of EdgeTriplet objects containing the source and destination vertex attributes and the edge attribute.
```

--------------------------------

### abstract def start(): Unit

Source: https://spark.apache.org/docs/3.5.6/api/scala/org/apache/spark/streaming/dstream/InputDStream.html

Abstract method called to start receiving data. Subclasses must implement this method.

```APIDOC
## abstract def start(): Unit

### Description
Method called to start receiving data. Subclasses must implement this method.

```

--------------------------------

### Remove broken links with GraphX subgraph operator

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Builds a graph from users and relationships RDDs, then uses subgraph with a vertex predicate to remove missing vertices and their edges. The example prints triplets before and after filtering; subgraph defaults to true for any predicate not provided.

```scala
// Create an RDD for the vertices
val users: RDD[(VertexId, (String, String))] =
  sc.parallelize(Seq((3L, ("rxin", "student")), (7L, ("jgonzal", "postdoc")),
                       (5L, ("franklin", "prof")), (2L, ("istoica", "prof")),
                       (4L, ("peter", "student"))))
// Create an RDD for edges
val relationships: RDD[Edge[String]] =
  sc.parallelize(Seq(Edge(3L, 7L, "collab"),    Edge(5L, 3L, "advisor"),
                       Edge(2L, 5L, "colleague"), Edge(5L, 7L, "pi"),
                       Edge(4L, 0L, "student"),   Edge(5L, 0L, "colleague")))
// Define a default user in case there are relationship with missing user
val defaultUser = ("John Doe", "Missing")
// Build the initial Graph
val graph = Graph(users, relationships, defaultUser)
// Notice that there is a user 0 (for which we have no information) connected to users
// 4 (peter) and 5 (franklin).
graph.triplets.map(
  triplet => triplet.srcAttr._1 + " is the " + triplet.attr + " of " + triplet.dstAttr._1
).collect.foreach(println(_))
// Remove missing vertices as well as the edges to connected to them
val validGraph = graph.subgraph(vpred = (id, attr) => attr._2 != "Missing")
// The valid subgraph will disconnect users 4 and 5 by removing user 0
validGraph.vertices.collect.foreach(println(_))
validGraph.triplets.map(
  triplet => triplet.srcAttr._1 + " is the " + triplet.attr + " of " + triplet.dstAttr._1
).collect.foreach(println(_))
```

--------------------------------

### RDD.repartition example

Source: https://spark.apache.org/docs/3.5.6/api/python/_modules/pyspark/rdd.html

Demonstrates repartitioning an RDD to a different number of partitions. The example shows the initial partition contents and the resulting partition counts after repartitioning to 2 and 10 partitions.

```python
>>> rdd = sc.parallelize([1,2,3,4,5,6,7], 4)
>>> sorted(rdd.glom().collect())
[[1], [2, 3], [4, 5], [6, 7]]
>>> len(rdd.repartition(2).glom().collect())
2
>>> len(rdd.repartition(10).glom().collect())
10
```

--------------------------------

### Graph.pregel

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Executes the Pregel algorithm for iterative graph-parallel computation.

```APIDOC
## [SCALA] Graph.pregel

### Description
Executes the Pregel algorithm for iterative graph-parallel computation.

### Method
Function (def)

### Endpoint
Graph.pregel[A](initialMsg: A, maxIterations: Int, activeDirection: EdgeDirection)(vprog: (VertexId, VD, A) => VD, sendMsg: EdgeTriplet[VD, ED] => Iterator[(VertexId, A)], mergeMsg: (A, A) => A): Graph[VD, ED]

### Parameters
- **initialMsg** (A) - Required - The initial message sent to all vertices.
- **maxIterations** (Int) - Required - The maximum number of iterations to run.
- **activeDirection** (EdgeDirection) - Required - The direction of edges to consider for active vertices.
- **vprog** (Function) - Required - Vertex program function.
- **sendMsg** (Function) - Required - Function to send messages along edges.
- **mergeMsg** (Function) - Required - Function to merge messages.

### Returns
- **Graph[VD, ED]** - The resulting graph after the Pregel computation.
```

--------------------------------

### Graph.mask

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Restricts the graph to the vertices and edges that are also in the other graph.

```APIDOC
## [SCALA] Graph.mask

### Description
Restricts the graph to the vertices and edges that are also in the other graph.

### Method
Function (def)

### Endpoint
Graph.mask[VD2, ED2](other: Graph[VD2, ED2]): Graph[VD, ED]

### Parameters
- **other** (Graph) - Required - The other graph to mask with.

### Returns
- **Graph[VD, ED]** - A new graph containing only the vertices and edges that exist in both graphs.
```

--------------------------------

### Start Spark shell (Scala)

Source: https://spark.apache.org/docs/3.5.6/quick-start.html

Launches the interactive Spark shell in Scala from the Spark directory.

```bash
./bin/spark-shell
```

--------------------------------

### Graph.unpersistVertices

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Unpersists the vertices of the graph, optionally blocking until completion.

```APIDOC
## [SCALA] Graph.unpersistVertices

### Description
Unpersists the vertices of the graph, optionally blocking until completion.

### Method
Function (def)

### Endpoint
Graph.unpersistVertices(blocking: Boolean = false): Graph[VD, ED]

### Parameters
- **blocking** (Boolean) - Optional - Whether to block until the unpersist operation completes. Defaults to false.

### Returns
- **Graph[VD, ED]** - The graph with vertices unpersisted.
```

--------------------------------

### Run Python example with spark-submit

Source: https://spark.apache.org/docs/3.5.6/rdd-programming-guide.html

Use spark-submit to run Python examples. The example shown runs pi.py from the examples directory.

```bash
./bin/spark-submit examples/src/main/python/pi.py
```

--------------------------------

### sparkR.uiWebUrl() example (dontrun)

Source: https://spark.apache.org/docs/3.5.6/api/R/reference/sparkR.uiWebUrl.html

Example wrapped in if (FALSE) to prevent execution during package checks. Starts a SparkR session and retrieves the SparkUI URL.

```r
if (FALSE) { # \dontrun{
sparkR.session()
url <- sparkR.uiWebUrl()
} # }
```

--------------------------------

### start(): Unit

Source: https://spark.apache.org/docs/3.5.6/api/scala/org/apache/spark/streaming/dstream/ConstantInputDStream.html

Method called to start receiving data. Subclasses must implement this method.

```APIDOC
## start(): Unit

### Description
Method called to start receiving data. Subclasses must implement this method.

### Method
start
```

--------------------------------

### Configure Spark for Arrow Optimization

Source: https://spark.apache.org/docs/3.5.6/api/python/getting_started/quickstart_ps.html

Sets up Spark and pandas-on-Spark configurations to enable Arrow optimization for faster pandas conversion, including saving the previous setting and ignoring warnings.

```python
prev = spark.conf.get("spark.sql.execution.arrow.pyspark.enabled")  # Keep its default value.
ps.set_option("compute.default_index_type", "distributed")  # Use default index prevent overhead.
import warnings
warnings.filterwarnings("ignore")  # Ignore warnings coming from Arrow optimizations.
```

--------------------------------

### currentCatalog() example

Source: https://spark.apache.org/docs/3.5.6/api/R/reference/currentCatalog.html

Example usage wrapped in a dontrun block. Starts a SparkR session and then calls currentCatalog() to get the current default catalog name.

```R
if (FALSE) { # \dontrun{
sparkR.session()
currentCatalog()
} # }
```

--------------------------------

### ADD FILE Examples

Source: https://spark.apache.org/docs/3.5.6/sql-ref-syntax-aux-resource-mgmt-add-file.html

Examples of adding files and directories, including paths with spaces and multiple files using ADD FILES.

```sql
ADD FILE /tmp/test;
ADD FILE "/path/to/file/abc.txt";
ADD FILE '/another/test.txt';
ADD FILE "/path with space/abc.txt";
ADD FILE "/path/to/some/directory";
ADD FILES "/path with space/cde.txt" '/path with space/fgh.txt';
```

--------------------------------

### Graph.subgraph

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Returns a subgraph of the graph based on edge and vertex predicates.

```APIDOC
## [SCALA] Graph.subgraph

### Description
Returns a subgraph of the graph based on edge and vertex predicates.

### Method
Function (def)

### Endpoint
Graph.subgraph(epred: EdgeTriplet[VD,ED] => Boolean = (x => true), vpred: (VertexId, VD) => Boolean = ((v, d) => true)): Graph[VD, ED]

### Parameters
- **epred** (Function) - Optional - Edge predicate. Defaults to always true.
- **vpred** (Function) - Optional - Vertex predicate. Defaults to always true.

### Returns
- **Graph[VD, ED]** - A subgraph containing only vertices and edges that satisfy the predicates.
```

--------------------------------

### startApplication(SparkAppHandle.Listener... listeners)

Source: https://spark.apache.org/docs/3.5.6/api/java/org/apache/spark/launcher/SparkLauncher.html

Starts a Spark application.

```APIDOC
## startApplication(SparkAppHandle.Listener... listeners)

### Description
Starts a Spark application.

### Method
startApplication

### Parameters
- **listeners** (SparkAppHandle.Listener...) - Required - Listeners for the application handle.

### Returns
SparkAppHandle - The handle for the started application.
```

--------------------------------

### Graph.mapTriplets

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Transforms the edge triplets of the graph using a user-defined function.

```APIDOC
## [SCALA] Graph.mapTriplets

### Description
Transforms the edge triplets of the graph using a user-defined function.

### Method
Function (def)

### Endpoint
Graph.mapTriplets[ED2](map: EdgeTriplet[VD, ED] => ED2): Graph[VD, ED2]

### Parameters
- **map** (Function) - Required - A function that takes an EdgeTriplet and returns a new edge attribute.

### Returns
- **Graph[VD, ED2]** - A new graph with transformed edge attributes.
```

--------------------------------

### Run Python streaming example with spark-submit

Source: https://spark.apache.org/docs/3.5.6/streaming-programming-guide.html

Submits the Python network_wordcount.py example to Spark, connecting to localhost:9999. Run in a different terminal after starting Netcat.

```bash
$ ./bin/spark-submit examples/src/main/python/streaming/network_wordcount.py localhost 9999
```

--------------------------------

### BucketedRandomProjectionLSH usage example with fit, transform, and similarity joins

Source: https://spark.apache.org/docs/3.5.6/api/python/_modules/pyspark/ml/feature.html

Shows how to create a BucketedRandomProjectionLSH, set parameters, fit a model, transform data, and use approxNearestNeighbors and approxSimilarityJoin. The example uses a Spark DataFrame with dense vectors and demonstrates the expected output rows and distance values.

```python
>>> from pyspark.ml.linalg import Vectors
>>> from pyspark.sql.functions import col
>>> data = [(0, Vectors.dense([-1.0, -1.0 ]),),
...         (1, Vectors.dense([-1.0, 1.0 ]),),
...         (2, Vectors.dense([1.0, -1.0 ]),),
...         (3, Vectors.dense([1.0, 1.0]),)]
>>> df = spark.createDataFrame(data, ["id", "features"])
>>> brp = BucketedRandomProjectionLSH()
>>> brp.setInputCol("features")
BucketedRandomProjectionLSH...
>>> brp.setOutputCol("hashes")
BucketedRandomProjectionLSH...
>>> brp.setSeed(12345)
BucketedRandomProjectionLSH...
>>> brp.setBucketLength(1.0)
BucketedRandomProjectionLSH...
>>> model = brp.fit(df)
>>> model.getBucketLength()
1.0
>>> model.setOutputCol("hashes")
BucketedRandomProjectionLSHModel...
>>> model.transform(df).head()
Row(id=0, features=DenseVector([-1.0, -1.0]), hashes=[DenseVector([-1.0])])
>>> data2 = [(4, Vectors.dense([2.0, 2.0 ]),),
...          (5, Vectors.dense([2.0, 3.0 ]),),
...          (6, Vectors.dense([3.0, 2.0 ]),),
...          (7, Vectors.dense([3.0, 3.0]),)]
>>> df2 = spark.createDataFrame(data2, ["id", "features"])
>>> model.approxNearestNeighbors(df2, Vectors.dense([1.0, 2.0]), 1).collect()
[Row(id=4, features=DenseVector([2.0, 2.0]), hashes=[DenseVector([1.0])], distCol=1.0)]
>>> model.approxSimilarityJoin(df, df2, 3.0, distCol="EuclideanDistance").select(
...     col("datasetA.id").alias("idA"),
...     col("datasetB.id").alias("idB"),
...     col("EuclideanDistance")).show()
+---+---+-----------------+
|idA|idB|EuclideanDistance|
+---+---+-----------------+
|  3|  6| 2.23606797749979|
+---+---+-----------------+
...
>>> model.approxSimilarityJoin(df, df2, 3, distCol="EuclideanDistance").select(
...     col("datasetA.id").alias("idA"),
...     col("datasetB.id").alias("idB"),
...     col("EuclideanDistance")).show()
+---+---+-----------------+
|idA|idB|EuclideanDistance|
+---+---+-----------------+
|  3|  6| 2.23606797749979|
+---+---+-----------------+
...
```

--------------------------------

### Graph.mapEdges

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Transforms the edge attributes of the graph using a user-defined function.

```APIDOC
## [SCALA] Graph.mapEdges

### Description
Transforms the edge attributes of the graph using a user-defined function.

### Method
Function (def)

### Endpoint
Graph.mapEdges[ED2](map: Edge[ED] => ED2): Graph[VD, ED2]

### Parameters
- **map** (Function) - Required - A function that takes an Edge and returns a new edge attribute.

### Returns
- **Graph[VD, ED2]** - A new graph with transformed edge attributes.
```

--------------------------------

### Graph.mapVertices

Source: https://spark.apache.org/docs/3.5.6/graphx-programming-guide.html

Transforms the vertex attributes of the graph using a user-defined function.

```APIDOC
## [SCALA] Graph.mapVertices

### Description
Transforms the vertex attributes of the graph using a user-defined function.

### Method
Function (def)

### Endpoint
Graph.mapVertices[VD2](map: (VertexId, VD) => VD2): Graph[VD2, ED]

### Parameters
- **map** (Function) - Required - A function that takes a VertexId and the current vertex attribute and returns a new vertex attribute.

### Returns
- **Graph[VD2, ED]** - A new graph with transformed vertex attributes.
```

--------------------------------

### DESCRIBE DATABASE Example with Output

Source: https://spark.apache.org/docs/3.5.6/sql-ref-syntax-aux-describe-database.html

Creates the employees database and then describes it. The output shows the database name, description, and root location on the filesystem.

```sql
-- Create employees DATABASE
CREATE DATABASE employees COMMENT 'For software companies';

-- Describe employees DATABASE.
-- Returns Database Name, Description and Root location of the filesystem
-- for the employees DATABASE.
DESCRIBE DATABASE employees;
+-------------------------+-----------------------------+
|database_description_item|   database_description_value|
+-------------------------+-----------------------------+
|            Database Name|                    employees|
|              Description|       For software companies|
|                 Location|file:/Users/Temp/employees.db|
+-------------------------+-----------------------------+
```
Tokens:
Raw