<jupyter_start><jupyter_text>Chapter 7: Optimizing and Tuning Spark Applications<NAME>    May, 2022   Python examples of chapter 7 (page 173 ff) in the book *Learning Spark*<jupyter_code># Import required python spark libraries
import pyspark
from pyspark.sql.functions import col, expr, when, concat, lit, avg
from pyspark.sql.types import StructType,StructField, StringType, IntegerType

#create a SparkSession

spark = (SparkSession \
         .builder \
         .enableHiveSupport() \
         .appName("Chapter_7") \
         .getOrCreate())

# Show the content of the environment variable $SPARK_HOME:
!echo $SPARK_HOME
# Show all config files
!ls -l $SPARK_HOME/conf
#Get single Spark configuiration values:
print(spark.conf.get("spark.sql.warehouse.dir"))
# Get the whole confiuguration context of a Spark Context:
scConf = sc.getConf().getAll()

for l in scConf:
    print (l[0] + ":")
    print (l[1])
    print ()

# Change single Spark config variables
spark.conf.set("spark.sql.shuffle.partitions", spark.sparkContext.defaultParallelism)
# Show the Spark SQL-specifdic Spark configs:
spark.sql("SET -v").select("key", "value").show(truncate=False)<jupyter_output><empty_output><jupyter_text>Spark's Web InterfaceTo see Spark's Web Interface, go the web address: http://127.0.0.1:4040    The tab *Environment* shows all environment variables. In the web interface, the variables are *read-only*, they cannot be modified.  Set configuration variables in a Spark program<jupyter_code># First check, if a configuration variable is modifiable:

# Example:
spark.conf.isModifiable("spark.sql.shuffle.partitions")
# Get the actual value of the variable:
spark.conf.get("spark.sql.shuffle.partitions")
# Set the variable to a new variable and check:
spark.conf.set("spark.sql.shuffle.partitions", 5)
spark.conf.get("spark.sql.shuffle.partitions")
# Set it back to the old value:
spark.conf.set("spark.sql.shuffle.partitions", 8)
spark.conf.get("spark.sql.shuffle.partitions")<jupyter_output><empty_output><jupyter_text>PartitionsPage 181 ff.<jupyter_code># Create a big DataFrame:
numDF = spark.range(1000 * 1000)
# Get the default number of partitions of this DataFrame
numDF.rdd.getNumPartitions()
# Now change the number of partitions to another value
numDF = spark.range(1000 * 1000).repartition(32)
# Check the number of partitions>:
numDF.rdd.getNumPartitions()<jupyter_output><empty_output><jupyter_text>Caching of DataPage 183 ff.Create a DataFrame with 10M records.  The time difference (approx. 10x faster) between *Count and load into cache*' and *Count in cache*can only be demonstrated when this code is run the first time in the notebook. In consecutive executions the DataFrame is already cached and there is basically no time difference.<jupyter_code>import time

start = time.time()
df = spark.range(1 * 10000000).toDF("id")
end = time.time()
print("Step 1 - Create:                    %f seconds" %(end - start))

start = time.time()
df = df.withColumn("square", df.id * df.id)
end = time.time()
print("Step 2 - Add Column:                %f seconds" %(end - start))

start = time.time()
df.cache()
end = time.time()
print("Step 3 - Cache df:                  %f seconds" %(end - start))

start = time.time()
df.count()
end = time.time()
print("Step 4 - Count and load into cache: %f seconds" %(end - start))

start = time.time()
df.count()
end = time.time()
print("Step 5 - Count in cache:            %f seconds" %(end - start))<jupyter_output><empty_output><jupyter_text>Caching Tables and Views in SQLIt is also possible to cache tables of views:<jupyter_code>df.createOrReplaceTempView("dfTable")
spark.sql("CACHE TABLE dfTable")
spark.sql("SELECT count(*) FROM dfTable").show()<jupyter_output><empty_output><jupyter_text>Persistance of DataPage 184 ffPersistance of data is synonymous to caching data, but let you apecify how the data is persisted with the parameter `pyspark.StorageLevel.LEVEL`. As we have specified the persistance on disk only, the time difference is much lower compared to the example above (this time approx. 5x faster compared to 12x faster above). Under the link http://127.0.0.1:4040/ you can see that the data is persisted on disk and not on memory for all partitions.<jupyter_code>start = time.time()
df2 = spark.range(1 * 10000000).toDF("id")
end = time.time()
print("Step 1 - Create:                    %f seconds" %(end - start))

start = time.time()
df2 = df2.withColumn("square", df2.id * df2.id)
end = time.time()
print("Step 2 - Add Column:                %f seconds" %(end - start))

start = time.time()
df2.persist(storageLevel=pyspark.StorageLevel.DISK_ONLY)
end = time.time()
print("Step 3 - Persist df DISK_ONLY:      %f seconds" %(end - start))

start = time.time()
df2.count()
end = time.time()
print("Step 4 - Count and load into cache: %f seconds" %(end - start))

start = time.time()
df2.count()
end = time.time()
print("Step 5 - Count in cache:            %f seconds" %(end - start))<jupyter_output><empty_output><jupyter_text>Shuffle Sort Merge Join (SMJ)Page 189 ff.<jupyter_code>from random import randint

