org.mule.modules.sparksql.config

Class ConnectorConfig

java.lang.Object

org.mule.modules.sparksql.config.ConnectorConfig

@ConnectionManagement(configElementName="config",
                      friendlyName="Configuration")
public class ConnectorConfig
extends Object

Spark SQL Configuration contains the required information to connect to a Apache Spark Cluster. The global configuration will hold what is usually called the Apache Spark 'Driver' or 'Spark Context', the process that sends tasks to executors and listen to them during their lifetime.
See (see Apache Spark configuration) and (see Apache Spark security)

XML Sample

    <spark-sql:config name="SparkSql__Configuration" appName="sparksql-demo" master="spark://192.168.1.1:7077" 
         cassandraIp="192.168.1.2" cassandraPort="9042" 
         addCassandraJar="true">
        <spark-sql:advanced-spark-conf>
            <spark-sql:advanced-spark-conf key="spark.executor.cores">1</spark-sql:advanced-spark-conf>
            <spark-sql:advanced-spark-conf key="spark.cores.max">2</spark-sql:advanced-spark-conf>
        </spark-sql:advanced-spark-conf>
        <spark-sql:list-file-data-props>
            <spark-sql:list-file-data-prop type="CSV" alias="csv_file" uri="hdfs://192.168.1.3:5001/demo/file1.csv" addJar="true" />
            <spark-sql:list-file-data-prop type="JSON" alias="json_fime" uri="hdfs://192.168.1.3:5001/demo/file2.json" addJar="true" />
        </spark-sql:list-file-data-props>
        <spark-sql:list-jdbc-data-props>
            <spark-sql:list-jdbc-data-prop driver="org.apache.derby.jdbc.ClientDriver" dbTable="" 
                url="jdbc:derby://192.168.1.4:19000/MyDbTest" addJar="true"/>
        </spark-sql:list-jdbc-data-props>
        <spark-sql:list-cassandra-data-props>
            <spark-sql:list-cassandra-data-prop keyspace="demo" tableName=""/>
        </spark-sql:list-cassandra-data-props>
    </spark-sql:config>
 

Author:

mnemo

Field Summary

Fields

Modifier and Type	Field and Description
static String	SEPARATOR

Constructor Summary

Constructors

Constructor and Description
ConnectorConfig()

Method Summary

Modifier and Type	Method and Description
void	addCsvAndAvroJars(org.apache.spark.SparkContext sc, FileDataProperties fileDataProperties)
protected void	addDrivers()
void	addJarDrivers(org.apache.spark.SparkContext sc, JdbcDataProperties jdbcDataProperties)
void	connect(String connectionKey, String master, String appName) Connect to Apache Spark Cluster
protected void	connectDataSources() Connect all defined data sources to the sql context
void	connectDataSources(CassandraDataProperties cassandraDataProperties) Connect all defined data sources to the sql context
void	connectDataSources(FileDataProperties fileDataProperties) Connect all defined data sources to the sql context
void	connectDataSources(JdbcDataProperties jdbcDataProperties) Connect all defined data sources to the sql context
String	connectionId() login ID
protected boolean	dataSourceAvro(FileDataProperties fileDataProperties)
protected boolean	dataSourceCsv(FileDataProperties fileDataProperties)
protected void	destroyRunner()
void	disconnect() Disconnect
protected boolean	empty(String field)
UrlDelegatingCl	ensureRunner()
Boolean	getAddCassandraJar()
Map<String,String>	getAdvancedSparkConf()
String	getCassandraIp()
String	getCassandraPort()
String	getHeartbeatInterval()
List<CassandraDataProperties>	getListCassandraDataProps()
List<FileDataProperties>	getListFileDataProps()
List<JdbcDataProperties>	getListJdbcDataProps()
String	getLocalIp()
String	getLocalPort()
org.apache.spark.SparkContext	getSparkContext()
org.apache.spark.sql.SQLContext	getSparkSqlContext()
SparkSqlEnum.SqlContextType	getSqlContextType()
protected void	innerConnect(String key, String master, String appName)
boolean	isConnected() Test if Spark Context (driver) exists and is valid (connected if non local)
void	setAddCassandraJar(Boolean addCassandraJar)
void	setAdvancedSparkConf(Map<String,String> advancedSparkConf)
void	setCassandraIp(String cassandraIp)
void	setCassandraPort(String cassandraPort)
void	setHeartbeatInterval(String heartbeatInterval)
void	setListCassandraDataProps(List<CassandraDataProperties> listCassandraDataProps)
void	setListFileDataProps(List<FileDataProperties> listFileDataProps)
void	setListJdbcDataProps(List<JdbcDataProperties> listJdbcDataProps)
void	setLocalIp(String localIp)
void	setLocalPort(String localPort)
void	setSqlContextType(SparkSqlEnum.SqlContextType sqlContext)
void	testConnect(String key, String master, String appName) Test Connect

Methods inherited from class java.lang.Object

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

Field Detail

SEPARATOR

public static final String SEPARATOR

See Also:

Constant Field Values

Constructor Detail

ConnectorConfig

public ConnectorConfig()

Method Detail

getAdvancedSparkConf

public Map<String,String> getAdvancedSparkConf()

setAdvancedSparkConf

public void setAdvancedSparkConf(Map<String,String> advancedSparkConf)

