com.pervasive.datarush.operators.io.paraccel

Class LoadMatrix

java.lang.Object

com.pervasive.datarush.operators.AbstractLogicalOperator

com.pervasive.datarush.operators.CompositeOperator

com.pervasive.datarush.operators.io.paraccel.MatrixOperatorBase<com.pervasive.datarush.operators.io.paraccel.ODIInvoker>

com.pervasive.datarush.operators.io.paraccel.LoadMatrix

All Implemented Interfaces:

LogicalOperator, RecordSinkOperator, SinkOperator<RecordPort>

public class LoadMatrix
extends MatrixOperatorBase<com.pervasive.datarush.operators.io.paraccel.ODIInvoker>

Load data from the input data stream into an Actian Matrix database table. The target table is specified using the setTable(String) method. A mapping of input data source fields into target table columns can be provided. Any unmapped input fields are ignored. Unmapped table columns are not loaded and must be nullable. If field mapping is not provided, input fields are mapped to output columns by position.

The Actian Matrix database supports the concept of slices. A slice represents a unit of parallelism within the database. Each node within a database instance normally supports multiple slices. The loader operator communicates directly with slices to transfer data into the target database. The number of connections made into the database is determined by the parallelism level of the loader operator. The input data is not repartitioned to match the number of target slices. If the parallelism (partition count) of the loader operator is less than the number of database slices, some slices will be left idle during the load process. To repartition the data to match the number of slices in the target database, explicitly set the parallelism level of the loader operator to match the number of database slices.

A component of the data loader executes within the database. This in-database component supports tracing for debugging. The tracing level can be set to values 0 to 9. By default, the trace level is set to zero. To increase the amount of tracing produced in the database by the loader, set the trace level to a positive, non-zero value. The setTraceLevel(int) method is used to set the trace level.

Network communication between this Dataflow operator and the Matrix database is accomplished using TCP/IP sockets. The properties startPortRange and endPortRange can be used to limit the range of TCP/IP ports utilized for this communication. The ports within this range will need to be available for two way communication. Within an environment containing a firewall, the ports will need to be "opened" within the firewall configuration. The default port range is from 40000 to 41000 (exclusive).

A network connection for data transfer is created for each parallel data worker. To limit the number of network connections used, limit the parallelism of the LoadMatrix operator. A limit on connections can be explicitly set on the operator using the maxConnections operator property. Reference the <@link MatrixOperatorBase.setMaxConnections(int)> property setter.

Note that modifying the parallelism level of the LoadMatrix operator either via the operator setting or the setMaxConnections method may incur the overhead of data redistribution to match the changed parallelism level. To prevent redistribution, set the overall parallelism of the DataFlow job to limit the number of data connections used into Matrix.

Nested Class Summary

Nested classes/interfaces inherited from class com.pervasive.datarush.operators.io.paraccel.MatrixOperatorBase

MatrixOperatorBase.Pair<A,B>, MatrixOperatorBase.Workers<E>

Field Summary

Fields inherited from class com.pervasive.datarush.operators.io.paraccel.MatrixOperatorBase

END_PORT_RANGE, host, logFrequency, logger, maxConnections, retryCount, START_PORT_RANGE, timeoutInterval

Constructor Summary

Constructors

Constructor and Description
LoadMatrix()

Method Summary