# Disable broadcast join
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")

# Generate synthetic data for two data frames
states = ['AZ', 'CO', 'CA', 'TX', 'NY', 'MI']
items  = ['SKU-0', 'SKU-1', 'SKU-2', 'SKU-3', 'SKU-4']

usersDF_schema = StructType([ \
    StructField("uid",StringType(),True), \
    StructField("login",StringType(),True), \
    StructField("email",StringType(),True), \
    StructField("user_state", StringType(), True) \
  ])

ordersDF_schema = StructType([ \
    StructField("transaction_id",StringType(),True), \
    StructField("quantity",StringType(),True), \
    StructField("users_id",StringType(),True), \
    StructField("amount", StringType(), True), \
    StructField("state", StringType(), True), \
    StructField("items", StringType(), True) \
])

usersDF_data = []
for i in range (100000):
    login = "user_{}".format(i)
    usersDF_row = (str(i), login, login + "@databricks.com",
                    states[randint(0, 5)])
    usersDF_data.append(usersDF_row)

usersDF = spark.createDataFrame(data=usersDF_data, schema=usersDF_schema)
usersDF.show(n=10)
ordersDF_data = []
for i in range (100000):
    login = "user_{}".format(i)
    ordersDF_row = (str(randint(100000, 999999)), 
                   str(randint(1, 100)),
                   str(randint(0, 9999)),
                   str((randint(10, 9999)/3.14)),
                   states[randint(0, 5)],
                   items[randint(0, 4)])
    ordersDF_data.append(ordersDF_row)

ordersDF = spark.createDataFrame(data=ordersDF_data,
                                 schema=ordersDF_schema)

ordersDF.show(n=10)
usersOrdersDF = ordersDF.join(usersDF, ordersDF.users_id == usersDF.uid)
usersOrdersDF.show()
usersOrdersDF.explain()<jupyter_output>== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- SortMergeJoin [users_id#27], [uid#0], Inner
   :- Sort [users_id#27 ASC NULLS FIRST], false, 0
   :  +- Exchange hashpartitioning(users_id#27, 200), ENSURE_REQUIREMENTS, [id=#136]
   :     +- Filter isnotnull(users_id#27)
   :        +- Scan ExistingRDD[transaction_id#25,quantity#26,users_id#27,amount#28,state#29,items#30]
   +- Sort [uid#0 ASC NULLS FIRST], false, 0
      +- Exchange hashpartitioning(uid#0, 200), ENSURE_REQUIREMENTS, [id=#137]
         +- Filter isnotnull(uid#0)
            +- Scan ExistingRDD[uid#0,login#1,email#2,user_state#3]<jupyter_text>Optimizing the Shuffle Sort Merge JoinPage 193 ff.<jupyter_code>(usersDF
     .orderBy(col("uid").asc())
     .write.format("parquet")
     .bucketBy(8, "uid")
     .mode("overWrite")
     .saveAsTable("UserTbl")
)
(ordersDF
     .orderBy(col("users_id").asc())
     .write.format("parquet")
     .bucketBy(8, "users_id")
     .mode("overWrite")
     .saveAsTable("OrderTbl")
)
spark.sql("CACHE TABLE UserTbl")
spark.sql("CACHE TABLE OrderTbl")
userBucketDF  = spark.table("UserTbl")
orderBucketDF = spark.table("OrderTbl")

joinUserOrderBucketDF = orderBucketDF.join(userBucketDF, orderBucketDF.users_id == userBucketDF.uid)

joinUserOrderBucketDF.show()
joinUserOrderBucketDF.explain()<jupyter_output>== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- SortMergeJoin [users_id#292], [uid#153], Inner
   :- Sort [users_id#292 ASC NULLS FIRST], false, 0
   :  +- Filter isnotnull(users_id#292)
   :     +- Scan In-memory table OrderTbl [transaction_id#290, quantity#291, users_id#292, amount#293, state#294, items#295], [isnotnull(users_id#292)]
   :           +- InMemoryRelation [transaction_id#290, quantity#291, users_id#292, amount#293, state#294, items#295], StorageLevel(disk, memory, deserialized, 1 replicas)
   :                 +- *(1) ColumnarToRow
   :                    +- FileScan parquet default.ordertbl[transaction_id#290,quantity#291,users_id#292,amount#293,state#294,items#295] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/Users/cwi/Dev/LearningSpark/spark-warehouse/ordertbl], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<transaction_id:string,quantity:string,users_id:string,amount:string,state:string,items:str[...]
{'fraction_non_alphanumeric': 0.136001745010361, 'fraction_numerical': 0.031082997055295016, 'mean_word_length': 4.279217040875072, 'pattern_counts': {'":': 1, '<': 33, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '644190', 'license': 'Apache-2.0', 'n_tokens_mistral': 3263, 'n_tokens_neox': 2960, 'n_words': 764, 'path': 'chapter_07_TuningSpark.ipynb', 'repo_name': 'ciwin/Spark', 'size': '9183'}