A Spark SQL server based on the PostgreSQL V3 protocol. For more information, see SPARK-15816. This Spark SQL server experimentally supports impersonation based on Apache Livy that the Spark Thrift server currently doesn't.
Note that the current master branch intends to support 3.0.0-preview2 on Scala 2.12.x. If you want to use the SQL server in Spark 2.4.x, please use branch-2.4.
To start the JDBC/ODBC server, you need to check out this repository and run the following command in the root directory:
$ ./sbin/start-sql-server.sh --conf spark.sql.server.psql.enabled=true
This script accepts all bin/spark-submit command line options in Spark, plus options for the SQL server.
You may run ./sbin/start-sql-server.sh --help for a complete list of all available options.
If you use spark-2.4.x, you can add the option below to install the JDBC/ODBC server:
$ ./bin/spark-shell --packages maropu:spark-sql-server:0.1.7-spark2.4
Then, you run the commands to launch the server:
scala> sql("SET spark.sql.server.psql.enabled=true")
scala> sql("SET spark.sql.crossJoin.enabled=true")
scala> org.apache.spark.sql.server.SQLServer.startWithContext(spark.sqlContext)
Now, you can use a PostgreSQL psql command to test the SQL JDBC/ODBC server:
$ psql -h localhost -d default
If you have no psql command and you use the Amazon Linux AMI, you can run sudo yum install postgresql95
to install PostgreSQL client programs. Since the SSL mode of this psql command is enabled by default,
you need to turn off the SSL mode to connect the SQL server:
$ psql postgresql://localhost:5432/default?sslmode=disable
To connect the SQL server, you can use mature and widely-used PostgreSQL JDBC drivers. You can get the driver, add it to a class path, and write code like;
import java.sql.*;
import java.util.Properties;
public class JdbcTest {
public static void main(String[] args) {
try {
// Register your PostgreSQL JDBC driver
Class.forName("org.postgresql.Driver");
// Connect to a 'default' database in the SPARK SQL server
String dbName = "default";
Properties props = new Properties();
props.put("user", "maropu");
Connection con = DriverManager.getConnection("jdbc:postgresql://localhost:5432/" + dbName, props);
// Do something...
Statement stmt = con.createStatement();
stmt.executeQuery("CREATE TEMPORARY VIEW t AS SELECT * FROM VALUES (1, 1), (1, 2) AS t(a, b)").close();
ResultSet rs = stmt.executeQuery("SELECT * FROM t");
while (rs.next()) {
System.out.println("a=" + rs.getInt("a") + " b=" + rs.getInt("b"));
}
rs.close();
stmt.close();
con.close();
} catch (Exception e) {
// Do error handling here...
}
}
}This Spark SQL server is only tested by using v42.2.2 of PostgreSQL JDBC drivers.
You can also use libpq
to connect the SQL server from C clients:
#include <stdio.h>
#include <stdlib.h>
#include <libpq-fe.h>
static void exit_nicely(PGconn *conn) {
PQfinish(conn);
exit(1);
}
int main(int argc, char **argv) {
// Connect to a 'default' database in the SPARK SQL server
PGconn *conn = PQconnectdb("host=localhost port=5432 dbname=default");
if (PQstatus(conn) != CONNECTION_OK) {
exit_nicely(conn);
}
// Do something...
PGresult *res = PQexec(conn, "SELECT * FROM VALUES (1, 1), (1, 2) AS t(a, b)");
if (PQresultStatus(res) != PGRES_TUPLES_OK) {
PQclear(res);
exit_nicely(conn);
}
for (int i = 0; i < PQntuples(res); i++) {
printf("a=%s b=%s\n", PQgetvalue(res, i, 0), PQgetvalue(res, i, 1));
}
PQclear(res);
PQfinish(conn);
return 0;
}
You can directly load SQL result data as Pandas DataFrame by using some libraries (e.g., psycopg2 and pg8000);
>>> import psycopg2
>>> import pandas as pd
>>> connection = psycopg2.connect("host=localhost port=5432 dbname=default user=maropu sslmode=disable")
>>> df = pd.read_sql(sql="SELECT * FROM VALUES (1, 1), (1, 2) AS t(a, b);", con=connection)
>>> df
a b
0 1 1
1 1 2
Note that you need to set false at spark.sql.server.binaryTransferMode for the psycopg2 library.
$ ./sbin/start-sql-server.sh -h
Usage: ./sbin/start-sql-server.sh [options] [SQL server options]
SQL server options:
--conf spark.sql.server.port=NUM Port number of SQL server interface (Default: 5432).
--conf spark.sql.server.executionMode=STR Execution mode: single-session, multi-session, or multi-context (Default: multi-session).
--conf spark.sql.server.worker.threads=NUM # of SQLServer worker threads (Default: 4).
--conf spark.sql.server.binaryTransferMode=BOOL Whether binary transfer mode is enabled (Default: true).
--conf spark.sql.server.ssl.enabled=BOOL Enable SSL encryption (Default: false).
--conf spark.sql.server.ssl.path=STR Keystore path.
--conf spark.sql.server.ssl.keystore.passwd=STR Keystore password.
--conf spark.sql.server.ssl.certificate.passwd=STR Certificate password.
--conf spark.yarn.keytab=STR Keytab file location.
--conf spark.yarn.principal=STR Principal name in a secure cluster.
--conf spark.yarn.impersonation.enabled=BOOL Whether authentication impersonates connected users (Default: false).
You can select one of thee different isolation levels: single-session, multi-session(default), and multi-context(experimental).
In the single-session mode, all the sessions in the SQL server share a single SparkSession.
In the multi-session mode, each session has an independent SparkSession with isolated SQL configurations, temporary tables,
and registered functions, but shares an underlying SparkContext and cached data.
In the multi-context mode, each session has the SparkSession that
Apache Livy launches on an independent JVM.
You can also use some GUI clients (e.g., PGnJ);
Note that you need to set false at spark.sql.server.binaryTransferMode for the PGnJ client.
To enable a cursor mode on your JDBC driver, you make sure autocommit is off and you need to set fetch size
throught Statement.setFetchSize (See descriptions in Chapter 5. Issuing a Query and Processing the Result);
// Make sure autocommit is off
Connection con = DriverManager.getConnection("jdbc:postgresql://localhost:5432/" + dbName, props);
con.setAutoCommit(false);
// Turn use of the cursor on.
Statement stmt = con.createStatement();
stmt.setFetchSize(50);
ResultSet rs = stmt.executeQuery("SELECT * FROM range(10000000)");
while (rs.next()) {
System.out.println("id=" + rs.getLong("id"));
}Also, you could set spark.sql.server.incrementalCollect.enabled for memory efficiency
when launching the SQL server. If enabled, the SQL server collects result data partition-by-parititon.
Spark has already implmented a lot of efficient optimization rules in Catalyst and the Spark community continues to develop it now. In some usecases, domain-specific knowledge and rules could make Spark more efficient and so it is useful to append user-defined optimizer rules in Catalyst. Spark has an injection point in SparkSessionExtensions for that purpose. To install them, you need to pass a builder class to inject rules in configurations below;
$ ./sbin/start-sql-server.sh \
--conf spark.jars=./assembly/extensions_2.11_2.4.0_0.1.7-spark2.4-SNAPSHOT.jar \
--conf spark.sql.server.extensions.builder=org.apache.spark.ExtensionBuilderExample
To enable SSL encryption, you need to set the following configurations in start-sql-server.sh;
$ ./sbin/start-sql-server.sh \
--conf spark.sql.server.ssl.enabled=true \
--conf spark.sql.server.ssl.keystore.path=<your keystore path> \
--conf spark.sql.server.ssl.keystore.passwd=<your keystore password>
If you use self-signed certificates, you follow 3 steps below to create self-signed SSL certificates;
// Create the self signed certificate and add it to a keystore file
$ keytool -genkey -alias ssltest -keyalg RSA -keystore server.keystore -keysize 2048
// Export this certificate from server.keystore to a certificate file
$ keytool -export -alias ssltest -file ssltest.crt -keystore server.keystore
// Add this certificate to the client's truststore to establish trust
$ keytool -import -trustcacerts -alias ssltest -file ssltest.crt -keystore client.truststore
You set the generated server.keystore to spark.sql.server.ssl.keystore.path and add a new entry (ssl=true) in Properties
when creating a JDBC connection. Then, you pass client.truststore when running JdbcTest
(See the PostgreSQL JDBC driver documentation for more information);
$ javac JdbcTest.java
$ java -Djavax.net.ssl.trustStore=client.truststore -Djavax.net.ssl.trustStorePassword=<password> JdbcTest
You can use the SQL server on a Kerberos-secure cluster.
To enable this, you need to set the following configurations in start-sql-server.sh;
$ ./sbin/start-sql-server.sh \
--conf spark.yarn.principal=<Kerberos principal> \
--conf spark.yarn.keytab=<keytab location>
Then, you set a Kerberos identity (kerberosServerName) in Properties to connect the SQL server when creating a JDBC connection.
See Connection Parameters for more information.
Moreover, you can enable impersonation for Apache Hadoop by setting true at spark.yarn.impersonation.enabled.
See a tutorial for quick try.
You first need to generate test data for TPC-DS queries:
$ git clone https://github.com/maropu/spark-tpcds-datagen.git
$ ./bin/dsdgen --output-location /tmp/spark-tpcds-data
Then, launches the SQL server with a Spark standalone mode:
$ ./sbin/start-sql-server.sh \
--conf spark.master=local[*] \
--conf spark.driver.extraJavaOptions=-XX:+UseG1GC \
--conf spark.driver.memory=8g
Finally, runs TPC-DS queries against the SQL server:
$ ./bin/run-tpcds-benchmark --data-location /tmp/spark-tpcds-data
This benchmark code is a good example about how to connect the SQL server with Postgre JDBC drivers.
If you hit some bugs and requests, please leave some comments on Issues or Twitter(@maropu).