<jupyter_start><jupyter_text>Whole stage codegenWhole-Stage Code Generation (aka Whole-Stage CodeGen) fuses multiple operators (as a subtree of plans that support code generation) together into a single Java function that is aimed at improving execution performance. It collapses a query into a single optimized function that eliminates virtual function calls and leverages CPU registers for intermediate data.- Identity chains of operators (“stage” is a chains of trainsformations followed by a shuffle)- Compile each stage into a single function (i.e. one can have map -> map -> flatMap -> groupBy -> sum() compiled into a single function)- Functionality of a general purpose execution engine; performance as if hand built system just to run your query**Note: Janino is used in Spark to compile a Java source code into a Java class.**<jupyter_code>import time

# Define a simple benchmark function for measuring time taken
def benchmark(name, f):
    startTime = time.time()
    f()
    endTime = time.time()
    print ("Time taken in %s: %.4f seconds" % (name, endTime - startTime))<jupyter_output><empty_output><jupyter_text>How fast can Spark 1.6 sum up 1 billion numbers?<jupyter_code># This config turns off whole stage code generation, effectively changing the execution path to be similar to Spark 1.6.
spark.conf.set("spark.sql.codegen.wholeStage", "false")
f = lambda: spark.range(1000 * 1000 * 1000).selectExpr("sum(id)").show()
benchmark("Spark 1.6", f)<jupyter_output>+------------------+
|           sum(id)|
+------------------+
|499999999500000000|
+------------------+

Time taken in Spark 1.6: 23.9768 seconds<jupyter_text>How fast can Spark 1.6 join 1 billion records?<jupyter_code># This config turns off whole stage code generation, effectively changing the execution path to be similar to Spark 1.6.
spark.conf.set("spark.sql.codegen.wholeStage", "false")
f = lambda: spark.range(1000 * 1000 * 1000).join(spark.range(1000), "id").count()
benchmark("Spark 1.6", f)<jupyter_output>Time taken in Spark 1.6: 37.7517 seconds<jupyter_text>How fast can Spark 2.0 sum up 1 billion numbers?<jupyter_code># Now we turn on whole stage code generation to get the full Spark 2.0 experience
spark.conf.set("spark.sql.codegen.wholeStage", "true")
f = lambda: spark.range(1000 * 1000 * 1000).selectExpr("sum(id)").show()
benchmark("Spark 2.0", f)<jupyter_output>+------------------+
|           sum(id)|
+------------------+
|499999999500000000|
+------------------+

Time taken in Spark 2.0: 5.2450 seconds<jupyter_text>How fast can Spark 2.0 join 1 billion records?<jupyter_code>spark.conf.set("spark.sql.codegen.wholeStage", "true")
f = lambda: spark.range(1000 * 1000 * 1000).join(spark.range(1000), "id").count()
benchmark("Spark 2.0", f)<jupyter_output>Time taken in Spark 2.0: 0.7305 seconds<jupyter_text>How fast can NumPy sum up 1 billion numbers?<jupyter_code>import numpy as np

f = lambda: np.sum(np.arange(1000 * 1000 * 1000))
benchmark("NumPy", f)<jupyter_output><empty_output><jupyter_text>How fast can Pandas join 100 million numbers?<jupyter_code>import numpy as np
import pandas as pd

a = pd.DataFrame({'id': np.arange(1000 * 1000 * 100)})
b = pd.DataFrame({'id': np.arange(1000)})
f = lambda: a.join(b, on='id', how='inner', lsuffix="_left", rsuffix="_right")
benchmark("Pandas", f)<jupyter_output><empty_output><jupyter_text>**Note: for this example, there's an obviously faster way if we can assume that we can just use indexing. For arbitrary keys, however, this approach won't work.**<jupyter_code>X = np.arange(1000 * 1000 * 1000)
keys = np.arange(1000)
f = lambda: np.shape(X[keys])[0]
benchmark("Numpy indexing", f)<jupyter_output><empty_output>
StarcoderdataJupyterStructuredCleanDedup