<jupyter_start><jupyter_text>Spark EDA<jupyter_code>from pyspark.sql import SparkSession

spark = SparkSession.builder \
                    .appName('Spark EDA') \
                    .getOrCreate()
data_path = '/home/lorenzo/spark-repo/1_spark_dataframes/data/california.csv'

df1 = spark.read.option('header', 'True') \
                .option('inferSchema', 'True') \
                .csv(data_path)

df1 = df1.withColumnRenamed('latitude', 'lat') \
         .withColumnRenamed('longitude', 'lng') \
         .withColumnRenamed('total_rooms', 'rooms') \
         .withColumnRenamed('total_bedrooms', 'bedrooms') \
         .withColumnRenamed('median_income', 'income') \
         .withColumnRenamed('median_house_value', 'value') \
         .withColumnRenamed('housing_median_age', 'age') 

df1.createOrReplaceTempView('california')<jupyter_output><empty_output><jupyter_text>Descriptive statistics**Describe() method:**<jupyter_code>df1.select('rooms', 'income', 'age', 'value').describe().show()<jupyter_output>+-------+------------------+------------------+------------------+------------------+
|summary|             rooms|            income|               age|             value|
+-------+------------------+------------------+------------------+------------------+
|  count|             20640|             20640|             20640|             20640|
|   mean|2635.7630813953488|3.8706710029070246|28.639486434108527|206855.81690891474|
| stddev|2181.6152515827944| 1.899821717945263| 12.58555761211163|115395.61587441359|
|    min|               2.0|            0.4999|               1.0|           14999.0|
|    max|           39320.0|           15.0001|              52.0|          500001.0|
+-------+------------------+------------------+------------------+------------------+<jupyter_text>**Correlations:**<jupyter_code>df1.stat.corr('rooms', 'value')
df1 = df1.withColumn('rooms_per_person', df1.rooms/df1.population)
df1.stat.corr('rooms_per_person', 'value')<jupyter_output><empty_output><jupyter_text>**Frequent values:**<jupyter_code>df1.stat.freqItems(['age']).show()
df1.stat.freqItems(['age', 'value']).show()<jupyter_output>+--------------------+--------------------+
|       age_freqItems|     value_freqItems|
+--------------------+--------------------+
|[23.0, 32.0, 41.0...|[55000.0, 68700.0...|
+--------------------+--------------------+<jupyter_text>**Crosstable:**<jupyter_code>df1.filter(df1.age <= 10).stat.crosstab('ocean_proximity', 'age').show()<jupyter_output>+-------------------+---+----+---+---+---+---+---+---+---+---+
|ocean_proximity_age|1.0|10.0|2.0|3.0|4.0|5.0|6.0|7.0|8.0|9.0|
+-------------------+---+----+---+---+---+---+---+---+---+---+
|             INLAND|  4| 151| 33| 33| 77|122| 80|104|112|108|
|           NEAR BAY|  0|  19|  4|  4|  9| 17|  7|  4|  7|  4|
|         NEAR OCEAN|  0|  16|  2|  2| 26| 20| 11| 18| 18| 23|
|          <1H OCEAN|  0|  78| 19| 23| 79| 85| 62| 49| 69| 70|
+-------------------+---+----+---+---+---+---+---+---+---+---+<jupyter_text>**Bucketing:**<jupyter_code>spark.sql("SELECT count(*), FLOOR(age*10) as age_bucket \
           FROM california GROUP BY age_bucket ORDER BY age_bucket").show()
spark.sql("SELECT count(*), round(mean(value), 2), FLOOR(age*10) as age_bucket \
           FROM california GROUP BY age_bucket ORDER BY age_bucket").show()<jupyter_output>+--------+--------------------+----------+
|count(1)|round(avg(value), 2)|age_bucket|
+--------+--------------------+----------+
|       4|            144300.0|        10|
|      58|           224475.91|        20|
|      62|           235643.58|        30|
|     191|           229235.14|        40|
|     244|           208417.66|        50|
|     160|           203794.39|        60|
|     175|           193296.03|        70|
|     206|           194414.58|        80|
|     205|            186672.7|        90|
|     264|            176580.7|       100|
|     254|            179907.9|       110|
|     238|           182046.24|       120|
|     302|           190181.81|       130|
|     412|            189597.1|       140|
|     512|           181808.42|       150|
|     771|           200180.56|       160|
|     698|           190494.29|       170|
|     570|           193354.77|       180|
|     502|           193197.84|       190|
|     465|           195659.82|       200|
+--------+-[...]<jupyter_text>Time-series data<jupyter_code>data_path = '/home/lorenzo/Desktop/utilization.csv'

df2 = spark.read.option('header', 'False') \
                .option('inferSchema', 'True') \
                .csv(data_path)
df2 = df2.withColumnRenamed("_c0", "event_datetime") \
            .withColumnRenamed ("_c1", "server_id") \
            .withColumnRenamed("_c2", "cpu_utilization") \
            .withColumnRenamed("_c3", "free_memory") \
            .withColumnRenamed("_c4", "session_count")

df2.createOrReplaceTempView('utilization')
df2.show(5)
spark.sql("SELECT server_id, min(cpu_utilization), max(cpu_utilization) \
           FROM utilization \
           GROUP BY server_id").show(5)<jupyter_output>+---------+--------------------+--------------------+
|server_id|min(cpu_utilization)|max(cpu_utilization)|
+---------+--------------------+--------------------+
|      148|                0.54|                0.94|
|      137|                0.54|                0.94|
|      133|                0.55|                0.95|
|      108|                0.55|                0.95|
|      101|                 0.6|                 1.0|
+---------+--------------------+--------------------+
only showing top 5 rows<jupyter_text>**Windowing:**<jupyter_code>spark.sql("SELECT event_datetime, server_id, cpu_utilization, \
                  avg(cpu_utilization) OVER (PARTITION BY server_id) as avg_cpu_util \
           FROM utilization").show(10)
spark.sql("SELECT event_datetime, server_id, cpu_utilization, \
                  avg(cpu_utilization) OVER (PARTITION BY server_id) as avg_cpu_util, \
                  cpu_utilization - avg(cpu_utilization) OVER (PARTITION BY server_id) as delta_cpu_util\
           FROM utilization").show(10)<jupyter_output>+-------------------+---------+---------------+------------------+--------------------+
|     event_datetime|server_id|cpu_utilization|      avg_cpu_util|      delta_cpu_util|
+-------------------+---------+---------------+------------------+--------------------+
|03/05/2019 08:07:41|      148|           0.85|0.7393840000000045|  0.1106159999999955|
|03/05/2019 08:12:41|      148|           0.94|0.7393840000000045| 0.20061599999999546|
|03/05/2019 08:17:41|      148|           0.89|0.7393840000000045| 0.15061599999999553|
|03/05/2019 08:22:41|      148|           0.74|0.7393840000000045| 6.15999999995509E-4|
|03/05/2019 08:27:41|      148|           0.63|0.7393840000000045|-0.10938400000000448|
|03/05/2019 08:32:41|      148|           0.89|0.7393840000000045| 0.15061599999999553|
|03/05/2019 08:37:41|      148|           0.77|0.7393840000000045|0.030615999999995536|
|03/05/2019 08:42:41|      148|           0.59|0.7393840000000045| -0.1493840000000045|
|03/05/2019 08:47:41|      148| [...]<jupyter_text>**Moving window:**<jupyter_code>spark.sql("SELECT event_datetime, server_id, cpu_utilization, \
                  avg(cpu_utilization) OVER (PARTITION BY server_id ORDER BY event_datetime  \
                        ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING) as avg_cpu_util \
           FROM utilization").show(10)<jupyter_output>+-------------------+---------+---------------+------------------+
|     event_datetime|server_id|cpu_utilization|      avg_cpu_util|
+-------------------+---------+---------------+------------------+
|03/05/2019 08:07:41|      148|           0.85|             0.895|
|03/05/2019 08:12:41|      148|           0.94|0.8933333333333334|
|03/05/2019 08:17:41|      148|           0.89|0.8566666666666668|
|03/05/2019 08:22:41|      148|           0.74|0.7533333333333333|
|03/05/2019 08:27:41|      148|           0.63|0.7533333333333334|
|03/05/2019 08:32:41|      148|           0.89|0.7633333333333333|
|03/05/2019 08:37:41|      148|           0.77|              0.75|
|03/05/2019 08:42:41|      148|           0.59|              0.71|
|03/05/2019 08:47:41|      148|           0.77|              0.69|
|03/05/2019 08:52:41|      148|           0.71|0.7766666666666667|
+-------------------+---------+---------------+------------------+
only showing top 10 rows
{'fraction_non_alphanumeric': 0.26514882011146684, 'fraction_numerical': 0.16933475631447883, 'mean_word_length': 1.448896631823461, 'pattern_counts': {'":': 0, '<': 32, '<?xml version=': 0, '>': 30, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 22, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '501138', 'license': 'Apache-2.0', 'n_tokens_mistral': 3722, 'n_tokens_neox': 2776, 'n_words': 421, 'path': '3_spark_eda/spark_eda.ipynb', 'repo_name': 'Lorenzo-Giardi/spark-repo', 'size': '8433'}