Apache Spark is an open-source data processing framework. Spark excels at iterative computation and includes numerous libraries for statistical analysis, graph computations, and machine learning. The SingleStore Spark Connector allows you to connect your Spark and SingleStore DB environments. You can use SingleStore DB and Spark together to accelerate workloads by taking advantage of computational power of spark in tandem with the fast ingest and persistent storage SingleStore DB has to offer.

The SingleStore Spark Connector integrates with Apache Spark 2.3 and 2.4 and supports both data loading and extraction from database tables and Spark DataFrames.

The connector is implemented as a native Spark SQL plugin, and supports Spark’s DataSource API. Spark SQL supports operating on a variety of data sources through the DataFrame interface, and the DataFrame API is the widely used framework for how Spark interacts with other systems.

In addition, the connector is a true Spark data source; it integrates with the Catalyst query optimizer, supports robust SQL pushdown, and leverages SingleStore DB LOAD DATA to accelerate ingest from Spark via compression.

You can download the Spark Connector from its GitHub repository and from Maven Central. The group is com.memsql and the artifact is memsql-spark-connector_2.11.

This topic discusses how to configure and start using the SingleStore Spark Connector 3.0.

Note: We’ve made significant changes between the Spark Connector 3.0 and Spark Connector 2.0. Please see the Migrating between the Spark Connector 2.0 and the Spark Connector 3.0 section.

Getting Started

The Spark Connector 3.0 library requires Apache Spark 2.3 or 2.4.

• You need a running cluster running versions 6.8 or higher, and a running Apache Spark environment.

• You can find the latest version of the connector on Maven Central and spark-packages.org. The group is com.memsql and the artifact is memsql-spark-connector_2.11.

• We release two versions of the memsql-spark-connector, one per Spark version. An example version number is: 3.0.0-spark-2.3.4 which is the 3.0.0 version of the connector, compiled and tested against Spark 2.3.4. Make sure you are using the most recent version of the connector.

• In addition to adding the memsql-spark-connector, you will also need to have the MariaDB JDBC driver installed. The driver will be pulled in automatically as a dependency when you import the connector into your library from a package manager. If you plan to install the connector jar manually, you will need to install the dependency separately. This library is tested against the following MariaDB driver version:

  "org.mariadb.jdbc" % "mariadb-java-client"  % "2.+"
  

Installation

You can add the Spark Connector 3.0 your Spark application using Spark-Shell, PySpark, or spark-submit by running the following command. Make sure to update the command with the connector and spark version.

$SPARK_HOME/bin/spark-shell --packages com.memsql:memsql-spark-connector_2.11:3.0.<insert-connector-version>spark-<insert-spark-version>

Alternatively, you may use Maven or SBT.

Configuration Settings

The SingleStore Spark Connector leverages Spark SQL’s Data Sources API. The connection to SingleStore DB relies on the following Spark configuration settings:

The memsql-spark-connector is configurable globally via Spark options and locally when constructing a DataFrame. The global and local options use the same names; however the global options have the prefix spark.datasource.memsql.:

Examples

Example 1: Configuring the memsql-spark-connector globally

spark.conf.set("spark.datasource.memsql.ddlEndpoint", "memsql-master.cluster.internal")
spark.conf.set("spark.datasource.memsql.dmlEndpoints", "memsql-master.cluster.internal,memsql-child-1.cluster.internal:3307")
spark.conf.set("spark.datasource.memsql.user", "admin")
spark.conf.set("spark.datasource.memsql.password", "s3cur3-pa$$word")

SparkConf conf = new SparkConf();
conf.set("spark.datasource.memsql.ddlEndpoint", "memsql-master.cluster.internal")
conf.set("spark.datasource.memsql.dmlEndpoints", "memsql-master.cluster.internal,memsql-child-1.cluster.internal:3307")
conf.set("spark.datasource.memsql.user", "admin")
conf.set("spark.datasource.memsql.password", "s3cur3-pa$$word")