getLocalIp

public String getLocalIp()

setLocalIp

public void setLocalIp(String localIp)

getLocalPort

public String getLocalPort()

setLocalPort

public void setLocalPort(String localPort)

getHeartbeatInterval

public String getHeartbeatInterval()

setHeartbeatInterval

public void setHeartbeatInterval(String heartbeatInterval)

getSqlContextType

public SparkSqlEnum.SqlContextType getSqlContextType()

setSqlContextType

public void setSqlContextType(SparkSqlEnum.SqlContextType sqlContext)

getListFileDataProps

public List<FileDataProperties> getListFileDataProps()

setListFileDataProps

public void setListFileDataProps(List<FileDataProperties> listFileDataProps)

getListJdbcDataProps

public List<JdbcDataProperties> getListJdbcDataProps()

setListJdbcDataProps

public void setListJdbcDataProps(List<JdbcDataProperties> listJdbcDataProps)

getCassandraIp

public String getCassandraIp()

setCassandraIp

public void setCassandraIp(String cassandraIp)

getCassandraPort

public String getCassandraPort()

getAddCassandraJar

public Boolean getAddCassandraJar()

setAddCassandraJar

public void setAddCassandraJar(Boolean addCassandraJar)

setCassandraPort

public void setCassandraPort(String cassandraPort)

getListCassandraDataProps

public List<CassandraDataProperties> getListCassandraDataProps()

setListCassandraDataProps

public void setListCassandraDataProps(List<CassandraDataProperties> listCassandraDataProps)

getSparkSqlContext

public org.apache.spark.sql.SQLContext getSparkSqlContext()

getSparkContext

public org.apache.spark.SparkContext getSparkContext()

ensureRunner

public UrlDelegatingCl ensureRunner()

destroyRunner

protected void destroyRunner()

connect

@Connect
public void connect(@Optional @Password @ConnectionKey
                              String connectionKey,
                              String master,
                              String appName)
                       throws org.mule.api.ConnectionException

Connect to Apache Spark Cluster

Parameters:

connectionKey - Shared secret currently supported by Apache Spark for authentication (Apache Spark 'spark.authenticate.secret' option, 'spark.authenticate' will be set to 'true' if not empty or null.).

master - Spark Master URL (Apache Spark 'spark.master' option). Next values are currently supported (see Apache Spark documentation):

• local - Run Spark locally with one worker thread (i.e. no parallelism at all).
• local[K] - Run Spark locally with K worker threads (ideally, set this to the number of cores on your machine).
• local[*] - Run Spark locally with as many worker threads as logical cores on your machine.
• spark://HOST:PORT - Connect to the given Spark standalone cluster master. The port must be whichever one your master is configured to use, which is 7077 by default.

appName - The name of your application (Apache Spark 'spark.app.name' option). This will appear in the UI and in log data.

Throws:

org.mule.api.ConnectionException - if the connection does not occur

testConnect

@TestConnectivity
public void testConnect(String key,
                                           String master,
                                           String appName)
                                    throws org.mule.api.ConnectionException

Test Connect

Parameters:

key - Key to connect to server (if necessary)

master - Spark connection

appName - App Name

Throws:

org.mule.api.ConnectionException - if the connection fails

innerConnect

protected void innerConnect(String key,
                            String master,
                            String appName)
                     throws SQLException,
                            org.mule.api.ConnectionException

Throws:

SQLException

org.mule.api.ConnectionException

addDrivers

protected void addDrivers()

empty

protected boolean empty(String field)

addCsvAndAvroJars

public void addCsvAndAvroJars(org.apache.spark.SparkContext sc,
                              FileDataProperties fileDataProperties)

dataSourceCsv

protected boolean dataSourceCsv(FileDataProperties fileDataProperties)

dataSourceAvro

protected boolean dataSourceAvro(FileDataProperties fileDataProperties)

addJarDrivers

public void addJarDrivers(org.apache.spark.SparkContext sc,
                          JdbcDataProperties jdbcDataProperties)

connectDataSources

protected void connectDataSources()
                           throws SQLException

Connect all defined data sources to the sql context

Throws:

SQLException - An exception that provides information on a database access error or other errors.

connectDataSources

public void connectDataSources(FileDataProperties fileDataProperties)

Connect all defined data sources to the sql context

Parameters:

fileDataProperties - Hadoop HDFS file options

Throws:

SparkSqlConnectorException - If file options are not valid

connectDataSources

public void connectDataSources(JdbcDataProperties jdbcDataProperties)
                        throws SQLException

Connect all defined data sources to the sql context

Parameters:

jdbcDataProperties - JDBC accesible database options

Throws:

SparkSqlConnectorException - If new JDBC data source options are not valid

SQLException - error thrown during jdbc configuration

connectDataSources

public void connectDataSources(CassandraDataProperties cassandraDataProperties)

Connect all defined data sources to the sql context

Parameters:

cassandraDataProperties - Cassandra database keyspace and table

Throws:

SparkSqlConnectorException - If schemas or table options are not valid

disconnect

@Disconnect
public void disconnect()

Disconnect

isConnected

@ValidateConnection
public boolean isConnected()

Test if Spark Context (driver) exists and is valid (connected if non local)

Returns:

true if connected

connectionId

@ConnectionIdentifier
public String connectionId()

login ID

Returns:

String login ID