org.mule.modules.sparksql

Class SparkSqlConnector

java.lang.Object

org.mule.modules.sparksql.SparkSqlConnector

@RequiresEnterpriseLicense(allowEval=true)
 @MetaDataScope(value=DataSenseResolver.class)
 @Connector(name="spark-sql",
           friendlyName="SparkSql",
           minMuleVersion="3.6")
public class SparkSqlConnector
extends Object

Annypoint connector SparkSql enables Apache Spark SQL to be used in Mule by providing integration inside Mule flows. The main functionality the Spark SQL Connector is to allow the execution of Spark job to extract structured data using Spark SQL capabilities. Version 1.0.0 allows a user to submit a job (defined as a SQL Query) into a Spark standalone Cluster and retrieve the results as a collection of entities. A use case can be defining a context with sources like a JSON file at hadoop, a cassandra DB and a PostgreSQL DB and execute a SQL Query (a SparkSQL Job) that joins and filters those data sources and produces, as result, a list of objects with requested data.

Author:

mnemo

Constructor Summary

Constructors

Constructor and Description
SparkSqlConnector()

Method Summary

Modifier and Type	Method and Description
void	addCassandraDataProperties(List<CassandraDataProperties> listCassandraDataProps) Dynamically add new datasources from Cassandra database keyspaces and tables
void	addFileDataProperties(List<FileDataProperties> listFileDataProps) Dynamically add new datasources from Hadoop HDFS files
void	addJdbcDataProperties(List<JdbcDataProperties> listJdbcDataProps) Dynamically add new datasources from JDBC accesible databases
Object	customSql(String sql) Run an arbitraty SQL query through Spark SQL Context.
void	disconnect() Force global configuration disconnection.
void	dropTemporaryTable(String temporaryTableName) Drop a temporary table present at the Spark SQL Context.
org.apache.spark.sql.DataFrame	executeSql(String sql) Execute sql in sql context
ConnectorConfig	getConfig()
boolean	isSqlSelectValid(String sql) Validate sql select
void	setConfig(ConnectorConfig config)
Object	sqlSelect(String sqlSelect, String temporaryTableName, Boolean streaming) SQL Select
String	toNativeQuery(org.mule.common.query.DsqlQuery query)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

SparkSqlConnector

public SparkSqlConnector()

Method Detail

addJdbcDataProperties

@Processor
public void addJdbcDataProperties(List<JdbcDataProperties> listJdbcDataProps)
                                       throws SQLException

Dynamically add new datasources from JDBC accesible databases

Parameters:

listJdbcDataProps - List of JDBC accesible databases options

Throws:

SparkSqlConnectorException - If new JDBC data source options are not valid

SQLException - error thrown during jdbc configuration

addFileDataProperties

@Processor
public void addFileDataProperties(List<FileDataProperties> listFileDataProps)

Dynamically add new datasources from Hadoop HDFS files

Parameters:

listFileDataProps - List of Hadoop HDFS files options

Throws:

SparkSqlConnectorException - If new file(s) options are not valid

addCassandraDataProperties

@Processor
public void addCassandraDataProperties(List<CassandraDataProperties> listCassandraDataProps)

Dynamically add new datasources from Cassandra database keyspaces and tables

Parameters:

listCassandraDataProps - List of Cassandra database keyspaces and tables

Throws:

SparkSqlConnectorException - If new schemas and/or tables options are not valid

customSql

@Processor
public Object customSql(String sql)

Run an arbitraty SQL query through Spark SQL Context.

Parameters:

sql - SQL Query

Returns:

List<Map<String, Object>> with query results or an empty connection if the query does not produce results.

Throws:

SparkSqlConnectorException - if there is any problem running the query (for example, query is not valid)

sqlSelect

@MetaDataScope(value=DataSenseResolver.class)
 @Processor
public Object sqlSelect(@Placement(order=1) @Query
                                                                                    String sqlSelect,
                                                                                    @Placement(order=2) @Optional
                                                                                    String temporaryTableName,
                                                                                    @Placement(order=3) @Optional
                                                                                    Boolean streaming)

SQL Select

Parameters:

sqlSelect - SQL select query to run into the Apache Spark Cluster through the Spark SQL Context.

temporaryTableName - If set, it will be the name of a new temporary table available at the Spark SQL Context with current query results.

streaming - if set (true) the result will be of type 'Iterator<Map<String, Object>>' and not a 'List<Map<String, Object>>'. This is the way to avoid loading large sizes of data into memory.

Returns:

the result will be the structured data found by the query.. If streaming argument is set (true) the result will be of type 'Iterator<Map<String, Object>>', else a 'List<Map<String, Object>>'.

Throws:

SqlSelectNotValidException - if the query is not a select query

SparkSqlConnectorException - if there is any problem running the query (for example, query is not valid)

isSqlSelectValid

public boolean isSqlSelectValid(String sql)

Validate sql select

Parameters:

sql - SQL select query

Returns:

boolean result of validating query

executeSql

public org.apache.spark.sql.DataFrame executeSql(String sql)

Execute sql in sql context

Parameters:

sql - SQL Query

Returns:

DataFrame result of the sql

Throws:

SparkSqlConnectorException - if there is any problem running the query (for example, query is not valid)

dropTemporaryTable

@Processor
public void dropTemporaryTable(String temporaryTableName)
                                    throws org.apache.spark.sql.catalyst.analysis.NoSuchTableException

Drop a temporary table present at the Spark SQL Context.

Parameters:

temporaryTableName - Temporary table name

Throws:

org.apache.spark.sql.catalyst.analysis.NoSuchTableException - if the table no exists at the Spark SQL Context.

EmptyTableNameException - if table name argument is null or empty

getConfig

public ConnectorConfig getConfig()

setConfig

public void setConfig(ConnectorConfig config)

disconnect

@Processor
public void disconnect()

Force global configuration disconnection. This way, the Spark Context / "driver" will be cleared and all related resources freed (Apache Spark Executors for example)

toNativeQuery

@QueryTranslator
public String toNativeQuery(org.mule.common.query.DsqlQuery query)