spark.conf.set("spark.datasource.memsql.ddlEndpoint", "memsql-master.cluster.internal")
spark.conf.set("spark.datasource.memsql.dmlEndpoints", "memsql-master.cluster.internal,memsql-child-1.cluster.internal:3307")
spark.conf.set("spark.datasource.memsql.user", "admin")
spark.conf.set("spark.datasource.memsql.password", "s3cur3-pa$$word")

Example 2: Reading a SingleStore DB table (foo) into a DataFrame

val df = spark.read
    .format("memsql")
    .option("ddlEndpoint", "memsql-master.cluster.internal")
    .option("user", "admin")
    .load("foo")

DataFrame df = spark
  .read()
  .format("memsql")
  .option("ddlEndpoint", "memsql-master.cluster.internal")
  .option("user", "admin")
  .load("foo");

df = spark \
  .read \
  .format("memsql") \
  .option("ddlEndpoint", "memsql-master.cluster.internal") \
  .option("user", "admin") \
  .load("foo")

Example 3: Reading a SingleStore DB table (foo) into a DataFrame and applying Data Frame operations

val x = spark.read
    .format("memsql")
    .option("ddlEndpoint")
    .option("user", "admin")
    .load("foo")
    .withColumn("hello", lit(2))
    .filter(col("id") > 1)
    .limit(1000)
    .groupBy(col("id"))
    .agg(count("*"))

Example 4: Configuring the memsql-spark-connector using an external table in Spark SQL and selecting from it

spark.sql("CREATE TABLE bar USING memsql OPTIONS ('ddlEndpoint'='memsql-master.cluster.internal','dbtable'='foo.bar')")

spark.sql("select * from bar limit 10").show()

Example 5: Using the Spark write API to save a Data Frame to SingleStore DB

df.write
    .format("memsql")
    .option("loadDataCompression", "LZ4")
    .option("overwriteBehavior", "dropAndCreate")
    .mode(SaveMode.Overwrite)
    .save("foo.bar") // in format: database.table

Note: This is the same for Java and Python.

If the target table (“foo” in the example above) does not exist in SingleStore DB the memsql-spark-connector will automatically attempt to create the table. If you specify SaveMode.Overwrite, if the target table already exists, it will be recreated or truncated before load. Specify overwriteBehavior = truncate to truncate rather than re-create.

Save Modes

Save operations can optionally take a SaveMode setting (the above example uses Overwrite). Save Modes specify how to handle existing data if present.

It is important to realize that these save modes do not utilize any locking and are not atomic. Additionally, when performing an overwrite, the data will be deleted before writing out the new data.

1. SaveMode.Append means that when saving a DataFrame to a data source, if data/table already exists, contents of the DataFrame are expected to be appended to existing data.

2. SaveMode.Overwrite means that when saving a DataFrame to a data source, if data/table already exists, existing data is expected to be overwritten by the contents of the DataFrame.

   Overwrite mode depends on overwriteBehavior option, for better understanding look at the section Merging On Save.

3. SaveMode.ErrorIfExists means that when saving a DataFrame to a data source, if data already exists, an exception is expected to be thrown.

4. SaveMode.Ignore means that when saving a DataFrame to a data source, if data already exists, the save operation is expected to not save the contents of the DataFrame and to not change the existing data.

Example of SaveMode option

df.write
    .mode(SaveMode.Append)
    .save("foo.bar")

Retrieving the number of written rows from taskMetrics

It is possible to add the listener and get the number of written rows.

spark.sparkContext.addSparkListener(new SparkListener() {
  override def onTaskEnd(taskEnd: SparkListenerTaskEnd) {
    println("Task id: " + taskEnd.taskInfo.id.toString)
    println("Records written: " + taskEnd.taskMetrics.outputMetrics.recordsWritten.toString)
  }
})

df.write.format("memsql").save("example")

Specifying keys for tables created by the Spark Connector

When creating a table, the memsql-spark-connector will read options prefixed with tableKey. These options must be formatted in a specific way in order to correctly specify the keys.

To explain we will refer to the following example:

df.write
    .format("memsql")
    .option("tableKey.primary", "id")
    .option("tableKey.key.created_firstname", "created, firstname")
    .option("tableKey.unique", "username")
    .mode(SaveMode.Overwrite)
    .save("foo.bar") // in format: database.table

In this example, we are creating three keys:

1. A primary key on the id column
2. A regular key on the combination of the firstname and created columns, with the key name created_firstname
3. A unique key on the username column Note on (2): Any key can optionally specify a name, just put it after the key type. Key names must be unique.

To change the default ColumnStore sort key you can specify it explicitly:

df.write
    .option("tableKey.columnstore", "id")

You can also customize the shard key like so.

df.write
    .option("tableKey.shard", "id, lastname")

Inserting rows into the table with ON DUPLICATE KEY UPDATE

When updating a rowstore table it is possible to insert rows via the ON DUPLICATE KEY UPDATE option. See sql reference for more details.

df.write
    .option("onDuplicateKeySQL", "age = age + 1")
    .option("insertBatchSize", 300)
    .mode(SaveMode.Append)
    .save("foo.bar")

As a result of the following query, all new rows will be appended without changes. If a row with the same PRIMARY KEY or UNIQUE index already exists then the corresponding age value will be increased.

When you use ON DUPLICATE KEY UPDATE, all rows of the data frame are split into batches, and every insert query will contain no more than the specified insertBatchSize rows setting.

Using the onDuplicateKeySQL setting to Perform a Partial Update

You can also use the OnDuplicateKeySQL setting to do a partial update on rows matching a primary key. For example, consider a data frame df, with a primary key id that we created and loaded into SingleStore DB from a Spark Dataframe.

df.write
      .format("memsql")
      .option("tableKey.primary", "id")
      .save("test")
spark.read.format("memsql").load("test").show()

Output:

+---+----+------------+
| id|data|dataToUpdate|
+---+----+------------+
|  2|   2|           2|
|  1|   1|           1|
|  3|   3|           3|
+---+----+------------+

In this case, the user only wants to update the column dataToUpdate, but not the column data. Note that if you would like to update both, you should instead use the overWriteBehavior setting and set it to merge.

First, we create a sample data frame that will hold the data which reflects our updates to the table. Note how both columns data and datatoUpdate in this updated data frame are different compared to the originally created object.

val updateDf = spark.createDF(
      List(
        (1, -1, 20),
        (2, -1, 100)
      ),
      List(
        ("id", IntegerType, true),
        ("data", IntegerType, true),
        ("dataToUpdate", IntegerType, true)
      )
)

The following command will only update the dataToUpdate field columns in the SingleStore DB table, not the data field.

updateDf.write
      .format("memsql")
      .option("onDuplicateKeySQL", "dataToUpdate = VALUES(dataToUpdate)")
      .mode(SaveMode.Append)
      .save("test")
spark.read.format("memsql").load("test").show()

The final output of the partial update looks like this:

+---+----+------------+
| id|data|dataToUpdate|
+---+----+------------+
|  2|   2|         100|
|  1|   1|          20|
|  3|   3|           3|
+---+----+------------+

Merging on save

When saving dataframes or datasets to SingleStore DB, you can manage how SaveMode.Overwrite is interpreted by the connector via the option overwriteBehavior. This option can take one of the following values:

1. dropAndCreate(default) - drop and create the table before writing new values.
2. truncate - truncate the table before writing new values.
3. merge - replace rows with new rows by matching on the primary key. (Use this option only if you need to fully rewrite existing rows with new ones. If you need to specify some rule for the update, use the onDuplicateKeySQL option instead.)

All these options are case-insensitive.

Example of merge option

Suppose you have the following table t, and the Id column is the primary key.

SELECT * FROM t;

If you save the following dataframe with overwriteBehavior = merge:

df.write
    .format("memsql")
    .option("overwriteBehavior", "merge")
    .mode(SaveMode.Overwrite)
    .save("db.t")

After the save is complete, the table will look like this:

note: rows with Id=2 and Id=3 were overwritten with new rows
note: the row with Id=1 was not touched and still exists in the result

SELECT * FROM t;

Supported Operations

For all supported operations in Spark, refer to Spark’s data source APIs.

SQL Pushdown

The memsql-spark-connector has extensive support for rewriting Spark SQL and dataframe operation query plans into standalone SingleStore DB queries. This allows most of the computation to be pushed into the SingleStore DB distributed system without any manual intervention. The SQL rewrites are enabled automatically but can be disabled using the disablePushdown option. We also support partial pushdown in the case where certain parts of a query can be evaluated in SingleStore DB and certain parts need to be evaluated in Spark.

SQL Pushdown is either enabled or disabled on the entire Spark Session. If you want to run multiple queries in parallel with different values of disablePushdown, make sure to run them on separate Spark Sessions.

We currently support most of the primary Logical Plan nodes in Spark SQL including: Project, Filter, Aggregate, Window, Join, Limit, Sort.

We also support most Spark SQL expressions. A full list of supported operators/functions can be found in the file ExpressionGen.scala.

The best place to look for examples of fully supported queries is in the tests. Check out this file as a starting point: SQLPushdownTest.scala.

SQL Pushdown Incompatibilities

• ToUnixTimestamp and UnixTimestamp handle only time less then 2038-01-19 03:14:08, if they get DateType or TimestampType as a first argument
• FromUnixTime with default format (yyyy-MM-dd HH:mm:ss) handle only time less then 2147483648 (2^31)

Parallel Read Support

If you enable parallel reads via the enableParallelRead option, the memsql-spark-connector will attempt to read results directly from leaf nodes. This can drastically improve performance in some cases.

Note: Parallel reads are not consistent

Parallel reads read directly from partitions on the leaf nodes which skips our entire transaction layer. This means that the individual reads will see an independent version of the databases distributed state. Make sure to take this into account when enabling parallel read.

Note: Parallel reads transparently fallback to single stream reads

Parallel reads currently only work for query-shapes which do no work on the Aggregator and thus can be pushed entirely down to the leaf nodes. To determine if a particular query is being pushed down you can ask the dataframe how many partitions it has like so:

df.rdd.getNumPartitions

If this value is > 1 then we are reading in parallel from leaf nodes.

Note: Parallel reads require consistent authentication and connectible leaf nodes

In order to use parallel reads, the username and password provided to the memsql-spark-connector must be the same across all nodes in the cluster.

In addition, the hostnames and ports listed by SHOW LEAVES must be directly connectible from Spark.

Load Balancing Requests

DML and DDL endpoints in the configuration refer to your child aggregators and your master aggregator, respectively. Use the DDL endpoint (master aggregator) for running DDL operations (e.g. CREATE TABLE). Use DML endpoints (child aggregators) to read data and perform other DML operations. When you provide both endpoints to the connector, JDBC will automatically load balance non-DDL queries across your aggregators.

Data Type Conversions

When saving a DataFrame to SingleStore DB, the DataFrame column will be converted to the following SingleStore DB type:

When reading a SingleStore DB table as a Spark DataFrame, the SingleStore DB column type will be converted to the following SparkType:

Data Type Conversion Remarks

• When using the onDuplicateKeySQL option, the connector will error when writing a null-terminated StringType (i.e., ‘\0’).
• DECIMAL in SingleStore DB and DecimalType in Spark have different maximum scales and precisions. An error will occur if you perform a read or write from a table or dataFrame with unsupported precision or scale. SingleStore DB’s maximum scale for a type DECIMAL is 30, while Spark’s maximum scale for DecimalType is 38. Similarly, SingleStore DB’s maximum precision is 65, while Spark’s maximum precision is 38.
• TIMESTAMP in SingleStore DB supports values from 1000 to 2147483647999. SingleStore DB treats a null value in the TIMESTAMP column as the current time.
• The Avro format does not support writing of TIMESTAMP and DATE types. As a result, the SingleStore Spark Connector currently does not support these types with Avro serialization.

Security and Permissions

