<jupyter_start><jupyter_text>Table of Contents1&nbsp;&nbsp;count distinct2&nbsp;&nbsp;Get distinct column values3&nbsp;&nbsp;Group by one column and filter rows with maximum3.1&nbsp;&nbsp;using leftsemi join3.2&nbsp;&nbsp;using sql3.3&nbsp;&nbsp;using pandas4&nbsp;&nbsp;Get average after filtering<jupyter_code>import numpy as np
import pandas as pd
import pyspark
from pyspark import SparkConf, SparkContext, SQLContext
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf # @udf("integer") def myfunc(x,y): return x - y
from pyspark.sql import functions as F # stddev format_number date_format, dayofyear, when
from pyspark.sql.types import StructField, StringType, IntegerType, StructType

print([(x.__name__,x.__version__) for x in [np, pd, pyspark]])

spark = pyspark.sql.SparkSession.builder.appName('example').getOrCreate()
sc = spark.sparkContext
sqlContext = SQLContext(sc)
sc.setLogLevel("INFO")<jupyter_output>[('numpy', '1.17.1'), ('pandas', '0.25.1'), ('pyspark', '2.4.4')]<jupyter_text>count distinct<jupyter_code>data = [("2001","id1"),("2002","id1"),
        ("2002","id1"),("2001","id1"),
        ("2001","id2"),("2001","id2"),
        ("2002","id2"),("2003","id1")]

df = spark.createDataFrame(data,["year","id"])
df.show()
# Note that countDistinct does not count Null as a distinct value! 
df.groupBy("year").agg(F.countDistinct("id").alias('count')).show()
df.count() # total number of rows
df.distinct().count()
df.select(F.countDistinct('year')).show()
dfp = pd.DataFrame(data,columns=['year','id'])
dfp
dfp.groupby(['year'])['id'].count().reset_index()
dfp.groupby('year').agg({'id': 'count'})<jupyter_output><empty_output><jupyter_text>Get distinct column values<jupyter_code>x = df.select('year').distinct()
x.show()
x.collect()
x.dropDuplicates(['year']).show()<jupyter_output>+----+
|year|
+----+
|2002|
|2001|
|2003|
+----+<jupyter_text>Group by one column and filter rows with maximum<jupyter_code>data = [
    ('a', 5, 'c'),
    ('a', 8, 'd'),
    ('a', 7, 'e'),
    ('b', 1, 'f'),
    ('b', 3, 'g')
]

df = spark.createDataFrame(data, ['A','B','C'])
df.show()
df.groupBy('A').agg({'B': 'max'}).show()
df.groupBy('A').agg(F.max('B')).show()<jupyter_output>+---+------+
|  A|max(B)|
+---+------+
|  b|     3|
|  a|     8|
+---+------+<jupyter_text>using leftsemi join<jupyter_code># we also want to get column with C column
df.join(df.groupBy('A').agg(F.max('B').alias('B')),on='B',how='leftsemi').show()<jupyter_output>+---+---+---+
|  B|  A|  C|
+---+---+---+
|  3|  b|  g|
|  8|  a|  d|
+---+---+---+<jupyter_text>using sql<jupyter_code>df.createOrReplaceTempView("table") # this is sql temporary view
q = '''SELECT *
FROM table a LEFT SEMI
JOIN (
    SELECT 
        A,
        max(B) as max_B
    FROM table
    GROUP BY A
    ) t
ON a.A=t.A AND a.B=t.max_B
'''

results = spark.sql(q)
results.show()<jupyter_output>+---+---+---+
|  A|  B|  C|
+---+---+---+
|  b|  3|  g|
|  a|  8|  d|
+---+---+---+<jupyter_text>using pandas<jupyter_code>dfp = df.toPandas()
dfp
pd.merge(dfp.groupby('A')['B'].max(), dfp)
dfp.groupby('A')['B'].max().to_frame().merge(dfp)
dfp.groupby('A',as_index=False)['B'].max().merge(dfp)<jupyter_output><empty_output><jupyter_text>Get average after filtering<jupyter_code>df.show()
df = df.withColumn('age', df.B )
df.show()
df.filter(df['B'] > 5).agg({'age': 'avg'}).show()
df.filter(df['B'] > 5).agg(F.avg(F.col('age'))).show()
from pyspark.sql.functions import avg, col, when
df.select(
    avg(
        when(df['B'] > 5, df['age'])
    )).show()<jupyter_output>+-----------------------------------+
|avg(CASE WHEN (B > 5) THEN age END)|
+-----------------------------------+
|                                7.5|
+-----------------------------------+
{'fraction_non_alphanumeric': 0.25897161221210496, 'fraction_numerical': 0.024102838778789503, 'mean_word_length': 3.8193548387096774, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '489504', 'license': 'Apache-2.0', 'n_tokens_mistral': 1522, 'n_tokens_neox': 1459, 'n_words': 272, 'path': 'Section9_SparkDF_Exercise/pyspark_examples.ipynb', 'repo_name': 'bhishanpdl/Spark_and_Python_for_Big_Data_with_PySpark_Udemy', 'size': '3734'}