com.pervasive.datarush.operators.io.paraccel

Class MatrixSink

java.lang.Object

com.pervasive.datarush.operators.AbstractLogicalOperator

com.pervasive.datarush.operators.CompositeOperator

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

com.pervasive.datarush.operators.io.paraccel.MatrixSink

All Implemented Interfaces:

LogicalOperator, RecordSinkOperator, SinkOperator<RecordPort>

public class MatrixSink
extends MatrixOperatorBase<Void>

Load data from the input data stream into Actian Matrix. Graphs containing this operator cannot be run standalone. They must be serialized to JSON as a LogicalGraphWithConfig and deployed to a Matrix database. Execution is triggered by a Matrix SQL UDF invocation that provides all required configuration settings and posts the graph to a cluster for execution.

The target schema is defined as a reference to an existing database table through a named parameter to the Matrix SQL UDF invocation. The names and types of the input port fields must match those of the target schema. All input port fields must be part of the target schema. Target schema columns that have no corresponding input port field will be set to the default ('null') value for their database type.

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.

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).

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
MatrixSink()

Method Summary

Modifier and Type	Method and Description
protected MatrixOperatorBase.Workers<Void>	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	getEndPort() End of the IP port range to be used for the slice registration server socket.
String	getMatrixPort() The server port on the Matrix instance listening for messages from this operator.
String	getSliceInfo() A comma separated list of Matrix slice specs in the form : describing the worker node connection points on the Matrix side.
String	getStartPort() Start of the IP port range to be used for the slice registration server socket.
String	getTableSchema() A comma separated list of column specs in the form ::: describing the target schema on the Matrix side.
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	setEndPort(String endPort) End of the IP port range to be used for the slice registration server socket.
void	setMatrixPort(String matrixPort) Sets the server port on the Matrix instance listening for messages from this operator.
void	setSliceInfo(String sliceInfo) Takes a comma separated list of Matrix slice specs in the form : describing the worker node connection points on the Matrix side.
void	setStartPort(String startPort) Sets the start of the IP port range to be used for the slice registration server socket.
void	setTableSchema(String tableSchema) Takes a comma separated list of column specs in the form ::: describing the target schema on the Matrix side.
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

MatrixSink

public MatrixSink()

Method Detail

getMatrixPort

public String getMatrixPort()

The server port on the Matrix instance listening for messages from this operator.

Returns:

A port number

setMatrixPort

public void setMatrixPort(String matrixPort)

Sets the server port on the Matrix instance listening for messages from this operator. This attribute is not intended to be configured by the user, its value is injected by the Matrix UDF.

getTableSchema

public String getTableSchema()

A comma separated list of column specs in the form ::: describing the target schema on the Matrix side.

Returns:

A string describing the target schema

setTableSchema

public void setTableSchema(String tableSchema)

Takes a comma separated list of column specs in the form ::: describing the target schema on the Matrix side. This attribute is not intended to be configured by the user, its value is injected by the Matrix UDF.

getSliceInfo

public String getSliceInfo()

A comma separated list of Matrix slice specs in the form : describing the worker node connection points on the Matrix side.

Returns:

A string describing the worker node slices on the Matrix side

setSliceInfo

public void setSliceInfo(String sliceInfo)

Takes a comma separated list of Matrix slice specs in the form : describing the worker node connection points on the Matrix side. This attribute is not intended to be configured by the user, its value is injected by the Matrix UDF.

getStartPort

public String getStartPort()

Start of the IP port range to be used for the slice registration server socket. String-valued for implementation purposes (JSON injection.)

Returns:

The start of the port range as a string

setStartPort

public void setStartPort(String startPort)

Sets the start of the IP port range to be used for the slice registration server socket. String-valued for implementation purposes (JSON injection.) This attribute is not intended to be configured by the user, its value is injected by the Matrix UDF.

getEndPort

public String getEndPort()

End of the IP port range to be used for the slice registration server socket. String-valued for implementation purposes (JSON injection.)

Returns:

The end of the port range as a string

setEndPort

public void setEndPort(String endPort)

End of the IP port range to be used for the slice registration server socket. String-valued for implementation purposes (JSON injection.) This attribute is not intended to be configured by the user, its value is injected by the Matrix UDF.

validateExtensions

protected void validateExtensions()

Specified by:

validateExtensions in class MatrixOperatorBase<Void>

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<Void>

createWorkers

protected MatrixOperatorBase.Workers<Void> 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<Void>

renameMapping

protected Map<String,String> renameMapping()

Specified by:

renameMapping in class MatrixOperatorBase<Void>

skipCompose

protected boolean skipCompose()

Specified by:

skipCompose in class MatrixOperatorBase<Void>

portRange

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

Specified by:

portRange in class MatrixOperatorBase<Void>