SQL Permissions

The spark user must have access to the Master Aggregator.

Additionally, SingleStore DB has a permission matrix which describes the permissions required to run each command.

To make any SQL operations through Spark connector you should have different permissions for different type of operation. The matrix below describes the minimum permissions you should have to perform some operation. As alternative to minimum required permissions, ALL PRIVILEGES allow you to perform any operation.

SSL Support

The SingleStore Spark Connector uses the MariaDB JDBC Driver under the hood and thus supports SSL configuration out of the box. In order to configure SSL, first ensure that your cluster has SSL configured. Documentation on how to set this up can be found here:

SSL Configuration

Once you have setup SSL on your server, you can enable SSL via setting the following options:

spark.conf.set("spark.datasource.memsql.useSSL", "true")
spark.conf.set("spark.datasource.memsql.serverSslCert", "PATH/TO/CERT")

Note: the serverSslCert option may be server’s certificate in DER form, or the server’s CA certificate. Can be used in one of 3 forms:

• serverSslCert=/path/to/cert.pem (full path to certificate)
• serverSslCert=classpath:relative/cert.pem (relative to current classpath)
• or as verbatim DER-encoded certificate string ------BEGIN CERTIFICATE-----...

You may also want to set these additional options depending on your SSL configuration:

spark.conf.set("spark.datasource.memsql.trustServerCertificate", "true")
spark.conf.set("spark.datasource.memsql.disableSslHostnameVerification", "true")

More information on the above parameters can be found at MariaDB’s documentation for their JDBC driver here: https://mariadb.com/kb/en/about-mariadb-connector-j/#tls-parameters

Connecting with a Kerberos-authenticated User

You can use the SingleStore Spark Connector with a Kerberized user without any additional configuration. To use a Kerberized user, you need to configure the connector with the given SingleStore DB database user that is authenticated with Kerberos (via the user option). Please visit our documentation here to learn about how to configure SingleStore DB users with Kerberos.

Here is an example of configuring the Spark connector globally with a Kerberized SingleStore DB user called krb_user.

spark = SparkSession.builder()
    .config(“spark.datasource.memsql.user”, “krb_user”)
    .getOrCreate()

You do not need to provide a password when configuring a Spark Connector user that is Kerberized. The connector driver (MariaDB) will be able to authenticate the Kerberos user from the cache by the provided username. Other than omitting a password with this configuration, using a Kerberized user with the Connector is no different than using a standard user. Note that if you do provide a password, it will be ignored.

Debugging

SQL Pushdown

If you want to know whether your queries are pushed down, df.explain() can show what parts of the query were pushed down and can also be used for debugging if you encounter an issue. Additionally, if you pass the argument true, you will get a lot more output that includes pre- and post-optimization passes.

Other

In addition, the memsql-spark-connector outputs a lot of helpful information when the TRACE log level is enabled for the com.memsql.spark package. You can do this in your log4j configuration by adding the following line: log4j.logger.com.memsql.spark=TRACE Make sure not to leave it in place since it generates a huge amount of tracing output.

Migrating between the Spark Connector 2.0 and the Spark Connector 3.0

You may have previously used the SingleStore Spark Connector 2.0. There are many enhancements between the two versions, and the sections below describe the differences in configuration and functionality between the SingleStore Spark Connector versions 3.0 and 2.0.

Configuration Differences

• If you are only using the Spark reader API, using version 3.0 of the connector will only require a change in the configuration options. Please see below for details.
• If you are using the previous Spark connector to write data to SingleStore DB, the existing savetoMemSQL function is deprecated and has been replaced with Spark’s df.write and .save() functions.

Configuration Comparisons

Functionality Differences

Version 2.0 of the SingleStore Spark Connector contains the following functionality that is not available in version 3.0:

• The SaveToMemSQL() function to write to SingleStore DB; this is replaced with using df.write directly
• Adding indexes to automatically created tables; this can be done via a JDBC query
• No formal Spark Streaming API integration.

Was this Article Helpful?

Thanks For Your Feedback!