Modifier and Type	Method and Description
protected MatrixOperatorBase.Workers<com.pervasive.datarush.operators.io.paraccel.ODIInvoker>	createWorkers(CompositionContext ctx, RecordTokenType inputType, com.pervasive.datarush.operators.io.paraccel.TableInfo tableInfo, List<String> columns, Map<String,String> columnMap, int startPortRange, int endPortRange, int dataWorkerParallelism)
String	getDatabase() Get the database name.
int	getEndPortRange() Get the end of the IP port range.
String	getFinalizeTableSQL() Retrieves the SQL statement to execute after processing all the records.
String	getInitializeTableSQL() Retrieves the SQL statement to execute before processing any records.
int	getLogFrequency() Get the log frequency property.
String	getPassword() Get the password.
int	getPort() Get the Matrix lead node port number.
Map<String,String>	getRenameMapping() Get the source to target field name mapping.
int	getRetryCount() Get the retry count property.
int	getStartPortRange() Get the start of the IP port range.
String	getTable() Get the target table name.
int	getTimeoutInterval() Get the communication timeout interval (in seconds).
int	getTraceLevel() Get the current trace level.
String	getUser() Get the user account name.
protected MatrixOperatorBase.Pair<Integer,Integer>	portRange()
protected Map<String,String>	renameMapping()
protected MatrixOperatorBase.Pair<com.pervasive.datarush.operators.io.paraccel.TableInfo,List<com.pervasive.datarush.operators.io.paraccel.SliceIdentifier>>	schemaAndSliceInfo()
void	setDatabase(String databaseName) Set the database name.
void	setEndPortRange(int endPortRange) Set the end of the IP port range to use for network connections to and from the target Matrix database.
void	setFinalizeTableSQL(String finalizeTableSQL) Sets the SQL statement to execute after processing all the records.
void	setInitializeTableSQL(String initializeTableSQL) Sets the SQL statement to execute before processing any records.
void	setLogFrequency(int logFrequency) Set the frequency in units of rows of data that progress is reported.
void	setPassword(String password) Set the user's password.
void	setPort(int port) Set the Matrix leader node port number.
void	setRenameMapping(Map<String,String> renameMapping) Set the source to target field name mapping.
void	setRetryCount(int retryCount) Set the number of times communication requests to the MPP database system are retried before being considered an error.
void	setStartPortRange(int startPortRange) Set the start of the IP port range to use for network connections to and from the Matrix database.
void	setTable(String tableName) Set the name of the table to load.
void	setTimeoutInterval(int timeoutInterval) Set the timeout interval (in seconds) for communication requests the target MPP database system.
void	setTraceLevel(int traceLevel) Set the trace level used by the query UDF function invoked to load the data in the target database.
void	setUser(String userName) Set the user name.
protected boolean	skipCompose()
protected void	validateExtensions()

Methods inherited from class com.pervasive.datarush.operators.io.paraccel.MatrixOperatorBase

compose, getHost, getInput, getMaxConnections, setHost, setMaxConnections, validatePortRange

Methods inherited from class com.pervasive.datarush.operators.AbstractLogicalOperator

disableParallelism, getInputPorts, getOutputPorts, newInput, newInput, newOutput, newRecordInput, newRecordInput, newRecordOutput, notifyError

Methods inherited from class java.lang.Object

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

Methods inherited from interface com.pervasive.datarush.operators.LogicalOperator

disableParallelism, getInputPorts, getOutputPorts

Constructor Detail

LoadMatrix

public LoadMatrix()

Method Detail

getDatabase

public String getDatabase()

Get the database name.

Returns:

database name

setDatabase

public void setDatabase(String databaseName)

Set the database name. Specify the database where the target table lives.

Parameters:

databaseName - database name

getInitializeTableSQL

public String getInitializeTableSQL()

Retrieves the SQL statement to execute before processing any records.

Returns:

the SQL statement to execute before processing any records

setInitializeTableSQL

public void setInitializeTableSQL(String initializeTableSQL)

Sets the SQL statement to execute before processing any records. If the table does not exist this value is required, and should contain a CREATE TABLE statement to create the table.

This statement is executed only once, regardless of #partitionCount.

Parameters:

initializeTableSQL - the SQL statement to execute before processing any records

getFinalizeTableSQL

public String getFinalizeTableSQL()

Retrieves the SQL statement to execute after processing all the records.

Returns:

the SQL statement to execute after processing all the records

setFinalizeTableSQL

public void setFinalizeTableSQL(String finalizeTableSQL)

Sets the SQL statement to execute after processing all the records. One example use is a CREATE INDEX statement.

This statement is executed only once, regardless of #partitionCount.

Parameters:

finalizeTableSQL - the SQL statement to execute after processing all the records

getUser

public String getUser()

Get the user account name.

Returns:

user account name

setUser

public void setUser(String userName)

Set the user name.

Parameters:

userName - user account name

getPassword

public String getPassword()

Get the password.

Returns:

password

setPassword

public void setPassword(String password)

Set the user's password.

Parameters:

password -

getPort

public int getPort()

Get the Matrix lead node port number.

Returns:

port number

setPort

public void setPort(int port)

Set the Matrix leader node port number.

Parameters:

port - port number

getStartPortRange

public int getStartPortRange()

Get the start of the IP port range.

Returns:

staring IP port within the range of ports to use

setStartPortRange

public void setStartPortRange(int startPortRange)

Set the start of the IP port range to use for network connections to and from the Matrix database. The port range has to cover a valid range of defined IP ports.

