DEV Community

SingleStore profile image Akmal Chaudhri
Akmal Chaudhri for SingleStore

Posted on

Quick tip: Using SingleStore Spark Connector's Query Pushdown with SingleStore Notebooks

#singlestoredb #apachespark #querypushdown

Abstract

The Singlestore Spark Connector supports the rewriting of Spark query execution plans into SingleStore queries, for both SQL and DataFrame operations. Computation is pushed into the SingleStore system automatically. In this short article, we'll look at an example of query pushdown.

The notebook file used in this article is available on GitHub.

Create a SingleStore Cloud account

A previous article showed the steps to create a free SingleStore Cloud account. We'll use the following settings:

  • Workspace Group Name: Spark Demo Group
  • Cloud Provider: AWS
  • Region: US East 1 (N. Virginia)
  • Workspace Name: spark-demo
  • Size: S-00

Create a new notebook

From the left navigation pane in the cloud portal, we'll select Develop > Notebooks.

In the top right of the web page, we'll select New Notebook > New Notebook, as shown in Figure 1.

Figure 1. New Notebook.

Figure 1. New Notebook.

We'll call the notebook spark_pushdown_demo, select a Blank notebook template from the available options, and save it in the Personal location.

Fill out the notebook

First, let's install Spark:

!pip cache purge --quiet
!conda install -y --quiet -c conda-forge openjdk pyspark
Enter fullscreen mode Exit fullscreen mode

Next, we'll create a directory to store some jar files:

import os

os.makedirs("jars", exist_ok = True)
Enter fullscreen mode Exit fullscreen mode

We'll now download some jar files, as follows:

import requests

def download_jar(url, destination):
    response = requests.get(url)
    with open(destination, "wb") as f:
        f.write(response.content)

jar_urls = [
    ("https://repo1.maven.org/maven2/com/singlestore/singlestore-jdbc-client/1.2.1/singlestore-jdbc-client-1.2.1.jar", "jars/singlestore-jdbc-client-1.2.1.jar"),
    ("https://repo1.maven.org/maven2/com/singlestore/singlestore-spark-connector_2.12/4.1.5-spark-3.5.0/singlestore-spark-connector_2.12-4.1.5-spark-3.5.0.jar", "jars/singlestore-spark-connector_2.12-4.1.5-spark-3.5.0.jar"),
    ("https://repo1.maven.org/maven2/org/apache/commons/commons-dbcp2/2.12.0/commons-dbcp2-2.12.0.jar", "jars/commons-dbcp2-2.12.0.jar"),
    ("https://repo1.maven.org/maven2/org/apache/commons/commons-pool2/2.12.0/commons-pool2-2.12.0.jar", "jars/commons-pool2-2.12.0.jar"),
    ("https://repo1.maven.org/maven2/io/spray/spray-json_3/1.3.6/spray-json_3-1.3.6.jar", "jars/spray-json_3-1.3.6.jar")
]

for url, destination in jar_urls:
    download_jar(url, destination)

print("JAR files downloaded successfully")
Enter fullscreen mode Exit fullscreen mode

These jar files include the SingleStore JDBC Client and the SingleStore Spark Connector, as well as several other jar files needed for connectivity and data management.

Now we are ready to create a SparkSession:

from pyspark.sql import SparkSession

# Create a Spark session
spark = (SparkSession
             .builder
             .config("spark.jars", ",".join([destination for _, destination in jar_urls]))
             .appName("Spark Pushdown Test")
             .getOrCreate()
        )

spark.sparkContext.setLogLevel("ERROR")
Enter fullscreen mode Exit fullscreen mode

Next, we'll obtain a weather dataset and load it using Pandas. The data are from the CORGIS Dataset Project on GitHub. The weather.csv file contains weather data for cities across the United States for 2016.

import pandas as pd

url = "https://raw.githubusercontent.com/corgis-edu/corgis/master/website/datasets/csv/weather/weather.csv"

pandas_df = pd.read_csv(url)
Enter fullscreen mode Exit fullscreen mode

We can check the first few rows:

pandas_df.head()
Enter fullscreen mode Exit fullscreen mode

Next, we'll define a Spark DataFrame schema:

from pyspark.sql.types import *

schema = StructType([
    StructField("Precipitation", FloatType(), True),
    StructField("Date", StringType(), True),
    StructField("Month", IntegerType(), True),
    StructField("Week", IntegerType(), True),
    StructField("Year", IntegerType(), True),
    StructField("City", StringType(), True),
    StructField("Code", StringType(), True),
    StructField("Location", StringType(), True),
    StructField("State", StringType(), True),
    StructField("Avg", IntegerType(), True),
    StructField("Max", IntegerType(), True),
    StructField("Min", IntegerType(), True),
    StructField("Wind_Direction", IntegerType(), True),
    StructField("Wind_Speed", FloatType(), True)
])
Enter fullscreen mode Exit fullscreen mode

This schema shortens some of the original column names in the CSV file.

We'll now create a Spark DataFrame using the Pandas data and the schema defined above:

spark_df = spark.createDataFrame(pandas_df, schema)
Enter fullscreen mode Exit fullscreen mode

We'll now check the number of rows:

spark_df.count()
Enter fullscreen mode Exit fullscreen mode

The output should be:

16743
Enter fullscreen mode Exit fullscreen mode

Now we'll use Plotly Express to create a line chart:

import plotly.express as px

def plot_data(df, x_col, y_cols, title):
    df = df.orderBy(x_col)
    fig = px.line(
        df.toPandas(),
        x = x_col,
        y = y_cols, 
        title = title
    )
    fig.show()
Enter fullscreen mode Exit fullscreen mode

and create a plot using the Spark DataFrame:

plot_data(
    spark_df.filter(spark_df["City"] == "San Francisco"),
    "Date",
    ["Max", "Min"],
    "Max and Min Temperatures in San Francisco (Fahrenheit)"
)
Enter fullscreen mode Exit fullscreen mode

The output should be as shown in Figure 2.

Figure 2. Max and Min in Fahrenheit for San Francisco.

Figure 2. Max and Min in Fahrenheit for San Francisco.

Now we'll create the connection details to SingleStore:

host = "<HOST>"
password = "<PASSWORD>"
port = "3306"
cluster = host + ":" + port
Enter fullscreen mode Exit fullscreen mode

Replace <HOST> and <PASSWORD> with the values for your environment. These values can be obtained from the workspace using Connect > SQL IDE.

We also need to set some configuration parameters:

spark.conf.set("spark.datasource.singlestore.ddlEndpoint", cluster)
spark.conf.set("spark.datasource.singlestore.user", "admin")
spark.conf.set("spark.datasource.singlestore.password", password)
spark.conf.set("spark.datasource.singlestore.disablePushdown", "false")
Enter fullscreen mode Exit fullscreen mode

A database is required, so we'll create one:

DROP DATABASE IF EXISTS spark_demo;
CREATE DATABASE IF NOT EXISTS spark_demo;
Enter fullscreen mode Exit fullscreen mode

We'll now write the DataFrame:

(spark_df.write
    .format("singlestore")
    .option("loadDataCompression", "LZ4")
    .mode("overwrite")
    .save("spark_demo.weather")
)
Enter fullscreen mode Exit fullscreen mode

In this case, the weather table will be created for us.

Next, we'll read the data back into a new DataFrame:

new_df = (spark.read
    .format("singlestore")
    .schema(schema)
    .load("spark_demo.weather")
)
Enter fullscreen mode Exit fullscreen mode

Next, we'll create a temporary Spark table, as follows:

new_df.createOrReplaceTempView("temperatures")
Enter fullscreen mode Exit fullscreen mode

We'll now create and register a Python UDF to convert temperatures from Fahrenheit to Celsius:

def convert_to_c(f):
    c = (f - 32) * (5 / 9)
    return int(round(c))

spark.udf.register("convert_to_c", convert_to_c, IntegerType())
Enter fullscreen mode Exit fullscreen mode

Now we'll formulate a query that uses the Python UDF:

temp_df = spark.sql(
    "SELECT Date, convert_to_c(Max) as Max_C, convert_to_c(Min) as Min_C FROM temperatures WHERE City = 'San Francisco'"
)
Enter fullscreen mode Exit fullscreen mode

Adding .explain() to the end of the query shows us the Physical Query Plan:

temp_df.explain()
Enter fullscreen mode Exit fullscreen mode

The output should look similar to the following:

== Physical Plan ==
*(2) Project [Date#91, pythonUDF0#125 AS Max_C#118, pythonUDF1#126 AS Min_C#119]
+- BatchEvalPython [convert_to_c(Max#100)#120, convert_to_c(Min#101)#121], [pythonUDF0#125, pythonUDF1#126]
   +- *(1) Scan 
---------------
SingleStore Query
Variables: (San Francisco)
SQL:
SELECT `Date#1` , `Max#4` , `Min#5` 
FROM (

  SELECT `Date#1` , `Max#4` , `Min#5` 
  FROM (

    SELECT * 
    FROM (
      SELECT ( `Precipitation` ) AS `Precipitat#8` , ( `Date` ) AS `Date#1` , ( `Month` ) AS `Month#9` , ( `Week` ) AS `Week#10` , ( `Year` ) AS `Year#11` , ( `City` ) AS `City#12` , ( `Code` ) AS `Code#13` , ( `Location` ) AS `Location#14` , ( `State` ) AS `State#15` , ( `Avg` ) AS `Avg#16` , ( `Max` ) AS `Max#4` , ( `Min` ) AS `Min#5` , ( `Wind_Direction` ) AS `Wind_Direc#17` , ( `Wind_Speed` ) AS `Wind_Speed#18` 
      FROM (
        SELECT * FROM `spark_demo`.`weather`
      ) AS `a2`
    ) AS `a3` 
    WHERE ( ( `City#12` = ? ) AND ( `City#12` ) IS NOT NULL )
  ) AS `a4`
) AS `a5`

EXPLAIN:
Gather partitions:all alias:remote_0 parallelism_level:segment
Project [a5.Date AS `Date#1`, a5.Max AS `Max#4`, a5.Min AS `Min#5`]
ColumnStoreFilter [a5.City = 'San Francisco' AND a5.City IS NOT NULL]
ColumnStoreScan spark_demo.weather AS a5, SORT KEY Precipitation (Precipitation) table_type:sharded_columnstore
---------------
       [Date#91,Max#100,Min#101] PushedFilters: [], ReadSchema: struct<Date:string,Max:int,Min:int>
Enter fullscreen mode Exit fullscreen mode

The final plan shows a single projection on top of a scan. We can also see query pushdown to SingleStore.

Finally, we can plot the data:

plot_data(
    temp_df,
    "Date",
    ["Max_C", "Min_C"],
    "Max and Min Temperatures in San Francisco (Celsius)"
)
Enter fullscreen mode Exit fullscreen mode

The output should be as shown in Figure 3.

Figure 3. Max and Min in Celsius for San Francisco.

Figure 3. Max and Min in Celsius for San Francisco.

Finally, we'll stop the SparkSession:

# Stop the Spark session
spark.stop()
Enter fullscreen mode Exit fullscreen mode

Summary

In this example, we've seen the ease with which we can use the SingleStore Spark Connector for query pushdown.

The key benefits of the SingleStore Spark Connector can be summarised as follows:

  • Implemented as a native Spark SQL plugin
  • Accelerates ingest from Spark via compression
  • Supports data loading and extraction from database tables and Spark DataFrames
  • Integrates with the Catalyst query optimiser and supports robust SQL pushdown
  • Accelerates ML workloads

Top comments (0)

Read next

mario130 profile image

Common Pitfalls When Using React Hooks

Mario Yonan -

vishalmx3 profile image

Day 30/366

vishal.codes -

kembec profile image

Unlock Scalability: Master CQRS for Cutting-Edge Software!

Manuel Benancio -

kumarsanu profile image

React

Kumar Sanu -