PySpark example with parquet

4.1. Explore possible partitions

It is a good idea to store dataframe in parquet by partitions. Usually they are partitioned by date. But what is important is that you partition by a column that you will usually filter or aggregate.

5.1. Explore one station

First let’s explore the table with the definitions of the stations. Since I have a home at a small town called Ulldemolins let’s try to find it.

sdf_stations = timeit(spark.read.csv)(c.URI_STATIONS, header=True)
sdf_stations[sdf_stations["NOM_ESTACIO"].startswith("Ull")].toPandas()

This gives 5 results. Two of them are from Ulldemolins, one station that was closed on 2008 and the new one with the code XD.

So let’s join the tables with the data and the one with the variable definitions.

sdf.createOrReplaceTempView("data")

sdf_vars = timeit(spark.read.csv)(c.URI_VARS, header=True)
sdf_vars.createOrReplaceTempView("vars")

ulldemolins = spark.sql("""
    SELECT d.DATA_LECTURA as datetime, d.VALOR_LECTURA as value, ACRONIM as acr, NOM_VARIABLE as name
    FROM data d
    LEFT JOIN vars v
    ON d.CODI_VARIABLE = v.CODI_VARIABLE
    WHERE CODI_ESTACIO = "XD"
""")
ulldemolins.createOrReplaceTempView("ulldemolins")

Now if we do a count of this new table with ulldemolins.count() we can see that is really fast (0.59 s).

The thing is that spark is only accessing one partition so it does not need to load a lot of data. It is only working with 2.8 million rows which is only 0.78% of the original data (360 million rows).

5.2. Explore temperatures of one station

Let’s keep only max temperatures (encoded as Tx) and min temperatures (as Tn). To filter it you can:

sdf = ulldemolins[ulldemolins["acr"].isin(["Tx", "Tn"])]

The result has only 356255 rows. So it is safe to transform to pandas to continue the exploration. However let’s work with spark to practice a little bit more.

First let’s extract the date from the datetime. To do so we need to craete a udf and then apply it:

@udf(StringType())
def get_day(x):
    """ Parse date and extract day """
    return pd.to_datetime(x).strftime("%Y-%m-%d")

sdf = sdf.withColumn("date", get_day(sdf["datetime"]))

Finally we can extract average min and max temperatures for each day. This will make the table even smaller.

df = sdf.groupBy(["date", "acr", "name"]).agg({"value": "mean"}).toPandas()
df = df.sort_values("date")

Now we can create a plot with the min and max temperatures.

Some thoughts about the plot:

• It seems that 2019 data is not good (missing values).
• If you zoom in you will see that there is some kind of pattern for the days.

Let’s check it out.

df["day"] = df["date"].str[-2:]
df["month"] = df["date"].str[5:-3]

df_avg = df[df["acr"] == "Tn"].pivot_table(
    values="avg(value)",
    index="month",
    columns="day"
)

@udf(StringType())
def get_day(x):
    """ Parse date and extract day """
    return pd.to_datetime(x, dayfirst=True).strftime("%Y-%m-%d")