Parameters:

startPortRange - start point of the IP port range

getEndPortRange

public int getEndPortRange()

Get the end of the IP port range.

Returns:

setEndPortRange

public void setEndPortRange(int endPortRange)

Set the end of the IP port range to use for network connections to and from the target Matrix database. The port range has to cover a valid range of defined IP ports.

Parameters:

endPortRange - end point of IP port range

getLogFrequency

public int getLogFrequency()

Get the log frequency property.

Returns:

log frequency

setLogFrequency

public void setLogFrequency(int logFrequency)

Set the frequency in units of rows of data that progress is reported. The default is zero, which implies no logging. Every time the row count is hit a log is output giving the current total of rows written to the database. The frequency is per parallel stream of processing.

Parameters:

logFrequency - the count of rows use as the frequency to log progress

getRetryCount

public int getRetryCount()

Get the retry count property.

Returns:

number of times to retry communication requests

setRetryCount

public void setRetryCount(int retryCount)

Set the number of times communication requests to the MPP database system are retried before being considered an error. The retry count may need to be increased for large MPP systems with many nodes.

Parameters:

retryCount - number of times to retry communication requests

getTimeoutInterval

public int getTimeoutInterval()

Get the communication timeout interval (in seconds).

Returns:

setTimeoutInterval

public void setTimeoutInterval(int timeoutInterval)

Set the timeout interval (in seconds) for communication requests the target MPP database system.

Parameters:

timeoutInterval - timeout interval in seconds

validateExtensions

protected void validateExtensions()

Specified by:

validateExtensions in class MatrixOperatorBase<com.pervasive.datarush.operators.io.paraccel.ODIInvoker>

schemaAndSliceInfo

protected MatrixOperatorBase.Pair<com.pervasive.datarush.operators.io.paraccel.TableInfo,List<com.pervasive.datarush.operators.io.paraccel.SliceIdentifier>> schemaAndSliceInfo()

Specified by:

schemaAndSliceInfo in class MatrixOperatorBase<com.pervasive.datarush.operators.io.paraccel.ODIInvoker>

createWorkers

protected MatrixOperatorBase.Workers<com.pervasive.datarush.operators.io.paraccel.ODIInvoker> createWorkers(CompositionContext ctx,
                                                                                                            RecordTokenType inputType,
                                                                                                            com.pervasive.datarush.operators.io.paraccel.TableInfo tableInfo,
                                                                                                            List<String> columns,
                                                                                                            Map<String,String> columnMap,
                                                                                                            int startPortRange,
                                                                                                            int endPortRange,
                                                                                                            int dataWorkerParallelism)

Specified by:

createWorkers in class MatrixOperatorBase<com.pervasive.datarush.operators.io.paraccel.ODIInvoker>

renameMapping

protected Map<String,String> renameMapping()

Specified by:

renameMapping in class MatrixOperatorBase<com.pervasive.datarush.operators.io.paraccel.ODIInvoker>

getRenameMapping

public Map<String,String> getRenameMapping()

Get the source to target field name mapping.

Returns:

field name mapping

setRenameMapping

public void setRenameMapping(Map<String,String> renameMapping)

Set the source to target field name mapping. This maps from a name in the input data flow to a name in the target database table. If the names are the same, the mapping is not required.

This is an optional property. If not provided, the input fields are mapped to the target database table by schema order.

Parameters:

renameMapping - source to target field name mapping

getTable

public String getTable()

Get the target table name.

Returns:

target table name

setTable

public void setTable(String tableName)

Set the name of the table to load.

Parameters:

tableName - target table name.

skipCompose

protected boolean skipCompose()

Specified by:

skipCompose in class MatrixOperatorBase<com.pervasive.datarush.operators.io.paraccel.ODIInvoker>

getTraceLevel

public int getTraceLevel()

Get the current trace level.

Returns:

trace level

setTraceLevel

public void setTraceLevel(int traceLevel)

Set the trace level used by the query UDF function invoked to load the data in the target database. The default level is 0 which implies on tracing. Levels greater than zero will enable more tracing.

The trace logs can be viewed in the database by querying the stl_udf_trace system table.

Parameters:

traceLevel - trace level (from 0 - 9)

portRange

protected MatrixOperatorBase.Pair<Integer,Integer> portRange()

Specified by:

portRange in class MatrixOperatorBase<com.pervasive.datarush.operators.io.paraccel.ODIInvoker>