Examples

The following example shows how to configure singlestore-spark-connector for SingleStoreDB.

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

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

The following example reads a SingleStoreDB table named foo into a DataFrame.

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

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

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

The following example reads a SingleStoreDB table named foo into a DataFrame and applies DataFrame transformations:

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

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

spark.sql("SELECT * FROM bar LIMIT 10").show()

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

If the target table (foo in this example) does not exist in SingleStoreDB, the singlestore-spark-connector automatically attempts to create the table. If you specify SaveMode.Overwrite and if the target table already exists, it will be recreated or truncated before load. Specify overwriteBehavior = truncate to truncate rather than re-create.

This example requires global options like ddlEndpoint, user, and password to be set. Note that if you specify the dbtable option, there's no need to specify the query option, and vice versa.

Here's an example of how to use dbtable:

df = spark.read\    
.format("singlestore")\     
.option("dbtable", tble_name).load()

This example requires global options like ddlEndpoint, user, and password to be set. Note that if you specify the dbtable option, there's no need to specify the query option, and vice versa.

When using the query option, you must specify the database option:

df = spark.read.format("singlestore")
   .option("database", "database_name")
   .option("query", "SELECT * FROM table_name").load()

Save operations can optionally take a SaveMode setting. Save modes specify how to handle existing data.

It is important to realize that these save modes do not utilize any locking and are not atomic.

Here's an example:

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

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

All these options are case-insensitive. Here's an example:

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("singlestore")
    .option("overwriteBehavior", "merge")
    .mode(SaveMode.Overwrite)
    .save("db.t")

After the save is complete, the table looks like this:

Note: Rows with Id=2 and Id=3 are overwritten with new rows. The row with Id=1 is not touched and still exists in the result.

SELECT * FROM t;

The following example shows how to add a 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("singlestore").save("example")

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

Here's an example:

df.write
    .format("singlestore")
    .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 create three keys:

Any key can optionally specify a name, just specify 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, for example:

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

When updating a rowstore table, it is possible to insert rows via the ON DUPLICATE KEY UPDATE option. For example:

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

As a result of this query, all new rows are 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 DataFrame are split into batches, and every insert query can have a maximum of insertBatchSize rows.

df.write
      .format("singlestore")
      .option("tableKey.primary", "id")
      .save("test")
spark.read.format("singlestore").load("test").show()
****
+---+----+------------+
| id|data|dataToUpdate|
+---+----+------------+
|  2|   2|           2|
|  1|   1|           1|
|  3|   3|           3|
+---+----+------------+

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

updateDf.write
      .format("singlestore")
      .option("onDuplicateKeySQL", "dataToUpdate = VALUES(dataToUpdate)")
      .mode(SaveMode.Append)
      .save("test")
spark.read.format("singlestore").load("test").show()
****
+---+----+------------+
| id|data|dataToUpdate|
+---+----+------------+
|  2|   2|         100|
|  1|   1|          20|
|  3|   3|           3|
+---+----+------------+